diff --git a/.secrets.baseline b/.secrets.baseline index 9e664126..a179896d 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -3,7 +3,7 @@ "files": "^.secrets.baseline$", "lines": null }, - "generated_at": "2022-11-03T20:16:27Z", + "generated_at": "2022-11-04T17:15:32Z", "plugins_used": [ { "name": "AWSKeyDetector" diff --git a/examples/test_user_management_v1_examples.py b/examples/test_user_management_v1_examples.py index f84cf40e..aeb12fd0 100644 --- a/examples/test_user_management_v1_examples.py +++ b/examples/test_user_management_v1_examples.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -# (C) Copyright IBM Corp. 2021. +# (C) Copyright IBM Corp. 2021, 2022. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -175,23 +175,23 @@ def test_list_users_example(self): """ list_users request example """ - assert account_id is not None - try: - print('\nlist_users() result:') # begin-list_users - user_list = user_management_service.list_users( + all_results = [] + pager = UsersPager( + client=user_management_service, account_id=account_id, - state='ACTIVE', - limit=100, - ).get_result() + ) + while pager.has_next(): + next_page = pager.get_next() + assert next_page is not None + all_results.extend(next_page) - print(json.dumps(user_list, indent=2)) + print(json.dumps(all_results, indent=2)) # end-list_users - except ApiException as e: pytest.fail(str(e)) diff --git a/ibm_platform_services/user_management_v1.py b/ibm_platform_services/user_management_v1.py index deee9b3b..3318fe81 100644 --- a/ibm_platform_services/user_management_v1.py +++ b/ibm_platform_services/user_management_v1.py @@ -1,6 +1,6 @@ # coding: utf-8 -# (C) Copyright IBM Corp. 2020. +# (C) Copyright IBM Corp. 2022. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -14,16 +14,18 @@ # See the License for the specific language governing permissions and # limitations under the License. -# IBM OpenAPI SDK Code Generator Version: 99-SNAPSHOT-d753183b-20201209-163011 +# IBM OpenAPI SDK Code Generator Version: 3.60.2-95dc7721-20221102-203229 """ Manage the lifecycle of your users using User Management APIs. + +API Version: 1.0 """ from typing import Dict, List import json -from ibm_cloud_sdk_core import BaseService, DetailedResponse +from ibm_cloud_sdk_core import BaseService, DetailedResponse, get_query_param from ibm_cloud_sdk_core.authenticators.authenticator import Authenticator from ibm_cloud_sdk_core.get_authenticator import get_authenticator_from_environment from ibm_cloud_sdk_core.utils import convert_model @@ -62,7 +64,7 @@ def __init__(self, Construct a new client for the User Management service. :param Authenticator authenticator: The authenticator specifies the authentication mechanism. - Get up to date information from https://github.com/IBM/python-sdk-core/blob/master/README.md + Get up to date information from https://github.com/IBM/python-sdk-core/blob/main/README.md about initializing the authenticator of your choice. """ BaseService.__init__(self, @@ -78,9 +80,9 @@ def __init__(self, def list_users(self, account_id: str, *, - state: str = None, limit: int = None, start: str = None, + user_id: str = None, **kwargs ) -> DetailedResponse: """ @@ -97,19 +99,19 @@ def list_users(self, paginated list with a default limit of 100 users. You can iterate through all users by following the `next_url` field. - :param str account_id: The account ID. - :param str state: (optional) The state of the user. + :param str account_id: The account ID of the specified user. :param int limit: (optional) The number of results to be returned. :param str start: (optional) An optional token that indicates the beginning of the page of results to be returned. If omitted, the first page of results is returned. This value is obtained from the 'next_url' field of the operation response. + :param str user_id: (optional) Filter users based on their user ID. :param dict headers: A `dict` containing the request headers :return: A `DetailedResponse` containing the result, headers and HTTP status code. :rtype: DetailedResponse with `dict` result representing a `UserList` object """ - if account_id is None: + if not account_id: raise ValueError('account_id must be provided') headers = {} sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, @@ -118,13 +120,14 @@ def list_users(self, headers.update(sdk_headers) params = { - 'state': state, 'limit': limit, - '_start': start + '_start': start, + 'user_id': user_id } if 'headers' in kwargs: headers.update(kwargs.get('headers')) + del kwargs['headers'] headers['Accept'] = 'application/json' path_param_keys = ['account_id'] @@ -136,7 +139,7 @@ def list_users(self, headers=headers, params=params) - response = self.send(request) + response = self.send(request, **kwargs) return response @@ -165,7 +168,7 @@ def invite_users(self, the user is transitioned to `ACTIVE` state. If the user email is already verified, no email is generated. - :param str account_id: The account ID. + :param str account_id: The account ID of the specified user. :param List[InviteUser] users: (optional) A list of users to be invited. :param List[InviteUserIamPolicy] iam_policy: (optional) A list of IAM policies. @@ -175,7 +178,7 @@ def invite_users(self, :rtype: DetailedResponse with `dict` result representing a `InvitedUserList` object """ - if account_id is None: + if not account_id: raise ValueError('account_id must be provided') if users is not None: users = [convert_model(x) for x in users] @@ -198,6 +201,7 @@ def invite_users(self, if 'headers' in kwargs: headers.update(kwargs.get('headers')) + del kwargs['headers'] headers['Accept'] = 'application/json' path_param_keys = ['account_id'] @@ -209,13 +213,15 @@ def invite_users(self, headers=headers, data=data) - response = self.send(request) + response = self.send(request, **kwargs) return response def get_user_profile(self, account_id: str, iam_id: str, + *, + include_activity: str = None, **kwargs ) -> DetailedResponse: """ @@ -226,16 +232,18 @@ def get_user_profile(self, requesting user or service ID must have at least the viewer, editor, or administrator role on the User Management service. - :param str account_id: The account ID. + :param str account_id: The account ID of the specified user. :param str iam_id: The user's IAM ID. + :param str include_activity: (optional) Include activity information of the + user, such as the last authentication timestamp. :param dict headers: A `dict` containing the request headers :return: A `DetailedResponse` containing the result, headers and HTTP status code. :rtype: DetailedResponse with `dict` result representing a `UserProfile` object """ - if account_id is None: + if not account_id: raise ValueError('account_id must be provided') - if iam_id is None: + if not iam_id: raise ValueError('iam_id must be provided') headers = {} sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, @@ -243,8 +251,13 @@ def get_user_profile(self, operation_id='get_user_profile') headers.update(sdk_headers) + params = { + 'include_activity': include_activity + } + if 'headers' in kwargs: headers.update(kwargs.get('headers')) + del kwargs['headers'] headers['Accept'] = 'application/json' path_param_keys = ['account_id', 'iam_id'] @@ -253,9 +266,10 @@ def get_user_profile(self, url = '/v2/accounts/{account_id}/users/{iam_id}'.format(**path_param_dict) request = self.prepare_request(method='GET', url=url, - headers=headers) + headers=headers, + params=params) - response = self.send(request) + response = self.send(request, **kwargs) return response @@ -270,6 +284,7 @@ def update_user_profile(self, phonenumber: str = None, altphonenumber: str = None, photo: str = None, + include_activity: str = None, **kwargs ) -> DetailedResponse: """ @@ -284,7 +299,7 @@ def update_user_profile(self, states. For other request body fields, a user can update their own profile without having User Management service permissions. - :param str account_id: The account ID. + :param str account_id: The account ID of the specified user. :param str iam_id: The user's IAM ID. :param str firstname: (optional) The first name of the user. :param str lastname: (optional) The last name of the user. @@ -296,14 +311,16 @@ def update_user_profile(self, :param str altphonenumber: (optional) The alternative phone number of the user. :param str photo: (optional) A link to a photo of the user. + :param str include_activity: (optional) Include activity information of the + user, such as the last authentication timestamp. :param dict headers: A `dict` containing the request headers :return: A `DetailedResponse` containing the result, headers and HTTP status code. :rtype: DetailedResponse """ - if account_id is None: + if not account_id: raise ValueError('account_id must be provided') - if iam_id is None: + if not iam_id: raise ValueError('iam_id must be provided') headers = {} sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, @@ -311,6 +328,10 @@ def update_user_profile(self, operation_id='update_user_profile') headers.update(sdk_headers) + params = { + 'include_activity': include_activity + } + data = { 'firstname': firstname, 'lastname': lastname, @@ -326,6 +347,7 @@ def update_user_profile(self, if 'headers' in kwargs: headers.update(kwargs.get('headers')) + del kwargs['headers'] path_param_keys = ['account_id', 'iam_id'] path_param_values = self.encode_path_vars(account_id, iam_id) @@ -334,15 +356,18 @@ def update_user_profile(self, request = self.prepare_request(method='PATCH', url=url, headers=headers, + params=params, data=data) - response = self.send(request) + response = self.send(request, **kwargs) return response def remove_user(self, account_id: str, iam_id: str, + *, + include_activity: str = None, **kwargs ) -> DetailedResponse: """ @@ -354,16 +379,18 @@ def remove_user(self, Management service. For more information, see the [Removing users](https://cloud.ibm.com/docs/account?topic=account-remove) documentation. - :param str account_id: The account ID. + :param str account_id: The account ID of the specified user. :param str iam_id: The user's IAM ID. + :param str include_activity: (optional) Include activity information of the + user, such as the last authentication timestamp. :param dict headers: A `dict` containing the request headers :return: A `DetailedResponse` containing the result, headers and HTTP status code. :rtype: DetailedResponse """ - if account_id is None: + if not account_id: raise ValueError('account_id must be provided') - if iam_id is None: + if not iam_id: raise ValueError('iam_id must be provided') headers = {} sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, @@ -371,18 +398,118 @@ def remove_user(self, operation_id='remove_user') headers.update(sdk_headers) + params = { + 'include_activity': include_activity + } + if 'headers' in kwargs: headers.update(kwargs.get('headers')) + del kwargs['headers'] path_param_keys = ['account_id', 'iam_id'] path_param_values = self.encode_path_vars(account_id, iam_id) path_param_dict = dict(zip(path_param_keys, path_param_values)) url = '/v2/accounts/{account_id}/users/{iam_id}'.format(**path_param_dict) + request = self.prepare_request(method='DELETE', + url=url, + headers=headers, + params=params) + + response = self.send(request, **kwargs) + return response + + + def accept(self, + *, + account_id: str = None, + **kwargs + ) -> DetailedResponse: + """ + Accept an invitation. + + Accept a user invitation to an account. You can use the user's token for + authorization. To use this method, the requesting user must provide the account ID + for the account that they are accepting an invitation for. If the user already + accepted the invitation request, it returns 204 with no response body. + + :param str account_id: (optional) The account ID. + :param dict headers: A `dict` containing the request headers + :return: A `DetailedResponse` containing the result, headers and HTTP status code. + :rtype: DetailedResponse + """ + + headers = {} + sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, + service_version='V1', + operation_id='accept') + headers.update(sdk_headers) + + data = { + 'account_id': account_id + } + data = {k: v for (k, v) in data.items() if v is not None} + data = json.dumps(data) + headers['content-type'] = 'application/json' + + if 'headers' in kwargs: + headers.update(kwargs.get('headers')) + del kwargs['headers'] + + url = '/v2/users/accept' + request = self.prepare_request(method='POST', + url=url, + headers=headers, + data=data) + + response = self.send(request, **kwargs) + return response + + + def v3_remove_user(self, + account_id: str, + iam_id: str, + **kwargs + ) -> DetailedResponse: + """ + Remove user from account (Asynchronous). + + Remove users from an account by using the user's IAM ID. You must use a user token + for authorization. Service IDs can't remove users from an account. If removing the + user fails it will set the user's state to ERROR_WHILE_DELETING. To use this + method, the requesting user must have the editor or administrator role on the User + Management service. For more information, see the [Removing + users](https://cloud.ibm.com/docs/account?topic=account-remove) documentation. + + :param str account_id: The account ID of the specified user. + :param str iam_id: The user's IAM ID. + :param dict headers: A `dict` containing the request headers + :return: A `DetailedResponse` containing the result, headers and HTTP status code. + :rtype: DetailedResponse + """ + + if not account_id: + raise ValueError('account_id must be provided') + if not iam_id: + raise ValueError('iam_id must be provided') + headers = {} + sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, + service_version='V1', + operation_id='v3_remove_user') + headers.update(sdk_headers) + + if 'headers' in kwargs: + headers.update(kwargs.get('headers')) + del kwargs['headers'] + + path_param_keys = ['account_id', 'iam_id'] + path_param_values = self.encode_path_vars(account_id, iam_id) + path_param_dict = dict(zip(path_param_keys, path_param_values)) + url = '/v3/accounts/{account_id}/users/{iam_id}'.format(**path_param_dict) request = self.prepare_request(method='DELETE', url=url, headers=headers) - response = self.send(request) + response = self.send(request, **kwargs) return response ######################### @@ -411,16 +538,16 @@ def get_user_settings(self, the `self_manage` field, review information about the [user-managed login setting](https://cloud.ibm.com/docs/account?topic=account-types). - :param str account_id: The account ID. + :param str account_id: The account ID of the specified user. :param str iam_id: The user's IAM ID. :param dict headers: A `dict` containing the request headers :return: A `DetailedResponse` containing the result, headers and HTTP status code. :rtype: DetailedResponse with `dict` result representing a `UserSettings` object """ - if account_id is None: + if not account_id: raise ValueError('account_id must be provided') - if iam_id is None: + if not iam_id: raise ValueError('iam_id must be provided') headers = {} sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, @@ -430,6 +557,7 @@ def get_user_settings(self, if 'headers' in kwargs: headers.update(kwargs.get('headers')) + del kwargs['headers'] headers['Accept'] = 'application/json' path_param_keys = ['account_id', 'iam_id'] @@ -440,7 +568,7 @@ def get_user_settings(self, url=url, headers=headers) - response = self.send(request) + response = self.send(request, **kwargs) return response @@ -464,7 +592,7 @@ def update_user_settings(self, `notification_language` fields. If `self_manage` is `true`, the user can also update the `allowed_ip_addresses` field. - :param str account_id: The account ID. + :param str account_id: The account ID of the specified user. :param str iam_id: The user's IAM ID. :param str language: (optional) The console UI language. By default, this field is empty. @@ -479,9 +607,9 @@ def update_user_settings(self, :rtype: DetailedResponse """ - if account_id is None: + if not account_id: raise ValueError('account_id must be provided') - if iam_id is None: + if not iam_id: raise ValueError('iam_id must be provided') headers = {} sdk_headers = get_sdk_headers(service_name=self.DEFAULT_SERVICE_NAME, @@ -501,6 +629,7 @@ def update_user_settings(self, if 'headers' in kwargs: headers.update(kwargs.get('headers')) + del kwargs['headers'] path_param_keys = ['account_id', 'iam_id'] path_param_values = self.encode_path_vars(account_id, iam_id) @@ -511,7 +640,7 @@ def update_user_settings(self, headers=headers, data=data) - response = self.send(request) + response = self.send(request, **kwargs) return response @@ -617,7 +746,7 @@ def from_dict(cls, _dict: Dict) -> 'InvitedUserList': """Initialize a InvitedUserList object from a json dictionary.""" args = {} if 'resources' in _dict: - args['resources'] = [InvitedUser.from_dict(x) for x in _dict.get('resources')] + args['resources'] = [InvitedUser.from_dict(v) for v in _dict.get('resources')] return cls(**args) @classmethod @@ -629,7 +758,13 @@ def to_dict(self) -> Dict: """Return a json dictionary representing this model.""" _dict = {} if hasattr(self, 'resources') and self.resources is not None: - _dict['resources'] = [x.to_dict() for x in self.resources] + resources_list = [] + for v in self.resources: + if isinstance(v, dict): + resources_list.append(v) + else: + resources_list.append(v.to_dict()) + _dict['resources'] = resources_list return _dict def _to_dict(self): @@ -701,7 +836,7 @@ def from_dict(cls, _dict: Dict) -> 'UserList': if 'next_url' in _dict: args['next_url'] = _dict.get('next_url') if 'resources' in _dict: - args['resources'] = [UserProfile.from_dict(x) for x in _dict.get('resources')] + args['resources'] = [UserProfile.from_dict(v) for v in _dict.get('resources')] return cls(**args) @classmethod @@ -721,7 +856,13 @@ def to_dict(self) -> Dict: if hasattr(self, 'next_url') and self.next_url is not None: _dict['next_url'] = self.next_url if hasattr(self, 'resources') and self.resources is not None: - _dict['resources'] = [x.to_dict() for x in self.resources] + resources_list = [] + for v in self.resources: + if isinstance(v, dict): + resources_list.append(v) + else: + resources_list.append(v.to_dict()) + _dict['resources'] = resources_list return _dict def _to_dict(self): @@ -763,6 +904,8 @@ class UserProfile(): :attr str photo: (optional) A link to a photo of the user. :attr str account_id: (optional) An alphanumeric value identifying the account ID. + :attr str added_on: (optional) The timestamp for when the user was added to the + account. """ def __init__(self, @@ -778,7 +921,8 @@ def __init__(self, phonenumber: str = None, altphonenumber: str = None, photo: str = None, - account_id: str = None) -> None: + account_id: str = None, + added_on: str = None) -> None: """ Initialize a UserProfile object. @@ -801,6 +945,8 @@ def __init__(self, :param str photo: (optional) A link to a photo of the user. :param str account_id: (optional) An alphanumeric value identifying the account ID. + :param str added_on: (optional) The timestamp for when the user was added + to the account. """ self.id = id self.iam_id = iam_id @@ -814,6 +960,7 @@ def __init__(self, self.altphonenumber = altphonenumber self.photo = photo self.account_id = account_id + self.added_on = added_on @classmethod def from_dict(cls, _dict: Dict) -> 'UserProfile': @@ -843,6 +990,8 @@ def from_dict(cls, _dict: Dict) -> 'UserProfile': args['photo'] = _dict.get('photo') if 'account_id' in _dict: args['account_id'] = _dict.get('account_id') + if 'added_on' in _dict: + args['added_on'] = _dict.get('added_on') return cls(**args) @classmethod @@ -877,6 +1026,8 @@ def to_dict(self) -> Dict: _dict['photo'] = self.photo if hasattr(self, 'account_id') and self.account_id is not None: _dict['account_id'] = self.account_id + if hasattr(self, 'added_on') and self.added_on is not None: + _dict['added_on'] = self.added_on return _dict def _to_dict(self): @@ -1146,9 +1297,9 @@ def from_dict(cls, _dict: Dict) -> 'InviteUserIamPolicy': else: raise ValueError('Required property \'type\' not present in InviteUserIamPolicy JSON') if 'roles' in _dict: - args['roles'] = [Role.from_dict(x) for x in _dict.get('roles')] + args['roles'] = [Role.from_dict(v) for v in _dict.get('roles')] if 'resources' in _dict: - args['resources'] = [Resource.from_dict(x) for x in _dict.get('resources')] + args['resources'] = [Resource.from_dict(v) for v in _dict.get('resources')] return cls(**args) @classmethod @@ -1162,9 +1313,21 @@ def to_dict(self) -> Dict: if hasattr(self, 'type') and self.type is not None: _dict['type'] = self.type if hasattr(self, 'roles') and self.roles is not None: - _dict['roles'] = [x.to_dict() for x in self.roles] + roles_list = [] + for v in self.roles: + if isinstance(v, dict): + roles_list.append(v) + else: + roles_list.append(v.to_dict()) + _dict['roles'] = roles_list if hasattr(self, 'resources') and self.resources is not None: - _dict['resources'] = [x.to_dict() for x in self.resources] + resources_list = [] + for v in self.resources: + if isinstance(v, dict): + resources_list.append(v) + else: + resources_list.append(v.to_dict()) + _dict['resources'] = resources_list return _dict def _to_dict(self): @@ -1207,7 +1370,7 @@ def from_dict(cls, _dict: Dict) -> 'Resource': """Initialize a Resource object from a json dictionary.""" args = {} if 'attributes' in _dict: - args['attributes'] = [Attribute.from_dict(x) for x in _dict.get('attributes')] + args['attributes'] = [Attribute.from_dict(v) for v in _dict.get('attributes')] return cls(**args) @classmethod @@ -1219,7 +1382,13 @@ def to_dict(self) -> Dict: """Return a json dictionary representing this model.""" _dict = {} if hasattr(self, 'attributes') and self.attributes is not None: - _dict['attributes'] = [x.to_dict() for x in self.attributes] + attributes_list = [] + for v in self.attributes: + if isinstance(v, dict): + attributes_list.append(v) + else: + attributes_list.append(v.to_dict()) + _dict['attributes'] = attributes_list return _dict def _to_dict(self): @@ -1295,3 +1464,77 @@ def __eq__(self, other: 'Role') -> bool: def __ne__(self, other: 'Role') -> bool: """Return `true` when self and other are not equal, false otherwise.""" return not self == other + +############################################################################## +# Pagers +############################################################################## + +class UsersPager(): + """ + UsersPager can be used to simplify the use of the "list_users" method. + """ + + def __init__(self, + *, + client: UserManagementV1, + account_id: str, + limit: int = None, + user_id: str = None, + ) -> None: + """ + Initialize a UsersPager object. + :param str account_id: The account ID of the specified user. + :param int limit: (optional) The number of results to be returned. + :param str user_id: (optional) Filter users based on their user ID. + """ + self._has_next = True + self._client = client + self._page_context = { 'next': None } + self._account_id = account_id + self._limit = limit + self._user_id = user_id + + def has_next(self) -> bool: + """ + Returns true if there are potentially more results to be retrieved. + """ + return self._has_next + + def get_next(self) -> List[dict]: + """ + Returns the next page of results. + :return: A List[dict], where each element is a dict that represents an instance of UserProfile. + :rtype: List[dict] + """ + if not self.has_next(): + raise StopIteration(message='No more results available') + + result = self._client.list_users( + account_id=self._account_id, + limit=self._limit, + user_id=self._user_id, + start=self._page_context.get('next'), + ).get_result() + + next = None + next_page_link = result.get('next_url') + if next_page_link is not None: + next = get_query_param(next_page_link, '_start') + self._page_context['next'] = next + if next is None: + self._has_next = False + + return result.get('resources') + + def get_all(self) -> List[dict]: + """ + Returns all results by invoking get_next() repeatedly + until all pages of results have been retrieved. + :return: A List[dict], where each element is a dict that represents an instance of UserProfile. + :rtype: List[dict] + """ + results = [] + while self.has_next(): + next_page = self.get_next() + results.extend(next_page) + return results diff --git a/test/integration/test_user_management_v1.py b/test/integration/test_user_management_v1.py index 0e481e79..3d405670 100644 --- a/test/integration/test_user_management_v1.py +++ b/test/integration/test_user_management_v1.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -# (C) Copyright IBM Corp. 2020. +# (C) Copyright IBM Corp. 2020, 2022. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -114,6 +114,30 @@ def test_03_list_users(self): num_users = len(results) print(f'\nlist_users() returned a total of {num_users} users.') + def test_03a_list_users_with_pager(self): + all_results = [] + + # Test get_next(). + pager = UsersPager( + client=self.user_management_service, + account_id=self.ACCOUNT_ID, + ) + while pager.has_next(): + next_page = pager.get_next() + assert next_page is not None + all_results.extend(next_page) + + # Test get_all(). + pager = UsersPager( + client=self.user_management_service, + account_id=self.ACCOUNT_ID, + ) + all_items = pager.get_all() + assert all_items is not None + + assert len(all_results) == len(all_items) + print(f'\nlist_users() returned a total of {len(all_results)} items(s) using UsersPager.') + def test_04_invite_users(self): # Construct a dict representation of a InviteUser model diff --git a/test/unit/test_user_management_v1.py b/test/unit/test_user_management_v1.py index 0c90212d..98fb9487 100644 --- a/test/unit/test_user_management_v1.py +++ b/test/unit/test_user_management_v1.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -# (C) Copyright IBM Corp. 2020. +# (C) Copyright IBM Corp. 2022. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -20,6 +20,7 @@ from ibm_cloud_sdk_core.authenticators.no_auth_authenticator import NoAuthAuthenticator import inspect import json +import os import pytest import re import requests @@ -28,31 +29,76 @@ from ibm_platform_services.user_management_v1 import * -service = UserManagementV1( +_service = UserManagementV1( authenticator=NoAuthAuthenticator() - ) +) + +_base_url = 'https://user-management.cloud.ibm.com' +_service.set_service_url(_base_url) + + +def preprocess_url(operation_path: str): + """ + Returns the request url associated with the specified operation path. + This will be base_url concatenated with a quoted version of operation_path. + The returned request URL is used to register the mock response so it needs + to match the request URL that is formed by the requests library. + """ + # First, unquote the path since it might have some quoted/escaped characters in it + # due to how the generator inserts the operation paths into the unit test code. + operation_path = urllib.parse.unquote(operation_path) + + # Next, quote the path using urllib so that we approximate what will + # happen during request processing. + operation_path = urllib.parse.quote(operation_path, safe='/') + + # Finally, form the request URL from the base URL and operation path. + request_url = _base_url + operation_path + + # If the request url does NOT end with a /, then just return it as-is. + # Otherwise, return a regular expression that matches one or more trailing /. + if re.fullmatch('.*/+', request_url) is None: + return request_url + else: + return re.compile(request_url.rstrip('/') + '/+') -base_url = 'https://user-management.cloud.ibm.com' -service.set_service_url(base_url) ############################################################################## # Start of Service: Users ############################################################################## # region -class TestListUsers(): +class TestNewInstance(): """ - Test Class for list_users + Test Class for new_instance """ - def preprocess_url(self, request_url: str): + def test_new_instance(self): + """ + new_instance() + """ + os.environ['TEST_SERVICE_AUTH_TYPE'] = 'noAuth' + + service = UserManagementV1.new_instance( + service_name='TEST_SERVICE', + ) + + assert service is not None + assert isinstance(service, UserManagementV1) + + def test_new_instance_without_authenticator(self): """ - Preprocess the request URL to ensure the mock response will be found. + new_instance_without_authenticator() """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') + with pytest.raises(ValueError, match='authenticator must be provided'): + service = UserManagementV1.new_instance( + service_name='TEST_SERVICE_NOT_FOUND', + ) + +class TestListUsers(): + """ + Test Class for list_users + """ @responses.activate def test_list_users_all_params(self): @@ -60,8 +106,8 @@ def test_list_users_all_params(self): list_users() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users') - mock_response = '{"total_results": 13, "limit": 5, "first_url": "first_url", "next_url": "next_url", "resources": [{"id": "id", "iam_id": "iam_id", "realm": "realm", "user_id": "user_id", "firstname": "firstname", "lastname": "lastname", "state": "state", "email": "email", "phonenumber": "phonenumber", "altphonenumber": "altphonenumber", "photo": "photo", "account_id": "account_id"}]}' + url = preprocess_url('/v2/accounts/testString/users') + mock_response = '{"total_results": 13, "limit": 5, "first_url": "first_url", "next_url": "next_url", "resources": [{"id": "id", "iam_id": "iam_id", "realm": "realm", "user_id": "user_id", "firstname": "firstname", "lastname": "lastname", "state": "state", "email": "email", "phonenumber": "phonenumber", "altphonenumber": "altphonenumber", "photo": "photo", "account_id": "account_id", "added_on": "added_on"}]}' responses.add(responses.GET, url, body=mock_response, @@ -70,16 +116,16 @@ def test_list_users_all_params(self): # Set up parameter values account_id = 'testString' - state = 'testString' limit = 100 start = 'testString' + user_id = 'testString' # Invoke method - response = service.list_users( + response = _service.list_users( account_id, - state=state, limit=limit, start=start, + user_id=user_id, headers={} ) @@ -89,10 +135,18 @@ def test_list_users_all_params(self): # Validate query params query_string = responses.calls[0].request.url.split('?',1)[1] query_string = urllib.parse.unquote_plus(query_string) - assert 'state={}'.format(state) in query_string assert 'limit={}'.format(limit) in query_string assert '_start={}'.format(start) in query_string + assert 'user_id={}'.format(user_id) in query_string + def test_list_users_all_params_with_retries(self): + # Enable retries and run test_list_users_all_params. + _service.enable_retries() + self.test_list_users_all_params() + + # Disable retries and run test_list_users_all_params. + _service.disable_retries() + self.test_list_users_all_params() @responses.activate def test_list_users_required_params(self): @@ -100,8 +154,8 @@ def test_list_users_required_params(self): test_list_users_required_params() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users') - mock_response = '{"total_results": 13, "limit": 5, "first_url": "first_url", "next_url": "next_url", "resources": [{"id": "id", "iam_id": "iam_id", "realm": "realm", "user_id": "user_id", "firstname": "firstname", "lastname": "lastname", "state": "state", "email": "email", "phonenumber": "phonenumber", "altphonenumber": "altphonenumber", "photo": "photo", "account_id": "account_id"}]}' + url = preprocess_url('/v2/accounts/testString/users') + mock_response = '{"total_results": 13, "limit": 5, "first_url": "first_url", "next_url": "next_url", "resources": [{"id": "id", "iam_id": "iam_id", "realm": "realm", "user_id": "user_id", "firstname": "firstname", "lastname": "lastname", "state": "state", "email": "email", "phonenumber": "phonenumber", "altphonenumber": "altphonenumber", "photo": "photo", "account_id": "account_id", "added_on": "added_on"}]}' responses.add(responses.GET, url, body=mock_response, @@ -112,7 +166,7 @@ def test_list_users_required_params(self): account_id = 'testString' # Invoke method - response = service.list_users( + response = _service.list_users( account_id, headers={} ) @@ -121,6 +175,14 @@ def test_list_users_required_params(self): assert len(responses.calls) == 1 assert response.status_code == 200 + def test_list_users_required_params_with_retries(self): + # Enable retries and run test_list_users_required_params. + _service.enable_retries() + self.test_list_users_required_params() + + # Disable retries and run test_list_users_required_params. + _service.disable_retries() + self.test_list_users_required_params() @responses.activate def test_list_users_value_error(self): @@ -128,8 +190,8 @@ def test_list_users_value_error(self): test_list_users_value_error() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users') - mock_response = '{"total_results": 13, "limit": 5, "first_url": "first_url", "next_url": "next_url", "resources": [{"id": "id", "iam_id": "iam_id", "realm": "realm", "user_id": "user_id", "firstname": "firstname", "lastname": "lastname", "state": "state", "email": "email", "phonenumber": "phonenumber", "altphonenumber": "altphonenumber", "photo": "photo", "account_id": "account_id"}]}' + url = preprocess_url('/v2/accounts/testString/users') + mock_response = '{"total_results": 13, "limit": 5, "first_url": "first_url", "next_url": "next_url", "resources": [{"id": "id", "iam_id": "iam_id", "realm": "realm", "user_id": "user_id", "firstname": "firstname", "lastname": "lastname", "state": "state", "email": "email", "phonenumber": "phonenumber", "altphonenumber": "altphonenumber", "photo": "photo", "account_id": "account_id", "added_on": "added_on"}]}' responses.add(responses.GET, url, body=mock_response, @@ -146,31 +208,94 @@ def test_list_users_value_error(self): for param in req_param_dict.keys(): req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} with pytest.raises(ValueError): - service.list_users(**req_copy) + _service.list_users(**req_copy) + + def test_list_users_value_error_with_retries(self): + # Enable retries and run test_list_users_value_error. + _service.enable_retries() + self.test_list_users_value_error() + + # Disable retries and run test_list_users_value_error. + _service.disable_retries() + self.test_list_users_value_error() + + @responses.activate + def test_list_users_with_pager_get_next(self): + """ + test_list_users_with_pager_get_next() + """ + # Set up a two-page mock response + url = preprocess_url('/v2/accounts/testString/users') + mock_response1 = '{"total_count":2,"limit":1,"next_url":"https://myhost.com/somePath?_start=1","resources":[{"id":"id","iam_id":"iam_id","realm":"realm","user_id":"user_id","firstname":"firstname","lastname":"lastname","state":"state","email":"email","phonenumber":"phonenumber","altphonenumber":"altphonenumber","photo":"photo","account_id":"account_id","added_on":"added_on"}]}' + mock_response2 = '{"total_count":2,"limit":1,"resources":[{"id":"id","iam_id":"iam_id","realm":"realm","user_id":"user_id","firstname":"firstname","lastname":"lastname","state":"state","email":"email","phonenumber":"phonenumber","altphonenumber":"altphonenumber","photo":"photo","account_id":"account_id","added_on":"added_on"}]}' + responses.add(responses.GET, + url, + body=mock_response1, + content_type='application/json', + status=200) + responses.add(responses.GET, + url, + body=mock_response2, + content_type='application/json', + status=200) + + # Exercise the pager class for this operation + all_results = [] + pager = UsersPager( + client=_service, + account_id='testString', + limit=10, + user_id='testString', + ) + while pager.has_next(): + next_page = pager.get_next() + assert next_page is not None + all_results.extend(next_page) + assert len(all_results) == 2 + @responses.activate + def test_list_users_with_pager_get_all(self): + """ + test_list_users_with_pager_get_all() + """ + # Set up a two-page mock response + url = preprocess_url('/v2/accounts/testString/users') + mock_response1 = '{"total_count":2,"limit":1,"next_url":"https://myhost.com/somePath?_start=1","resources":[{"id":"id","iam_id":"iam_id","realm":"realm","user_id":"user_id","firstname":"firstname","lastname":"lastname","state":"state","email":"email","phonenumber":"phonenumber","altphonenumber":"altphonenumber","photo":"photo","account_id":"account_id","added_on":"added_on"}]}' + mock_response2 = '{"total_count":2,"limit":1,"resources":[{"id":"id","iam_id":"iam_id","realm":"realm","user_id":"user_id","firstname":"firstname","lastname":"lastname","state":"state","email":"email","phonenumber":"phonenumber","altphonenumber":"altphonenumber","photo":"photo","account_id":"account_id","added_on":"added_on"}]}' + responses.add(responses.GET, + url, + body=mock_response1, + content_type='application/json', + status=200) + responses.add(responses.GET, + url, + body=mock_response2, + content_type='application/json', + status=200) + # Exercise the pager class for this operation + pager = UsersPager( + client=_service, + account_id='testString', + limit=10, + user_id='testString', + ) + all_results = pager.get_all() + assert all_results is not None + assert len(all_results) == 2 class TestInviteUsers(): """ Test Class for invite_users """ - def preprocess_url(self, request_url: str): - """ - Preprocess the request URL to ensure the mock response will be found. - """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') - @responses.activate def test_invite_users_all_params(self): """ invite_users() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users') + url = preprocess_url('/v2/accounts/testString/users') mock_response = '{"resources": [{"email": "email", "id": "id", "state": "state"}]}' responses.add(responses.POST, url, @@ -209,7 +334,7 @@ def test_invite_users_all_params(self): access_groups = ['testString'] # Invoke method - response = service.invite_users( + response = _service.invite_users( account_id, users=users, iam_policy=iam_policy, @@ -226,6 +351,14 @@ def test_invite_users_all_params(self): assert req_body['iam_policy'] == [invite_user_iam_policy_model] assert req_body['access_groups'] == ['testString'] + def test_invite_users_all_params_with_retries(self): + # Enable retries and run test_invite_users_all_params. + _service.enable_retries() + self.test_invite_users_all_params() + + # Disable retries and run test_invite_users_all_params. + _service.disable_retries() + self.test_invite_users_all_params() @responses.activate def test_invite_users_required_params(self): @@ -233,7 +366,7 @@ def test_invite_users_required_params(self): test_invite_users_required_params() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users') + url = preprocess_url('/v2/accounts/testString/users') mock_response = '{"resources": [{"email": "email", "id": "id", "state": "state"}]}' responses.add(responses.POST, url, @@ -245,7 +378,7 @@ def test_invite_users_required_params(self): account_id = 'testString' # Invoke method - response = service.invite_users( + response = _service.invite_users( account_id, headers={} ) @@ -254,6 +387,14 @@ def test_invite_users_required_params(self): assert len(responses.calls) == 1 assert response.status_code == 202 + def test_invite_users_required_params_with_retries(self): + # Enable retries and run test_invite_users_required_params. + _service.enable_retries() + self.test_invite_users_required_params() + + # Disable retries and run test_invite_users_required_params. + _service.disable_retries() + self.test_invite_users_required_params() @responses.activate def test_invite_users_value_error(self): @@ -261,7 +402,7 @@ def test_invite_users_value_error(self): test_invite_users_value_error() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users') + url = preprocess_url('/v2/accounts/testString/users') mock_response = '{"resources": [{"email": "email", "id": "id", "state": "state"}]}' responses.add(responses.POST, url, @@ -279,32 +420,74 @@ def test_invite_users_value_error(self): for param in req_param_dict.keys(): req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} with pytest.raises(ValueError): - service.invite_users(**req_copy) + _service.invite_users(**req_copy) + def test_invite_users_value_error_with_retries(self): + # Enable retries and run test_invite_users_value_error. + _service.enable_retries() + self.test_invite_users_value_error() + # Disable retries and run test_invite_users_value_error. + _service.disable_retries() + self.test_invite_users_value_error() class TestGetUserProfile(): """ Test Class for get_user_profile """ - def preprocess_url(self, request_url: str): + @responses.activate + def test_get_user_profile_all_params(self): """ - Preprocess the request URL to ensure the mock response will be found. + get_user_profile() """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') + # Set up mock + url = preprocess_url('/v2/accounts/testString/users/testString') + mock_response = '{"id": "id", "iam_id": "iam_id", "realm": "realm", "user_id": "user_id", "firstname": "firstname", "lastname": "lastname", "state": "state", "email": "email", "phonenumber": "phonenumber", "altphonenumber": "altphonenumber", "photo": "photo", "account_id": "account_id", "added_on": "added_on"}' + responses.add(responses.GET, + url, + body=mock_response, + content_type='application/json', + status=200) + + # Set up parameter values + account_id = 'testString' + iam_id = 'testString' + include_activity = 'testString' + + # Invoke method + response = _service.get_user_profile( + account_id, + iam_id, + include_activity=include_activity, + headers={} + ) + + # Check for correct operation + assert len(responses.calls) == 1 + assert response.status_code == 200 + # Validate query params + query_string = responses.calls[0].request.url.split('?',1)[1] + query_string = urllib.parse.unquote_plus(query_string) + assert 'include_activity={}'.format(include_activity) in query_string + + def test_get_user_profile_all_params_with_retries(self): + # Enable retries and run test_get_user_profile_all_params. + _service.enable_retries() + self.test_get_user_profile_all_params() + + # Disable retries and run test_get_user_profile_all_params. + _service.disable_retries() + self.test_get_user_profile_all_params() @responses.activate - def test_get_user_profile_all_params(self): + def test_get_user_profile_required_params(self): """ - get_user_profile() + test_get_user_profile_required_params() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users/testString') - mock_response = '{"id": "id", "iam_id": "iam_id", "realm": "realm", "user_id": "user_id", "firstname": "firstname", "lastname": "lastname", "state": "state", "email": "email", "phonenumber": "phonenumber", "altphonenumber": "altphonenumber", "photo": "photo", "account_id": "account_id"}' + url = preprocess_url('/v2/accounts/testString/users/testString') + mock_response = '{"id": "id", "iam_id": "iam_id", "realm": "realm", "user_id": "user_id", "firstname": "firstname", "lastname": "lastname", "state": "state", "email": "email", "phonenumber": "phonenumber", "altphonenumber": "altphonenumber", "photo": "photo", "account_id": "account_id", "added_on": "added_on"}' responses.add(responses.GET, url, body=mock_response, @@ -316,7 +499,7 @@ def test_get_user_profile_all_params(self): iam_id = 'testString' # Invoke method - response = service.get_user_profile( + response = _service.get_user_profile( account_id, iam_id, headers={} @@ -326,6 +509,14 @@ def test_get_user_profile_all_params(self): assert len(responses.calls) == 1 assert response.status_code == 200 + def test_get_user_profile_required_params_with_retries(self): + # Enable retries and run test_get_user_profile_required_params. + _service.enable_retries() + self.test_get_user_profile_required_params() + + # Disable retries and run test_get_user_profile_required_params. + _service.disable_retries() + self.test_get_user_profile_required_params() @responses.activate def test_get_user_profile_value_error(self): @@ -333,8 +524,8 @@ def test_get_user_profile_value_error(self): test_get_user_profile_value_error() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users/testString') - mock_response = '{"id": "id", "iam_id": "iam_id", "realm": "realm", "user_id": "user_id", "firstname": "firstname", "lastname": "lastname", "state": "state", "email": "email", "phonenumber": "phonenumber", "altphonenumber": "altphonenumber", "photo": "photo", "account_id": "account_id"}' + url = preprocess_url('/v2/accounts/testString/users/testString') + mock_response = '{"id": "id", "iam_id": "iam_id", "realm": "realm", "user_id": "user_id", "firstname": "firstname", "lastname": "lastname", "state": "state", "email": "email", "phonenumber": "phonenumber", "altphonenumber": "altphonenumber", "photo": "photo", "account_id": "account_id", "added_on": "added_on"}' responses.add(responses.GET, url, body=mock_response, @@ -353,31 +544,29 @@ def test_get_user_profile_value_error(self): for param in req_param_dict.keys(): req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} with pytest.raises(ValueError): - service.get_user_profile(**req_copy) + _service.get_user_profile(**req_copy) + def test_get_user_profile_value_error_with_retries(self): + # Enable retries and run test_get_user_profile_value_error. + _service.enable_retries() + self.test_get_user_profile_value_error() + # Disable retries and run test_get_user_profile_value_error. + _service.disable_retries() + self.test_get_user_profile_value_error() class TestUpdateUserProfile(): """ Test Class for update_user_profile """ - def preprocess_url(self, request_url: str): - """ - Preprocess the request URL to ensure the mock response will be found. - """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') - @responses.activate def test_update_user_profile_all_params(self): """ update_user_profile() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users/testString') + url = preprocess_url('/v2/accounts/testString/users/testString') responses.add(responses.PATCH, url, status=204) @@ -392,9 +581,10 @@ def test_update_user_profile_all_params(self): phonenumber = 'testString' altphonenumber = 'testString' photo = 'testString' + include_activity = 'testString' # Invoke method - response = service.update_user_profile( + response = _service.update_user_profile( account_id, iam_id, firstname=firstname, @@ -404,12 +594,17 @@ def test_update_user_profile_all_params(self): phonenumber=phonenumber, altphonenumber=altphonenumber, photo=photo, + include_activity=include_activity, headers={} ) # Check for correct operation assert len(responses.calls) == 1 assert response.status_code == 204 + # Validate query params + query_string = responses.calls[0].request.url.split('?',1)[1] + query_string = urllib.parse.unquote_plus(query_string) + assert 'include_activity={}'.format(include_activity) in query_string # Validate body params req_body = json.loads(str(responses.calls[0].request.body, 'utf-8')) assert req_body['firstname'] == 'testString' @@ -420,6 +615,14 @@ def test_update_user_profile_all_params(self): assert req_body['altphonenumber'] == 'testString' assert req_body['photo'] == 'testString' + def test_update_user_profile_all_params_with_retries(self): + # Enable retries and run test_update_user_profile_all_params. + _service.enable_retries() + self.test_update_user_profile_all_params() + + # Disable retries and run test_update_user_profile_all_params. + _service.disable_retries() + self.test_update_user_profile_all_params() @responses.activate def test_update_user_profile_required_params(self): @@ -427,7 +630,7 @@ def test_update_user_profile_required_params(self): test_update_user_profile_required_params() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users/testString') + url = preprocess_url('/v2/accounts/testString/users/testString') responses.add(responses.PATCH, url, status=204) @@ -437,7 +640,7 @@ def test_update_user_profile_required_params(self): iam_id = 'testString' # Invoke method - response = service.update_user_profile( + response = _service.update_user_profile( account_id, iam_id, headers={} @@ -447,6 +650,14 @@ def test_update_user_profile_required_params(self): assert len(responses.calls) == 1 assert response.status_code == 204 + def test_update_user_profile_required_params_with_retries(self): + # Enable retries and run test_update_user_profile_required_params. + _service.enable_retries() + self.test_update_user_profile_required_params() + + # Disable retries and run test_update_user_profile_required_params. + _service.disable_retries() + self.test_update_user_profile_required_params() @responses.activate def test_update_user_profile_value_error(self): @@ -454,7 +665,7 @@ def test_update_user_profile_value_error(self): test_update_user_profile_value_error() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users/testString') + url = preprocess_url('/v2/accounts/testString/users/testString') responses.add(responses.PATCH, url, status=204) @@ -471,31 +682,70 @@ def test_update_user_profile_value_error(self): for param in req_param_dict.keys(): req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} with pytest.raises(ValueError): - service.update_user_profile(**req_copy) + _service.update_user_profile(**req_copy) + def test_update_user_profile_value_error_with_retries(self): + # Enable retries and run test_update_user_profile_value_error. + _service.enable_retries() + self.test_update_user_profile_value_error() + # Disable retries and run test_update_user_profile_value_error. + _service.disable_retries() + self.test_update_user_profile_value_error() class TestRemoveUser(): """ Test Class for remove_user """ - def preprocess_url(self, request_url: str): + @responses.activate + def test_remove_user_all_params(self): """ - Preprocess the request URL to ensure the mock response will be found. + remove_user() """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') + # Set up mock + url = preprocess_url('/v2/accounts/testString/users/testString') + responses.add(responses.DELETE, + url, + status=204) + + # Set up parameter values + account_id = 'testString' + iam_id = 'testString' + include_activity = 'testString' + + # Invoke method + response = _service.remove_user( + account_id, + iam_id, + include_activity=include_activity, + headers={} + ) + + # Check for correct operation + assert len(responses.calls) == 1 + assert response.status_code == 204 + # Validate query params + query_string = responses.calls[0].request.url.split('?',1)[1] + query_string = urllib.parse.unquote_plus(query_string) + assert 'include_activity={}'.format(include_activity) in query_string + + def test_remove_user_all_params_with_retries(self): + # Enable retries and run test_remove_user_all_params. + _service.enable_retries() + self.test_remove_user_all_params() + + # Disable retries and run test_remove_user_all_params. + _service.disable_retries() + self.test_remove_user_all_params() @responses.activate - def test_remove_user_all_params(self): + def test_remove_user_required_params(self): """ - remove_user() + test_remove_user_required_params() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users/testString') + url = preprocess_url('/v2/accounts/testString/users/testString') responses.add(responses.DELETE, url, status=204) @@ -505,7 +755,7 @@ def test_remove_user_all_params(self): iam_id = 'testString' # Invoke method - response = service.remove_user( + response = _service.remove_user( account_id, iam_id, headers={} @@ -515,6 +765,14 @@ def test_remove_user_all_params(self): assert len(responses.calls) == 1 assert response.status_code == 204 + def test_remove_user_required_params_with_retries(self): + # Enable retries and run test_remove_user_required_params. + _service.enable_retries() + self.test_remove_user_required_params() + + # Disable retries and run test_remove_user_required_params. + _service.disable_retries() + self.test_remove_user_required_params() @responses.activate def test_remove_user_value_error(self): @@ -522,7 +780,7 @@ def test_remove_user_value_error(self): test_remove_user_value_error() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users/testString') + url = preprocess_url('/v2/accounts/testString/users/testString') responses.add(responses.DELETE, url, status=204) @@ -539,9 +797,159 @@ def test_remove_user_value_error(self): for param in req_param_dict.keys(): req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} with pytest.raises(ValueError): - service.remove_user(**req_copy) + _service.remove_user(**req_copy) + + def test_remove_user_value_error_with_retries(self): + # Enable retries and run test_remove_user_value_error. + _service.enable_retries() + self.test_remove_user_value_error() + + # Disable retries and run test_remove_user_value_error. + _service.disable_retries() + self.test_remove_user_value_error() + +class TestAccept(): + """ + Test Class for accept + """ + + @responses.activate + def test_accept_all_params(self): + """ + accept() + """ + # Set up mock + url = preprocess_url('/v2/users/accept') + responses.add(responses.POST, + url, + status=202) + + # Set up parameter values + account_id = 'testString' + + # Invoke method + response = _service.accept( + account_id=account_id, + headers={} + ) + + # Check for correct operation + assert len(responses.calls) == 1 + assert response.status_code == 202 + # Validate body params + req_body = json.loads(str(responses.calls[0].request.body, 'utf-8')) + assert req_body['account_id'] == 'testString' + + def test_accept_all_params_with_retries(self): + # Enable retries and run test_accept_all_params. + _service.enable_retries() + self.test_accept_all_params() + + # Disable retries and run test_accept_all_params. + _service.disable_retries() + self.test_accept_all_params() + + @responses.activate + def test_accept_required_params(self): + """ + test_accept_required_params() + """ + # Set up mock + url = preprocess_url('/v2/users/accept') + responses.add(responses.POST, + url, + status=202) + + # Invoke method + response = _service.accept() + # Check for correct operation + assert len(responses.calls) == 1 + assert response.status_code == 202 + + def test_accept_required_params_with_retries(self): + # Enable retries and run test_accept_required_params. + _service.enable_retries() + self.test_accept_required_params() + + # Disable retries and run test_accept_required_params. + _service.disable_retries() + self.test_accept_required_params() + +class TestV3RemoveUser(): + """ + Test Class for v3_remove_user + """ + + @responses.activate + def test_v3_remove_user_all_params(self): + """ + v3_remove_user() + """ + # Set up mock + url = preprocess_url('/v3/accounts/testString/users/testString') + responses.add(responses.DELETE, + url, + status=202) + + # Set up parameter values + account_id = 'testString' + iam_id = 'testString' + + # Invoke method + response = _service.v3_remove_user( + account_id, + iam_id, + headers={} + ) + + # Check for correct operation + assert len(responses.calls) == 1 + assert response.status_code == 202 + + def test_v3_remove_user_all_params_with_retries(self): + # Enable retries and run test_v3_remove_user_all_params. + _service.enable_retries() + self.test_v3_remove_user_all_params() + + # Disable retries and run test_v3_remove_user_all_params. + _service.disable_retries() + self.test_v3_remove_user_all_params() + + @responses.activate + def test_v3_remove_user_value_error(self): + """ + test_v3_remove_user_value_error() + """ + # Set up mock + url = preprocess_url('/v3/accounts/testString/users/testString') + responses.add(responses.DELETE, + url, + status=202) + + # Set up parameter values + account_id = 'testString' + iam_id = 'testString' + + # Pass in all but one required param and check for a ValueError + req_param_dict = { + "account_id": account_id, + "iam_id": iam_id, + } + for param in req_param_dict.keys(): + req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} + with pytest.raises(ValueError): + _service.v3_remove_user(**req_copy) + + def test_v3_remove_user_value_error_with_retries(self): + # Enable retries and run test_v3_remove_user_value_error. + _service.enable_retries() + self.test_v3_remove_user_value_error() + + # Disable retries and run test_v3_remove_user_value_error. + _service.disable_retries() + self.test_v3_remove_user_value_error() # endregion ############################################################################## @@ -553,19 +961,37 @@ def test_remove_user_value_error(self): ############################################################################## # region -class TestGetUserSettings(): +class TestNewInstance(): """ - Test Class for get_user_settings + Test Class for new_instance """ - def preprocess_url(self, request_url: str): + def test_new_instance(self): """ - Preprocess the request URL to ensure the mock response will be found. + new_instance() """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') + os.environ['TEST_SERVICE_AUTH_TYPE'] = 'noAuth' + + service = UserManagementV1.new_instance( + service_name='TEST_SERVICE', + ) + + assert service is not None + assert isinstance(service, UserManagementV1) + + def test_new_instance_without_authenticator(self): + """ + new_instance_without_authenticator() + """ + with pytest.raises(ValueError, match='authenticator must be provided'): + service = UserManagementV1.new_instance( + service_name='TEST_SERVICE_NOT_FOUND', + ) + +class TestGetUserSettings(): + """ + Test Class for get_user_settings + """ @responses.activate def test_get_user_settings_all_params(self): @@ -573,7 +999,7 @@ def test_get_user_settings_all_params(self): get_user_settings() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users/testString/settings') + url = preprocess_url('/v2/accounts/testString/users/testString/settings') mock_response = '{"language": "language", "notification_language": "notification_language", "allowed_ip_addresses": "32.96.110.50,172.16.254.1", "self_manage": false}' responses.add(responses.GET, url, @@ -586,7 +1012,7 @@ def test_get_user_settings_all_params(self): iam_id = 'testString' # Invoke method - response = service.get_user_settings( + response = _service.get_user_settings( account_id, iam_id, headers={} @@ -596,6 +1022,14 @@ def test_get_user_settings_all_params(self): assert len(responses.calls) == 1 assert response.status_code == 200 + def test_get_user_settings_all_params_with_retries(self): + # Enable retries and run test_get_user_settings_all_params. + _service.enable_retries() + self.test_get_user_settings_all_params() + + # Disable retries and run test_get_user_settings_all_params. + _service.disable_retries() + self.test_get_user_settings_all_params() @responses.activate def test_get_user_settings_value_error(self): @@ -603,7 +1037,7 @@ def test_get_user_settings_value_error(self): test_get_user_settings_value_error() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users/testString/settings') + url = preprocess_url('/v2/accounts/testString/users/testString/settings') mock_response = '{"language": "language", "notification_language": "notification_language", "allowed_ip_addresses": "32.96.110.50,172.16.254.1", "self_manage": false}' responses.add(responses.GET, url, @@ -623,31 +1057,29 @@ def test_get_user_settings_value_error(self): for param in req_param_dict.keys(): req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} with pytest.raises(ValueError): - service.get_user_settings(**req_copy) + _service.get_user_settings(**req_copy) + def test_get_user_settings_value_error_with_retries(self): + # Enable retries and run test_get_user_settings_value_error. + _service.enable_retries() + self.test_get_user_settings_value_error() + # Disable retries and run test_get_user_settings_value_error. + _service.disable_retries() + self.test_get_user_settings_value_error() class TestUpdateUserSettings(): """ Test Class for update_user_settings """ - def preprocess_url(self, request_url: str): - """ - Preprocess the request URL to ensure the mock response will be found. - """ - if re.fullmatch('.*/+', request_url) is None: - return request_url - else: - return re.compile(request_url.rstrip('/') + '/+') - @responses.activate def test_update_user_settings_all_params(self): """ update_user_settings() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users/testString/settings') + url = preprocess_url('/v2/accounts/testString/users/testString/settings') responses.add(responses.PATCH, url, status=204) @@ -661,7 +1093,7 @@ def test_update_user_settings_all_params(self): self_manage = True # Invoke method - response = service.update_user_settings( + response = _service.update_user_settings( account_id, iam_id, language=language, @@ -681,6 +1113,14 @@ def test_update_user_settings_all_params(self): assert req_body['allowed_ip_addresses'] == '32.96.110.50,172.16.254.1' assert req_body['self_manage'] == True + def test_update_user_settings_all_params_with_retries(self): + # Enable retries and run test_update_user_settings_all_params. + _service.enable_retries() + self.test_update_user_settings_all_params() + + # Disable retries and run test_update_user_settings_all_params. + _service.disable_retries() + self.test_update_user_settings_all_params() @responses.activate def test_update_user_settings_required_params(self): @@ -688,7 +1128,7 @@ def test_update_user_settings_required_params(self): test_update_user_settings_required_params() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users/testString/settings') + url = preprocess_url('/v2/accounts/testString/users/testString/settings') responses.add(responses.PATCH, url, status=204) @@ -698,7 +1138,7 @@ def test_update_user_settings_required_params(self): iam_id = 'testString' # Invoke method - response = service.update_user_settings( + response = _service.update_user_settings( account_id, iam_id, headers={} @@ -708,6 +1148,14 @@ def test_update_user_settings_required_params(self): assert len(responses.calls) == 1 assert response.status_code == 204 + def test_update_user_settings_required_params_with_retries(self): + # Enable retries and run test_update_user_settings_required_params. + _service.enable_retries() + self.test_update_user_settings_required_params() + + # Disable retries and run test_update_user_settings_required_params. + _service.disable_retries() + self.test_update_user_settings_required_params() @responses.activate def test_update_user_settings_value_error(self): @@ -715,7 +1163,7 @@ def test_update_user_settings_value_error(self): test_update_user_settings_value_error() """ # Set up mock - url = self.preprocess_url(base_url + '/v2/accounts/testString/users/testString/settings') + url = preprocess_url('/v2/accounts/testString/users/testString/settings') responses.add(responses.PATCH, url, status=204) @@ -732,9 +1180,16 @@ def test_update_user_settings_value_error(self): for param in req_param_dict.keys(): req_copy = {key:val if key is not param else None for (key,val) in req_param_dict.items()} with pytest.raises(ValueError): - service.update_user_settings(**req_copy) + _service.update_user_settings(**req_copy) + def test_update_user_settings_value_error_with_retries(self): + # Enable retries and run test_update_user_settings_value_error. + _service.enable_retries() + self.test_update_user_settings_value_error() + # Disable retries and run test_update_user_settings_value_error. + _service.disable_retries() + self.test_update_user_settings_value_error() # endregion ############################################################################## @@ -746,7 +1201,7 @@ def test_update_user_settings_value_error(self): # Start of Model Tests ############################################################################## # region -class TestInvitedUser(): +class TestModel_InvitedUser(): """ Test Class for InvitedUser """ @@ -777,7 +1232,7 @@ def test_invited_user_serialization(self): invited_user_model_json2 = invited_user_model.to_dict() assert invited_user_model_json2 == invited_user_model_json -class TestInvitedUserList(): +class TestModel_InvitedUserList(): """ Test Class for InvitedUserList """ @@ -813,7 +1268,7 @@ def test_invited_user_list_serialization(self): invited_user_list_model_json2 = invited_user_list_model.to_dict() assert invited_user_list_model_json2 == invited_user_list_model_json -class TestUserList(): +class TestModel_UserList(): """ Test Class for UserList """ @@ -838,6 +1293,7 @@ def test_user_list_serialization(self): user_profile_model['altphonenumber'] = 'testString' user_profile_model['photo'] = 'testString' user_profile_model['account_id'] = 'testString' + user_profile_model['added_on'] = 'testString' # Construct a json representation of a UserList model user_list_model_json = {} @@ -862,7 +1318,7 @@ def test_user_list_serialization(self): user_list_model_json2 = user_list_model.to_dict() assert user_list_model_json2 == user_list_model_json -class TestUserProfile(): +class TestModel_UserProfile(): """ Test Class for UserProfile """ @@ -886,6 +1342,7 @@ def test_user_profile_serialization(self): user_profile_model_json['altphonenumber'] = 'testString' user_profile_model_json['photo'] = 'testString' user_profile_model_json['account_id'] = 'testString' + user_profile_model_json['added_on'] = 'testString' # Construct a model instance of UserProfile by calling from_dict on the json representation user_profile_model = UserProfile.from_dict(user_profile_model_json) @@ -902,7 +1359,7 @@ def test_user_profile_serialization(self): user_profile_model_json2 = user_profile_model.to_dict() assert user_profile_model_json2 == user_profile_model_json -class TestUserSettings(): +class TestModel_UserSettings(): """ Test Class for UserSettings """ @@ -934,7 +1391,7 @@ def test_user_settings_serialization(self): user_settings_model_json2 = user_settings_model.to_dict() assert user_settings_model_json2 == user_settings_model_json -class TestAttribute(): +class TestModel_Attribute(): """ Test Class for Attribute """ @@ -964,7 +1421,7 @@ def test_attribute_serialization(self): attribute_model_json2 = attribute_model.to_dict() assert attribute_model_json2 == attribute_model_json -class TestInviteUser(): +class TestModel_InviteUser(): """ Test Class for InviteUser """ @@ -994,7 +1451,7 @@ def test_invite_user_serialization(self): invite_user_model_json2 = invite_user_model.to_dict() assert invite_user_model_json2 == invite_user_model_json -class TestInviteUserIamPolicy(): +class TestModel_InviteUserIamPolicy(): """ Test Class for InviteUserIamPolicy """ @@ -1037,7 +1494,7 @@ def test_invite_user_iam_policy_serialization(self): invite_user_iam_policy_model_json2 = invite_user_iam_policy_model.to_dict() assert invite_user_iam_policy_model_json2 == invite_user_iam_policy_model_json -class TestResource(): +class TestModel_Resource(): """ Test Class for Resource """ @@ -1072,7 +1529,7 @@ def test_resource_serialization(self): resource_model_json2 = resource_model.to_dict() assert resource_model_json2 == resource_model_json -class TestRole(): +class TestModel_Role(): """ Test Class for Role """