diff --git a/orm/orm_client/ormcli/cmscli.py b/orm/orm_client/ormcli/cmscli.py index b09e00a4..7de88bd4 100644 --- a/orm/orm_client/ormcli/cmscli.py +++ b/orm/orm_client/ormcli/cmscli.py @@ -278,198 +278,6 @@ def add_to_parser(service_sub): parser_list_customer.add_argument('--metadata', action='append', nargs="+", type=str, help='') - # create group - parser_create_group = subparsers.add_parser('create_group', - help='[<"X-RANGER-Client" ' - 'header>] ') - parser_create_group.add_argument('client', - **cli_common.ORM_CLIENT_KWARGS) - parser_create_group.add_argument('datafile', - type=argparse.FileType('r'), - help='') - # delete group - parser_delete_group = subparsers.add_parser('delete_group', - help='[<"X-RANGER-Client" ' - 'header>] ') - parser_delete_group.add_argument('client', **cli_common.ORM_CLIENT_KWARGS) - parser_delete_group.add_argument('groupid', type=str, help='') - - # get group - h1, h2 = '[<"X-RANGER-Client" header>]', '' - parser_get_group = subparsers.add_parser('get_group', - help='%s %s' % (h1, h2)) - parser_get_group.add_argument('client', **cli_common.ORM_CLIENT_KWARGS) - parser_get_group.add_argument('groupid', type=str, help=h2) - - # list groups - h1 = '[<"X-RANGER-Client" header>]' - h2 = '[--region ] [--starts_with ] [--contains ]' - parser_list_groups = subparsers.add_parser('list_groups', - help='%s %s' % (h1, h2)) - parser_list_groups.add_argument('client', **cli_common.ORM_CLIENT_KWARGS) - parser_list_groups.add_argument('--region', type=str, help='region name') - parser_list_groups.add_argument('--starts_with', type=str, - help='group name') - parser_list_groups.add_argument('--contains', type=str, - help='* contains in group name') - - # groups region - parser_add_groups_region = subparsers.add_parser( - 'add_groups_region', - help='[<"X-RANGER-Client" ' - 'header>] ' - '') - parser_add_groups_region.add_argument( - 'client', **cli_common.ORM_CLIENT_KWARGS) - parser_add_groups_region.add_argument( - 'groupid', type=str, help='') - parser_add_groups_region.add_argument( - 'datafile', type=argparse.FileType('r'), - help='') - - parser_delete_groups_region = subparsers.add_parser( - 'delete_groups_region', - help='[<"X-RANGER-Client" header>] [--force_delete] ' - ' ') - parser_delete_groups_region.add_argument('client', - **cli_common.ORM_CLIENT_KWARGS) - parser_delete_groups_region.add_argument('groupid', - type=str, - help='') - parser_delete_groups_region.add_argument('regionid', - type=str, - help='') - parser_delete_groups_region.add_argument('--force_delete', - help='force delete groups region', - action="store_true") - - # assign groups roles - parser_assign_group_roles = subparsers.add_parser( - 'assign_group_roles', - help='[<"X-RANGER-Client" ' - 'header>] ' - '') - parser_assign_group_roles.add_argument( - 'client', **cli_common.ORM_CLIENT_KWARGS) - parser_assign_group_roles.add_argument( - 'groupid', type=str, help='') - parser_assign_group_roles.add_argument( - 'datafile', type=argparse.FileType('r'), - help='') - - # assign group region roles - parser_assign_group_region_roles = subparsers.add_parser( - 'assign_group_region_roles', - help='[<"X-RANGER-Client" ' - 'header>] ' - '') - parser_assign_group_region_roles.add_argument( - 'client', **cli_common.ORM_CLIENT_KWARGS) - parser_assign_group_region_roles.add_argument( - 'groupid', type=str, help='') - parser_assign_group_region_roles.add_argument( - 'regionid', type=str, help='') - parser_assign_group_region_roles.add_argument( - 'datafile', type=argparse.FileType('r'), - help='') - - # unassign group roles - parser_unassign_group_role = subparsers.add_parser( - 'unassign_group_role', - help='[<"X-RANGER-Client" ' - 'header>] <--customer > ' - 'or <--domain >') - parser_unassign_group_role.add_argument( - 'client', **cli_common.ORM_CLIENT_KWARGS) - parser_unassign_group_role.add_argument( - 'groupid', type=str, help='') - parser_unassign_group_role.add_argument( - 'role', type=str, help='') - parser_unassign_group_role.add_argument( - '--customer', type=str, help='customer id') - parser_unassign_group_role.add_argument( - '--domain', type=str, help='domain name') - - # list group roles - h1 = '[<"X-RANGER-Client" header>]' - h2 = ' <--region > <--customer > ' \ - 'or <--domain >' - parser_list_group_roles = subparsers.add_parser('list_group_roles', - help='%s %s' % (h1, h2)) - - parser_list_group_roles.add_argument( - 'client', **cli_common.ORM_CLIENT_KWARGS) - parser_list_group_roles.add_argument( - 'groupid', type=str, help='') - parser_list_group_roles.add_argument( - '--region', type=str, help='region name') - parser_list_group_roles.add_argument( - '--customer', type=str, help='customer id') - parser_list_group_roles.add_argument( - '--domain', type=str, help='domain name') - - # groups - add group default users - parser_add_group_default_users = subparsers.add_parser( - 'add_group_default_users', - help='[<"X-RANGER-Client" ' - 'header>] ' - '') - parser_add_group_default_users.add_argument( - 'client', **cli_common.ORM_CLIENT_KWARGS) - parser_add_group_default_users.add_argument( - 'groupid', type=str, help='') - parser_add_group_default_users.add_argument( - 'datafile', type=argparse.FileType('r'), - help='') - - # groups - delete group default user - parser_delete_group_default_user = \ - subparsers.add_parser('delete_group_default_user', - help='[<"X-RANGER-Client" header>] ') - parser_delete_group_default_user.add_argument( - 'client', **cli_common.ORM_CLIENT_KWARGS) - parser_delete_group_default_user.add_argument( - 'groupid', type=str, help='') - parser_delete_group_default_user.add_argument( - 'userid', type=str, help='') - parser_delete_group_default_user.add_argument('userdomain', type=str, - help='') - - # groups - add_group_region_users - parser_add_group_region_user = subparsers.add_parser( - 'add_group_region_users', - help='[<"X-RANGER-Client" header>] ' - ' ') - parser_add_group_region_user.add_argument( - 'client', **cli_common.ORM_CLIENT_KWARGS) - parser_add_group_region_user.add_argument( - 'groupid', type=str, help='') - parser_add_group_region_user.add_argument( - 'regionid', type=str, help='') - parser_add_group_region_user.add_argument( - 'datafile', type=argparse.FileType('r'), - help='') - - # groups - delete group region user - parser_delete_group_region_user = subparsers.add_parser( - 'delete_group_region_user', - help='[<"X-RANGER-Client" header>] ' - ' ') - parser_delete_group_region_user.add_argument( - 'client', **cli_common.ORM_CLIENT_KWARGS) - parser_delete_group_region_user.add_argument('groupid', type=str, - help='') - parser_delete_group_region_user.add_argument('regionid', type=str, - help='') - parser_delete_group_region_user.add_argument('userid', type=str, - help='') - parser_delete_group_region_user.add_argument('userdomain', type=str, - help='') - return parser @@ -535,77 +343,6 @@ def cmd_details(args): param += '%smetadata=%s' % (preparm(param), meta[0]) return requests.get, 'customers/%s' % param - # following are groups CLIs - elif args.subcmd == 'create_group': - return requests.post, 'groups/' - elif args.subcmd == 'delete_group': - return requests.delete, 'groups/%s' % args.groupid - elif args.subcmd == 'add_groups_region': - return requests.post, 'groups/%s/regions' % args.groupid - elif args.subcmd == 'delete_groups_region': - return requests.delete, 'groups/%s/regions/%s/%s' % ( - args.groupid, - args.regionid, - args.force_delete) - elif args.subcmd == 'get_group': - return requests.get, 'groups/%s' % args.groupid - elif args.subcmd == 'list_groups': - param = '' - if args.region: - param += '%sregion=%s' % (preparm(param), args.region) - if args.starts_with: - param += '%sstarts_with=%s' % (preparm(param), args.starts_with) - if args.contains: - param += '%scontains=%s' % (preparm(param), args.contains) - return requests.get, 'groups/%s' % param - elif args.subcmd == 'assign_group_roles': - return requests.post, 'groups/%s/roles' % args.groupid - elif args.subcmd == 'assign_group_region_roles': - return requests.post, 'groups/%s/regions/%s/roles' % ( - args.groupid, args.regionid) - elif args.subcmd == 'unassign_group_role': - if args.customer and args.domain: - print("--customer and --domain cannot be specified " - "at the same time") - exit(1) - assignment_type = assignment_value = "" - if args.customer: - assignment_type = "customer" - assignment_value = args.customer - elif args.domain: - assignment_type = "domain" - assignment_value = args.domain - return requests.delete, 'groups/%s/roles/%s/%s/%s' % ( - args.groupid, - args.role, - assignment_type, - assignment_value) - elif args.subcmd == 'list_group_roles': - param = '' - if args.region: - param += '%sregion=%s' % (preparm(param), args.region) - if args.customer: - param += '%scustomer=%s' % (preparm(param), args.customer) - if args.domain: - param += '%sdomain=%s' % (preparm(param), args.domain) - return requests.get, 'groups/%s/roles/%s' % (args.groupid, param) - elif args.subcmd == 'add_group_default_users': - return requests.post, 'groups/%s/users' % args.groupid - elif args.subcmd == 'delete_group_default_user': - return requests.delete, 'groups/%s/users/%s/%s' % ( - args.groupid, - args.userid, - args.userdomain) - elif args.subcmd == 'add_group_region_users': - return requests.post, 'groups/%s/regions/%s/users' % ( - args.groupid, args.regionid) - elif args.subcmd == 'delete_group_region_user': - return requests.delete, 'groups/%s/regions/%s/users/%s/%s' % ( - args.groupid, - args.regionid, - args.userid, - args.userdomain) - def validate_args(args): for argument in ('tenant_name', 'username', 'password', diff --git a/orm/services/customer_manager/cms_rest/controllers/v1/orm/root.py b/orm/services/customer_manager/cms_rest/controllers/v1/orm/root.py index f1fe0ab9..af6e0f74 100755 --- a/orm/services/customer_manager/cms_rest/controllers/v1/orm/root.py +++ b/orm/services/customer_manager/cms_rest/controllers/v1/orm/root.py @@ -2,7 +2,6 @@ from ..orm.configuration import ConfigurationController from ..orm.customer.root import CustomerController -from ..orm.group.root import GroupController from ..orm.logs import LogsController from pecan.rest import RestController @@ -10,5 +9,4 @@ from pecan.rest import RestController class OrmController(RestController): configuration = ConfigurationController() customers = CustomerController() - groups = GroupController() logs = LogsController() diff --git a/orm/tests/unit/cms/test_group_logic.py b/orm/tests/unit/cms/test_group_logic.py deleted file mode 100644 index 611e8a02..00000000 --- a/orm/tests/unit/cms/test_group_logic.py +++ /dev/null @@ -1,557 +0,0 @@ -from orm.services.customer_manager.cms_rest.data.sql_alchemy \ - import models as sql_models -from orm.services.customer_manager.cms_rest.logic.error_base \ - import ErrorStatus -from orm.services.customer_manager.cms_rest.logic import group_logic -import orm.services.customer_manager.cms_rest.model.GroupModels as models -from orm.tests.unit.cms import FunctionalTest - -import mock - -group = None -data_manager_mock = None -record_mock = None -mock_returns_error = False -flow_type = 0 - - -STATUS_JSON = { - "regions": [ - { - "status": "Success", - "region": "GRP1", - "error_code": "", - "error_msg": "" - } - ], - "status": "Success" -} - - -class RdsStatus(object): - - def __init__(self, status_code=200, status="Success", oy=False): - self.status_code = status_code - self.status = status - self.oy = oy - - def json(self): - if self.oy: - return {} - else: - return {"status": self.status} - - -class TestGroupLogic(FunctionalTest): - def setUp(self): - global group - - FunctionalTest.setUp(self) - group_logic.DataManager = get_mock_datamanager - group_logic.pecan = mock.MagicMock() - group_logic.utils = mock.MagicMock() - group_logic.utils.make_transid.return_value = 'some_trans_id' - group_logic.utils.audit_trail.return_value = None - group_logic.utils.make_uuid.return_value = 'some_uuid' - group_logic.utils.get_time_human.return_value = '111' - - group_logic.RdsProxy = mock.MagicMock() - group_logic.RdsProxy.send_group_dict.return_value = None - group_logic.RdsProxy.get_status.return_value = RdsStatus() - - group_logic.build_response = mock.MagicMock() - - group = models.Group() - - global flow_type - flow_type = 0 - - def tearDown(self): - global mock_returns_error - FunctionalTest.tearDown(self) - mock_returns_error = False - - def test_create_group_success_with_regions(self): - group.regions = [models.Region(name="a")] - group.name = 'Group Name' - logic = group_logic.GroupLogic() - logic.create_group(group, 'some_uuid', 'some_trans_id') - - assert data_manager_mock.commit.called - assert not data_manager_mock.rollback.called - - def test_delete_region_success(self): - logic = group_logic.GroupLogic() - logic.delete_region('group_id', 'region_id', 'transaction_is', False) - - assert record_mock.delete_region_for_group.called - assert data_manager_mock.rollback.called - - def test_delete_region_success_force_delete(self): - logic = group_logic.GroupLogic() - logic.delete_region('group_id', 'region_id', 'transaction_is', True) - - assert record_mock.delete_region_for_group.called - assert data_manager_mock.commit.called - - def test_delete_region_error(self): - global mock_returns_error - mock_returns_error = True - - logic = group_logic.GroupLogic() - - self.assertRaises(SystemError, logic.delete_region, 'group_id', - 'region_id', 'transaction_is', False) - assert data_manager_mock.rollback.called - - def test_get_group_list_by_criteria(self): - logic = group_logic.GroupLogic() - logic.get_group_list_by_criteria(None, None, None, None) - self.assertTrue(data_manager_mock.get_record.called) - self.assertTrue(record_mock.get_groups_by_criteria.called) - - def test_get_group_success(self): - logic = group_logic.GroupLogic() - get_mock = mock.MagicMock() - get_mock.json.return_value = STATUS_JSON - group_logic.requests.get = mock.MagicMock(return_value=get_mock) - logic.get_group('group_id') - self.assertTrue(data_manager_mock.get_group_by_uuid_or_name.called) - - def test_get_group_not_found(self): - global flow_type - flow_type = 1 - logic = group_logic.GroupLogic() - self.assertRaises(ErrorStatus, logic.get_group, 'group_id') - self.assertTrue(data_manager_mock.get_group_by_uuid_or_name.called) - - def test_delete_group_by_uuid_success(self): - logic = group_logic.GroupLogic() - logic.delete_group_by_uuid('group_id') - - # Customer found in CMS DB but not found in RDS - group_logic.RdsProxy.get_status.return_value = RdsStatus( - status_code=404) - logic.delete_group_by_uuid('group_id') - - def test_delete_group_by_uuid_not_found(self): - global flow_type - # Change the flow to "customer not found in CMS DB" - flow_type = 1 - logic = group_logic.GroupLogic() - - # test that ErrorStatus exception is raised when no customer found - with self.assertRaises(group_logic.ErrorStatus): - logic.delete_group_by_uuid('group_id') - - def test_delete_group_by_uuid_errors(self): - global mock_returns_error - mock_returns_error = True - logic = group_logic.GroupLogic() - self.assertRaises(SystemError, logic.delete_group_by_uuid, 'group_id') - - # RDS returned an empty json - mock_returns_error = False - group_logic.RdsProxy.get_status.return_value = RdsStatus(oy=True) - self.assertRaises(group_logic.ErrorStatus, - logic.delete_group_by_uuid, - 'group_id') - - # RDS returned 500 - group_logic.RdsProxy.get_status.return_value = RdsStatus( - status_code=500) - self.assertRaises(group_logic.ErrorStatus, - logic.delete_group_by_uuid, - 'group_id') - - # RDS returned Error status - group_logic.RdsProxy.get_status.return_value = RdsStatus( - status='Error') - self.assertRaises(group_logic.ErrorStatus, - logic.delete_group_by_uuid, - 'group_id') - - def test_delete_group_by_uuid_conflict(self): - global flow_type - flow_type = 2 - logic = group_logic.GroupLogic() - self.assertRaises(group_logic.ErrorStatus, logic.delete_group_by_uuid, - 'group_id') - - def test_add_group_default_users_success(self): - global flow_type - user_assignment = [models.User( - id=['user'], domain='domain')] - - logic = group_logic.GroupLogic() - logic.add_group_default_users('group_uuid', - user_assignment, - 'some_trans_id') - - assert data_manager_mock.add_groups_user.called - assert data_manager_mock.get_group_by_uuid_or_name.called - assert data_manager_mock.commit.called - assert record_mock.read_group_by_uuid.called - - def test_add_group_default_users_error(self): - global mock_returns_error - mock_returns_error = True - - user_assignment = [models.User( - id=['user'], domain='domain')] - - logic = group_logic.GroupLogic() - self.assertRaises(SystemError, logic.add_group_default_users, - 'group_uuid', user_assignment, 'some_trans_id') - assert not data_manager_mock.add_groups_user.called - assert data_manager_mock.rollback.called - - def test_add_group_region_users_success(self): - - user_assignment = [models.RegionUser( - id=['user'], domain='domain')] - - logic = group_logic.GroupLogic() - logic.add_group_region_users('group_uuid', - 'some_region', - user_assignment, - 'some_trans_id') - - assert data_manager_mock.add_groups_user.called - assert data_manager_mock.get_group_by_uuid_or_name.called - assert data_manager_mock.get_region_id_by_name.called - assert data_manager_mock.commit.called - assert not data_manager_mock.rollback.called - assert record_mock.read_group_by_uuid.called - - def test_add_group_region_users_error(self): - global mock_returns_error - mock_returns_error = True - user_assignment = [models.RegionUser( - id=['user'], domain='domain')] - - logic = group_logic.GroupLogic() - self.assertRaises(SystemError, logic.add_group_region_users, - 'group_uuid', 'some_region', - user_assignment, 'some_trans_id') - - assert not data_manager_mock.add_groups_user.called - assert data_manager_mock.rollback.called - - def test_delete_group_default_user_success(self): - logic = group_logic.GroupLogic() - - logic.delete_group_default_user('group_uuid', - 'user', - 'user_domain', 'some_trans_id') - - assert data_manager_mock.get_group_by_uuid_or_name.called - assert record_mock.remove_user_from_group.called - assert data_manager_mock.commit.called - - def test_delete_group_default_user_error(self): - global mock_returns_error - mock_returns_error = True - - logic = group_logic.GroupLogic() - self.assertRaises(SystemError, logic.delete_group_default_user, - 'group_uuid', 'user', 'user_domain', - 'some_trans_id') - - assert record_mock.remove_user_from_group.called - assert data_manager_mock.rollback.called - - def test_assign_roles_to_group_on_domain_error(self): - roles_assginments = [models.RoleAssignment( - roles=['a_role'], domain='domain')] - - logic = group_logic.GroupLogic() - logic.assign_roles('group_uuid', roles_assginments, 'some_trans_id') - - assert not data_manager_mock.remove_user_from_group.called - assert data_manager_mock.rollback.called - - def test_assign_roles_to_group_on_customer_success(self): - roles_assginments = [models.RoleAssignment( - roles=['a_role'], customer='customer')] - - logic = group_logic.GroupLogic() - logic.assign_roles('group_uuid', roles_assginments, 'some_trans_id') - - assert data_manager_mock.add_groups_role_on_customer.called - assert data_manager_mock.get_customer_id_by_uuid.called - assert data_manager_mock.commit.called - assert data_manager_mock.add_role.called - assert not data_manager_mock.rollback.called - assert record_mock.read_group_by_uuid.called - assert record_mock.get_regions_for_group.called - - def test_assign_roles_to_group_region_on_customer_success(self): - roles_assginments = [models.RoleAssignment( - roles=['a_role'], customer='customer')] - - logic = group_logic.GroupLogic() - logic.assign_roles( - 'group_uuid', roles_assginments, 'some_trans_id', region='local') - - assert data_manager_mock.add_groups_role_on_customer.called - assert data_manager_mock.get_customer_id_by_uuid.called - assert data_manager_mock.commit.called - assert data_manager_mock.add_role.called - assert not data_manager_mock.rollback.called - assert record_mock.read_group_by_uuid.called - assert record_mock.get_region_by_keys.called - - def test_assign_roles_to_group_on_domain_error(self): - global mock_returns_error - mock_returns_error = True - - roles_assginments = [models.RoleAssignment( - roles=['a_role'], domain='domain')] - - logic = group_logic.GroupLogic() - self.assertRaises(SystemError, logic.assign_roles, 'group_uuid', - roles_assginments, 'some_trans_id') - - assert data_manager_mock.add_groups_role_on_domain.called - assert data_manager_mock.rollback.called - - def test_assign_roles_to_group_on_customer_error(self): - global mock_returns_error - mock_returns_error = True - - roles_assginments = [models.RoleAssignment( - roles=['a_role'], customer='customer')] - - logic = group_logic.GroupLogic() - self.assertRaises(SystemError, logic.assign_roles, 'group_uuid', - roles_assginments, 'some_trans_id') - - assert data_manager_mock.add_groups_role_on_customer.called - assert data_manager_mock.rollback.called - - def test_assign_roles_to_group_region_error(self): - global mock_returns_error - global flow_type - mock_returns_error = True - flow_type = 1 - - roles_assginments = [models.RoleAssignment( - roles=['a_role'], customer='customer')] - - logic = group_logic.GroupLogic() - self.assertRaises(SystemError, logic.assign_roles, 'group_uuid', - roles_assginments, 'some_trans_id') - - assert record_mock.get_regions_for_group.called - assert not data_manager_mock.add_groups_role_on_customer.called - assert data_manager_mock.rollback.called - - def test_assign_roles_to_group_region_region_error(self): - global mock_returns_error - global flow_type - mock_returns_error = True - flow_type = 1 - - roles_assginments = [models.RoleAssignment( - roles=['a_role'], customer='customer')] - - logic = group_logic.GroupLogic() - self.assertRaises(SystemError, logic.assign_roles, 'group_uuid', - roles_assginments, 'some_trans_id', region='local') - - assert record_mock.get_region_by_keys.called - assert not data_manager_mock.add_groups_role_on_customer.called - assert data_manager_mock.rollback.called - - def test_unassign_roles_from_group_on_domain_success(self): - logic = group_logic.GroupLogic() - logic.unassign_roles('group_uuid', 'role', 'domain', 'domain_name', - 'some_trans_id') - - assert record_mock.get_regions_for_group.called - assert record_mock.remove_domain_role_from_group.called - assert record_mock.check_groups_customer_role_exist.called - assert record_mock.check_groups_domain_role_exist.called - assert record_mock.read_group_by_uuid.called - assert record_mock.remove_role_from_group.called - assert data_manager_mock.commit.called - assert data_manager_mock.get_role_id_by_name.called - - def test_unassign_roles_from_group_on_customer_success(self): - logic = group_logic.GroupLogic() - logic.unassign_roles('group_uuid', 'role', 'customer', 'customer_id', - 'some_trans_id') - - assert record_mock.get_regions_for_group.called - assert record_mock.remove_customer_role_from_group.called - assert record_mock.check_groups_customer_role_exist.called - assert record_mock.check_groups_domain_role_exist.called - assert record_mock.read_group_by_uuid.called - assert record_mock.remove_role_from_group.called - assert data_manager_mock.commit.called - assert data_manager_mock.get_role_id_by_name.called - - def test_unassign_roles_from_group_group_role_not_removed(self): - global flow_type - flow_type = 3 - logic = group_logic.GroupLogic() - logic.unassign_roles('group_uuid', 'role', 'customer', 'customer_id', - 'some_trans_id') - - assert record_mock.get_regions_for_group.called - assert record_mock.remove_customer_role_from_group.called - assert record_mock.check_groups_customer_role_exist.called - assert not record_mock.check_groups_domain_role_exist.called - assert record_mock.read_group_by_uuid.called - assert not record_mock.remove_role_from_group.called - assert data_manager_mock.commit.called - assert data_manager_mock.get_role_id_by_name.called - - def test_unassign_roles_from_group_on_domain_error(self): - global mock_returns_error - mock_returns_error = True - - logic = group_logic.GroupLogic() - self.assertRaises(SystemError, logic.unassign_roles, 'group_uuid', - 'role', 'domain', 'domain_name', 'some_trans_id') - - assert record_mock.get_regions_for_group.called - assert record_mock.remove_domain_role_from_group.called - assert data_manager_mock.rollback.called - assert data_manager_mock.get_role_id_by_name.called - - def test_unassign_roles_from_group_on_customer_error(self): - global mock_returns_error - mock_returns_error = True - - logic = group_logic.GroupLogic() - self.assertRaises(SystemError, logic.unassign_roles, 'group_uuid', - 'role', 'customer', 'customer_id', 'some_trans_id') - - assert record_mock.get_regions_for_group.called - assert record_mock.remove_customer_role_from_group.called - assert data_manager_mock.rollback.called - assert data_manager_mock.get_role_id_by_name.called - - def test_unassign_roles_from_group_group_not_found(self): - global flow_type - flow_type = 1 - logic = group_logic.GroupLogic() - self.assertRaises(ErrorStatus, logic.unassign_roles, 'group_uuid', - 'role', 'customer', 'customer_id', 'some_trans_id') - - def test_unassign_roles_from_group_incorrect_type(self): - logic = group_logic.GroupLogic() - self.assertRaises(ErrorStatus, logic.unassign_roles, 'group_uuid', - 'role', 'wrong_type', 'customer_id', 'some_trans_id') - - def test_get_group_roles_by_criteria_on_customer(self): - logic = group_logic.GroupLogic() - logic.get_group_roles_by_criteria( - 'group_uuid', 'region', 'customer', None) - - self.assertTrue(data_manager_mock.get_customer_id_by_uuid.called) - self.assertTrue(data_manager_mock.get_record.called) - self.assertTrue(record_mock.get_customer_roles_by_criteria.called) - - def test_get_group_roles_by_criteria_on_domain(self): - logic = group_logic.GroupLogic() - logic.get_group_roles_by_criteria( - 'group_uuid', 'region', None, 'domain') - - self.assertTrue(data_manager_mock.get_record.called) - self.assertTrue(record_mock.get_domain_roles_by_criteria.called) - - def test_get_group_roles_by_criteria_missing_required_parms(self): - logic = group_logic.GroupLogic() - with self.assertRaises(ErrorStatus) as cm: - logic.get_group_roles_by_criteria('group', None, None, None) - self.assertEqual(cm.exception.status_code, 400) - self.assertIn('region must be specified', cm.exception.message) - - def test_get_group_roles_by_criteria_conflicting_optional_parms(self): - logic = group_logic.GroupLogic() - with self.assertRaises(ErrorStatus) as cm: - logic.get_group_roles_by_criteria( - 'group', 'region', 'customer', 'domain') - self.assertEqual(cm.exception.status_code, 400) - self.assertIn('customer and domain cannot be used at the same time', - cm.exception.message) - - -def get_mock_datamanager(): - global data_manager_mock - global record_mock - - sql_group = sql_models.Groups(name='a') - sql_group.group_regions = [] - - data_manager_mock = mock.MagicMock() - record_mock = mock.MagicMock() - record_mock.get_groups_by_criteria.return_value = [sql_group] - record_mock.get_customer_roles_by_criteria.return_value = [] - record_mock.get_domain_roles_by_criteria.return_value = [] - result_mock = mock.Mock() - result_mock.rowcount = 1 - - def _get_group(): - def mock_to_wsme(): - return models.Group(regions=[models.Region()]) - - sql_group = sql_models.Groups() - sql_group.to_wsme = mock_to_wsme - sql_group.uuid = '1234' - sql_group.status = 'Success' - sql_group.name = 'GRP1' - - return sql_group - - if not mock_returns_error: - data_manager_mock.get_group_by_uuid_or_name.return_value = _get_group() - data_manager_mock.get_role_id_by_name.return_value = 1 - record_mock.delete_region_for_group.return_value = None - record_mock.delete_group_by_uuid.return_value = None - - # mock for assign/unassign roles - record_mock.read_group_by_uuid.return_value = sql_group - record_mock.get_regions_for_group.return_value = [ - sql_models.GroupsRegion(region_id=1, group_id="group_id")] - record_mock.get_region_by_keys.return_value = \ - sql_models.GroupsRegion(region_id=1, group_id="group_id") - record_mock.remove_customer_role_from_group.return_value = result_mock - record_mock.remove_domain_role_from_group.return_value = result_mock - record_mock.check_groups_customer_role_exist.return_value = False - record_mock.check_groups_domain_role_exist.return_value = False - record_mock.remove_role_from_group.return_value = result_mock - - if flow_type == 1: - record_mock.read_group_by_uuid.return_value = None - data_manager_mock.get_group_by_uuid_or_name.return_value = None - elif flow_type == 2: - q = mock.MagicMock() - q.get_group_regions.return_value = [mock.MagicMock()] - record_mock.read_group_by_uuid.return_value = q - record_mock.delete_group_by_uuid.side_effect = SystemError() - elif flow_type == 3: - record_mock.check_groups_customer_role_exist.return_value = True - else: - record_mock.read_group_by_uuid.side_effect = SystemError() - record_mock.delete_region_for_group.side_effect = SystemError() - - # mock for assign roles - data_manager_mock.add_groups_role_on_domain.side_effect = \ - SystemError() - data_manager_mock.add_groups_role_on_customer.side_effect = \ - SystemError() - record_mock.get_regions_for_group.return_value = [ - sql_models.GroupsRegion(region_id=1, group_id="group_id")] - record_mock.remove_domain_role_from_group.side_effect = SystemError() - record_mock.remove_customer_role_from_group.side_effect = SystemError() - - if flow_type == 1: - record_mock.get_regions_for_group.side_effect = SystemError() - record_mock.get_region_by_keys.side_effect = SystemError() - - data_manager_mock.get_record.return_value = record_mock - return data_manager_mock diff --git a/orm/tests/unit/cms/test_group_regions.py b/orm/tests/unit/cms/test_group_regions.py deleted file mode 100755 index dd2959bf..00000000 --- a/orm/tests/unit/cms/test_group_regions.py +++ /dev/null @@ -1,170 +0,0 @@ -import mock -import requests - -from wsme.exc import ClientSideError - -from orm.services.customer_manager.cms_rest.controllers.v1.orm.group \ - import regions -from orm.services.customer_manager.cms_rest.logic.error_base import ErrorStatus -from orm.services.customer_manager.cms_rest.model import GroupModels -from orm.tests.unit.cms import FunctionalTest - -group_logic_mock = None - - -class TestRegionController(FunctionalTest): - def setUp(self): - FunctionalTest.setUp(self) - - regions.authentication = mock.MagicMock() - - regions.GroupLogic = get_mock_group_logic - - regions.GroupLogic.return_error = 0 - - regions.utils = mock.MagicMock() - regions.utils.make_transid.return_value = 'some_trans_id' - regions.utils.audit_trail.return_value = None - regions.utils.make_uuid.return_value = 'some_uuid' - - regions.err_utils = mock.MagicMock() - - def tearDown(self): - FunctionalTest.tearDown(self) - - def test_add_regions(self): - # given - requests.post = mock.MagicMock(return_value=ResponseMock(200)) - - # when - response = self.app.post_json('/v1/orm/groups/{group id}/regions/', - REGION_JSON) - - # assert - assert response.status_int == 200 - assert regions.utils.audit_trail.called - assert group_logic_mock.add_regions.called - - def test_add_regions_fail(self): - # given - requests.post = mock.MagicMock() - - regions.GroupLogic.return_error = 1 - - regions.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - - # when - response = self.app.post_json('/v1/orm/groups/{group id}/regions/', - REGION_JSON, expect_errors=True) - - # assert - self.assertEqual(response.status_int, 500) - - def test_add_regions_fail_bad(self): - # given - requests.post = mock.MagicMock() - - regions.GroupLogic.return_error = 2 - - regions.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 404)) - - # when - response = self.app.post_json('/v1/orm/groups/{group id}/regions/', - REGION_JSON, expect_errors=True) - - # assert - self.assertEqual(response.status_int, 404) - - @mock.patch.object(regions, 'request') - def test_delete_regions(self, request): - # given - requests.delete = mock.MagicMock(return_value=ResponseMock(200)) - request.headers = {'X-RANGER-Requester': "rds_resource_service_proxy"} - # when - response = self.app.delete( - '/v1/orm/groups/{group id}/regions/{region_id}') - - # assert - assert response.status_int == 204 - assert regions.utils.audit_trail.called - assert group_logic_mock.delete_region.called - - def test_delete_regions_fail_bad(self): - # given - requests.delete = mock.MagicMock() - - regions.GroupLogic.return_error = 1 - - regions.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - - # when - response = self.app.delete( - '/v1/orm/groups/{group id}/regions/{region_id}', - expect_errors=True) - - # assert - self.assertEqual(response.status_int, 500) - - @mock.patch.object(regions, 'request') - def test_delete_regions_fail(self, request): - # given - requests.delete = mock.MagicMock() - - regions.GroupLogic.return_error = 2 - request.headers = {'X-RANGER-Requester': "rds_resource_service_proxy"} - regions.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 404)) - - # when - response = self.app.delete( - '/v1/orm/groups/{group id}/regions/{region_id}', - expect_errors=True) - - # assert - self.assertEqual(response.status_int, 404) - - def test_get(self): - # given - requests.get = mock.MagicMock(return_value=ResponseMock(200)) - - # when - response = self.app.get('/v1/orm/groups/some_id/regions/some_id') - - # assert - assert response.status_int == 200 - - -def get_mock_group_logic(): - global group_logic_mock - group_logic_mock = mock.MagicMock() - - if regions.GroupLogic.return_error == 0: - res = GroupModels.RegionResultWrapper(transaction_id='1', regions=[]) - group_logic_mock.add_regions.return_value = res - - elif regions.CustomerLogic.return_error == 1: - group_logic_mock.add_regions.side_effect = SystemError() - group_logic_mock.delete_region.side_effect = SystemError() - - else: - group_logic_mock.add_regions.side_effect = ErrorStatus(status_code=404) - group_logic_mock.delete_region.side_effect = ErrorStatus( - status_code=404) - - return group_logic_mock - - -class ResponseMock: - def __init__(self, status_code=200): - self.status_code = status_code - - -REGION_JSON = [ - { - "name": "SAN1", - "type": "single", - } -] diff --git a/orm/tests/unit/cms/test_groups.py b/orm/tests/unit/cms/test_groups.py deleted file mode 100644 index ea3fa07e..00000000 --- a/orm/tests/unit/cms/test_groups.py +++ /dev/null @@ -1,217 +0,0 @@ -import mock -import requests - -from orm.services.customer_manager.cms_rest.controllers.v1.orm.group\ - import root -from orm.services.customer_manager.cms_rest.logic.error_base import ErrorStatus -from orm.services.customer_manager.cms_rest.model import GroupModels -from orm.tests.unit.cms import FunctionalTest, test_utils -from wsme.exc import ClientSideError - -group_logic_mock = None - - -class TestGroupController(FunctionalTest): - def setUp(self): - FunctionalTest.setUp(self) - - root.authentication = mock.MagicMock() - - root.GroupLogic = get_mock_group_logic - root.GroupLogic.return_error = 0 - - root.utils = mock.MagicMock() - root.utils.make_transid.return_value = 'some_trans_id' - root.utils.audit_trail.return_value = None - root.utils.make_uuid.return_value = 'some_uuid' - - root.err_utils = mock.MagicMock() - - def tearDown(self): - FunctionalTest.tearDown(self) - - def test_create_group(self): - # given - requests.post = mock.MagicMock(return_value=ResponseMock(201)) - - # when - response = self.app.post_json('/v1/orm/groups', GROUP_JSON) - - # assert - assert response.status_int == 201 - assert root.utils.audit_trail.called - assert root.utils.create_or_validate_uuid.called - assert group_logic_mock.create_group_called - - def test_create_group_fail(self): - # given - requests.post = mock.MagicMock() - - root.GroupLogic.return_error = 1 - - root.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - # when - response = self.app.post_json('/v1/orm/groups', - GROUP_JSON, expect_errors=True) - # assert - self.assertEqual(response.status_int, 500) - - def test_get_group(self): - # given - requests.get = mock.MagicMock(return_value=ResponseMock(200)) - - # when - response = self.app.get('/v1/orm/groups/some_id') - - # assert - assert response.status_int == 200 - assert group_logic_mock.get_group.called - - def test_get_group_fail_bad_request(self): - # given - requests.put = mock.MagicMock() - root.GroupLogic.return_error = 1 - root.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - - # when - response = self.app.get('/v1/orm/groups/some_id', expect_errors=True) - - # assert - self.assertEqual(response.status_int, 500) - assert group_logic_mock.get_group.called - - def test_get_group_fail(self): - # given - requests.put = mock.MagicMock() - root.GroupLogic.return_error = 2 - root.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 404)) - - # when - response = self.app.get('/v1/orm/groups/some_id', expect_errors=True) - - # assert - self.assertEqual(response.status_int, 404) - assert group_logic_mock.get_group.called - - def test_get_list_group(self): - # given - requests.get = mock.MagicMock(return_value=ResponseMock(200)) - - # when - response = self.app.get('/v1/orm/groups?region=region') - - # assert - assert group_logic_mock.get_group_list_by_criteria.called - - def test_get_list_group_fail(self): - # given - requests.get = mock.MagicMock() - root.GroupLogic.return_error = 1 - root.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - - # when - response = self.app.get('/v1/orm/groups?region=region', - expect_errors=True) - - # assert - self.assertEqual(response.status_int, 500) - - def test_get_list_group_bad_request(self): - # given - requests.get = mock.MagicMock() - root.GroupLogic.return_error = 2 - root.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - - # when - response = self.app.get('/v1/orm/groups?region=region', - expect_errors=True) - - # assert - self.assertEqual(response.status_int, 500) - - @mock.patch.object(root, 'authentication') - def test_delete_group_success(self, mock_auth): - response = self.app.delete('/v1/orm/groups/test') - self.assertEqual(response.status_int, 204) - - @mock.patch.object(root, 'authentication') - def test_delete_group_conflict(self, mock_auth): - root.GroupLogic.return_error = 2 - root.err_utils.get_error = test_utils.get_error - response = self.app.delete('/v1/orm/groups/test', expect_errors=True) - - self.assertEqual(response.status_int, 409) - - @mock.patch.object(root, 'authentication') - def test_delete_group_error(self, mock_auth): - root.GroupLogic.return_error = 1 - root.err_utils.get_error = test_utils.get_error - response = self.app.delete('/v1/orm/groups/test', expect_errors=True) - - self.assertEqual(response.status_int, 500) - - -def get_mock_group_logic(): - global group_logic_mock - group_logic_mock = mock.MagicMock() - - if root.GroupLogic.return_error == 0: - res = GroupModels.GroupResultWrapper(transaction_id='1', - id='1', - links={}, - updated=None, - created='1') - list_res = GroupModels.GroupSummaryResponse() - list_res.groups.append( - GroupModels.GroupSummary(name='1', id='1', description='1')) - group_logic_mock.get_group.return_value = GroupModels.Group( - **RET_GROUP_JSON) - group_logic_mock.get_group_list_by_criteria.return_value = list_res - group_logic_mock.create_group.return_value = res - - elif root.GroupLogic.return_error == 1: - group_logic_mock.create_group.side_effect = SystemError() - group_logic_mock.get_group.side_effect = SystemError() - group_logic_mock.get_group_list_by_criteria.side_effect = SystemError() - group_logic_mock.delete_group_by_uuid.side_effect = SystemError() - - else: - group_logic_mock.get_group.side_effect = ErrorStatus(status_code=404) - group_logic_mock.get_group_list_by_criteria.side_effect = ErrorStatus( - status_code=404) - group_logic_mock.delete_group_by_uuid.side_effect = ErrorStatus( - status_code=409) - - return group_logic_mock - - -class ResponseMock: - def __init__(self, status_code=200): - self.status_code = status_code - - -GROUP_JSON = { - "description": "Group description", - "enabled": True, - "name": "myGroup", - "domain": "default", - "regions": [ - { - "name": "SAN1", - "type": "single" - } - ] -} - -RET_GROUP_JSON = { - "description": "Group description", - "name": "myName", - "domain": "default", - "enabled": True, - "regions": [GroupModels.Region()] -} diff --git a/orm/tests/unit/cms/test_groups_region_role.py b/orm/tests/unit/cms/test_groups_region_role.py deleted file mode 100644 index b5a48eb0..00000000 --- a/orm/tests/unit/cms/test_groups_region_role.py +++ /dev/null @@ -1,110 +0,0 @@ -import mock -import requests - -from orm.services.customer_manager.cms_rest.controllers.v1.orm.group\ - import region_roles -from orm.services.customer_manager.cms_rest.logic.error_base import ErrorStatus -from orm.services.customer_manager.cms_rest.model import GroupModels -from orm.tests.unit.cms import FunctionalTest -from wsme.exc import ClientSideError - -group_logic_mock = None - - -class TestGroupsRegionRoleController(FunctionalTest): - def setUp(self): - FunctionalTest.setUp(self) - - region_roles.authentication = mock.MagicMock() - region_roles.GroupLogic = get_mock_group_logic - - region_roles.GroupLogic.return_error = 0 - region_roles.utils = mock.MagicMock() - region_roles.utils.audit_trail.return_value = None - region_roles.err_utils = mock.MagicMock() - - def tearDown(self): - FunctionalTest.tearDown(self) - - def test_assign_roles_to_group_region(self): - # given - requests.post = mock.MagicMock(return_value=ResponseMock(200)) - - # when - response = self.app.post_json( - '/v1/orm/groups/{groups id}/regions/{region id}/roles/', - GROUPS_ROLE_JSON) - - # assert - assert response.status_int == 200 - assert region_roles.utils.audit_trail.called - assert group_logic_mock.assign_roles_called - - def test_assign_roles_to_group_region_fail(self): - # given - requests.post = mock.MagicMock() - - region_roles.GroupLogic.return_error = 1 - - region_roles.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - # when - response = self.app.post_json( - '/v1/orm/groups/{groups id}/regions/{region id}/roles/', - GROUPS_ROLE_JSON, expect_errors=True) - # assert - self.assertEqual(response.status_int, 500) - - def test_assign_roles_to_group_region_bad_request(self): - # given - requests.post = mock.MagicMock() - - region_roles.GroupLogic.return_error = 2 - region_roles.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 404)) - - # when - response = self.app.post_json( - '/v1/orm/groups/{groups_id}/regions/{region id}/roles/', - GROUPS_ROLE_JSON, expect_errors=True) - - # assert - self.assertEqual(response.status_int, 404) - - -def get_mock_group_logic(): - global group_logic_mock - group_logic_mock = mock.MagicMock() - - if region_roles.GroupLogic.return_error == 0: - res = GroupModels.RoleResultWrapper(transaction_id='1', - roles=[], - links={}, - created='1') - - group_logic_mock.assign_roles.return_value = res - - elif region_roles.GroupLogic.return_error == 1: - group_logic_mock.assign_roles.side_effect = SystemError() - - elif region_roles.GroupLogic.return_error == 2: - group_logic_mock.assign_roles.side_effect = ErrorStatus( - status_code=404) - - return group_logic_mock - - -class ResponseMock: - def __init__(self, status_code=200): - self.status_code = status_code - - -GROUPS_ROLE_JSON = [ - { - "customer": "customer-id", - "roles": [ - "role1", - "role2" - ] - } -] diff --git a/orm/tests/unit/cms/test_groups_region_users.py b/orm/tests/unit/cms/test_groups_region_users.py deleted file mode 100644 index e7ff5c41..00000000 --- a/orm/tests/unit/cms/test_groups_region_users.py +++ /dev/null @@ -1,156 +0,0 @@ -import mock -import requests -from wsme.exc import ClientSideError - -from orm.services.customer_manager.cms_rest.controllers.v1.orm.group \ - import region_users -from orm.services.customer_manager.cms_rest.logic.error_base import ErrorStatus -from orm.services.customer_manager.cms_rest.model import GroupModels -from orm.tests.unit.cms import FunctionalTest - -group_logic_mock = None - - -class TestGroupsRgnUserController(FunctionalTest): - def setUp(self): - FunctionalTest.setUp(self) - - region_users.authentication = mock.MagicMock() - - region_users.GroupLogic = get_mock_group_logic - region_users.GroupLogic.return_error = 0 - - region_users.utils = mock.MagicMock() - region_users.utils.audit_trail.return_value = None - - region_users.err_utils = mock.MagicMock() - - def tearDown(self): - FunctionalTest.tearDown(self) - - def test_add_region_users(self): - # given - requests.post = mock.MagicMock(return_value=ResponseMock(200)) - - # when - response = self.app.post_json( - '/v1/orm/groups/{group id}/regions/{region_id}/users/', - GROUPS_USER_JSON) - - # assert - self.assertEqual(response.status_int, 200) - self.assertTrue(group_logic_mock.add_group_region_users.called) - - def test_add_region_users_fail(self): - # given - requests.post = mock.MagicMock() - - region_users.GroupLogic.return_error = 1 - region_users.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - # when - response = self.app.post_json( - '/v1/orm/groups/{group id}/regions/{region_id}/users/', - GROUPS_USER_JSON, - expect_errors=True) - # assert - self.assertEqual(response.status_int, 500) - - def test_add_region_users_fail_bad_request(self): - # given - requests.post = mock.MagicMock() - - region_users.GroupLogic.return_error = 2 - region_users.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 404)) - # when - response = self.app.post_json( - '/v1/orm/groups/{group id}/regions/{region_id}/users/', - GROUPS_USER_JSON, - expect_errors=True) - - # assert - self.assertEqual(response.status_int, 404) - - def test_delete_region_user(self): - # given - requests.delete = mock.MagicMock(return_value=ResponseMock(200)) - - # when - response = self.app.delete( - '/v1/orm/groups/{group id}/regions/{region_id}/users/' - '{user_id}/{domain}') - - # assert - self.assertEqual(response.status_int, 204) - self.assertTrue(group_logic_mock.delete_group_region_user.called) - - def test_delete_region_user_fail(self): - # given - requests.delete = mock.MagicMock() - - region_users.GroupLogic.return_error = 1 - region_users.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - - # when - response = self.app.delete( - '/v1/orm/groups/{group id}/regions/{region_id}/users/' - '{user_id}/{domain}', expect_errors=True) - - # assert - self.assertEqual(response.status_int, 500) - - def test_delete_region_user_fail_bad_request(self): - # given - requests.delete = mock.MagicMock() - - region_users.GroupLogic.return_error = 2 - region_users.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 404)) - - # when - response = self.app.delete( - '/v1/orm/groups/{group id}/regions/{region_id}/users/' - '{user_id}/{domain}', expect_errors=True) - - # assert - self.assertEqual(response.status_int, 404) - - -def get_mock_group_logic(): - global group_logic_mock - group_logic_mock = mock.MagicMock() - - if region_users.GroupLogic.return_error == 0: - res = GroupModels.RegionUserResultWrapper(transaction_id='1', - users=[], - links={}, - created='1') - - group_logic_mock.add_group_region_users.return_value = res - - elif region_users.GroupLogic.return_error == 1: - group_logic_mock.add_group_region_users.side_effect = SystemError() - group_logic_mock.delete_group_region_user.side_effect = SystemError() - - else: - group_logic_mock.add_group_region_users.side_effect = ErrorStatus( - status_code=404) - group_logic_mock.delete_group_region_user.side_effect = ErrorStatus( - status_code=404) - - return group_logic_mock - - -class ResponseMock: - def __init__(self, status_code=200): - self.status_code = status_code - - -GROUPS_USER_JSON = [ - { - "id": ["attuser1"], - "domain": "nc" - } -] diff --git a/orm/tests/unit/cms/test_groups_role.py b/orm/tests/unit/cms/test_groups_role.py deleted file mode 100644 index 33a5f4fb..00000000 --- a/orm/tests/unit/cms/test_groups_role.py +++ /dev/null @@ -1,216 +0,0 @@ -import mock -import requests - -from orm.services.customer_manager.cms_rest.controllers.v1.orm.group\ - import roles -from orm.services.customer_manager.cms_rest.logic.error_base import ErrorStatus -from orm.services.customer_manager.cms_rest.model import GroupModels -from orm.tests.unit.cms import FunctionalTest -from wsme.exc import ClientSideError - -group_logic_mock = None - - -class TestGroupsRoleController(FunctionalTest): - def setUp(self): - FunctionalTest.setUp(self) - - roles.authentication = mock.MagicMock() - - roles.GroupLogic = get_mock_group_logic - roles.GroupLogic.return_error = 0 - - roles.utils = mock.MagicMock() - roles.utils.make_transid.return_value = 'some_trans_id' - roles.utils.audit_trail.return_value = None - roles.utils.make_uuid.return_value = 'some_uuid' - - roles.err_utils = mock.MagicMock() - - def tearDown(self): - FunctionalTest.tearDown(self) - - def test_assign_roles_to_group(self): - # given - requests.post = mock.MagicMock(return_value=ResponseMock(200)) - - # when - response = self.app.post_json('/v1/orm/groups/{groups id}/roles/', - GROUPS_ROLE_JSON) - - # assert - assert response.status_int == 200 - assert roles.utils.audit_trail.called - assert group_logic_mock.assign_roles_called - - def test_assign_roles_to_group_fail(self): - # given - requests.post = mock.MagicMock() - - roles.GroupLogic.return_error = 1 - - roles.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - # when - response = self.app.post_json('/v1/orm/groups/{groups id}/roles/', - GROUPS_ROLE_JSON, expect_errors=True) - # assert - self.assertEqual(response.status_int, 500) - - def test_assign_roles_to_group_bad_request(self): - # given - requests.post = mock.MagicMock() - - roles.GroupLogic.return_error = 2 - roles.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 404)) - - # when - response = self.app.post_json('/v1/orm/groups/{groups_id}/roles/', - GROUPS_ROLE_JSON, expect_errors=True) - - # assert - self.assertEqual(response.status_int, 404) - - @mock.patch.object(roles, 'request') - def test_unassign_roles_to_group(self, request): - # given - requests.delete = mock.MagicMock(return_value=ResponseMock(204)) - request.headers = {'X-RANGER-Requester': "rds_resource_service_proxy"} - - response = self.app.delete( - '/v1/orm/groups/{groups id}/roles/{role name}/{type}/{type id}') - - # assert - assert response.status_int == 204 - assert roles.utils.audit_trail.called - assert group_logic_mock.unassign_roles_called - - def test_unassign_roles_to_group_fail(self): - # given - requests.delete = mock.MagicMock() - - roles.GroupLogic.return_error = 1 - roles.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - - # when - response = self.app.delete( - '/v1/orm/groups/{groups id}/roles/{role name}/{type}/{type id}', - expect_errors=True) - - # assert - self.assertEqual(response.status_int, 500) - - @mock.patch.object(roles, 'request') - def test_unassign_roles_to_group_bad_request(self, request): - # given - requests.delete = mock.MagicMock() - request.headers = {'X-RANGER-Requester': "rds_resource_service_proxy"} - - roles.GroupLogic.return_error = 2 - roles.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 404)) - - # when - response = self.app.delete( - '/v1/orm/groups/{groups id}/roles/{role name}/{type}/{type id}', - expect_errors=True) - - # assert - self.assertEqual(response.status_int, 404) - - def test_list_group_roles(self): - # given - requests.get = mock.MagicMock(return_value=ResponseMock(200)) - - # when - response = self.app.get( - '/v1/orm/groups/{groups id}/roles/?region=region') - - # assert - assert group_logic_mock.get_group_roles_by_criteria.called - - def test_list_group_roles_fail(self): - # given - requests.get = mock.MagicMock() - roles.GroupLogic.return_error = 1 - roles.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - - # when - response = self.app.get( - '/v1/orm/groups/{groups id}/roles/?region=region', - expect_errors=True) - - # assert - self.assertEqual(response.status_int, 500) - - def test_list_group_roles_bad_request(self): - # given - requests.get = mock.MagicMock() - roles.GroupLogic.return_error = 2 - roles.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 404)) - - # when - response = self.app.get( - '/v1/orm/groups/{groups id}/roles/?region=region', - expect_errors=True) - - # assert - self.assertEqual(response.status_int, 404) - - -def get_mock_group_logic(): - global group_logic_mock - group_logic_mock = mock.MagicMock() - - if roles.GroupLogic.return_error == 0: - res = GroupModels.RoleResultWrapper(transaction_id='1', - roles=[], - links={}, - created='1') - - res1 = GroupModels.RoleResultWrapper(transaction_id='1', - roles=[], - links={}, - created='1') - - list_res = GroupModels.RoleResult(roles=[]) - - group_logic_mock.assign_roles.return_value = res - group_logic_mock.unassign_roles.return_value = res1 - group_logic_mock.get_group_roles_by_criteria.return_value = [list_res] - - elif roles.GroupLogic.return_error == 1: - group_logic_mock.assign_roles.side_effect = SystemError() - group_logic_mock.unassign_roles.side_effect = SystemError() - group_logic_mock.get_group_roles_by_criteria.side_effect = \ - SystemError() - - else: - group_logic_mock.assign_roles.side_effect = ErrorStatus( - status_code=404) - group_logic_mock.unassign_roles.side_effect = ErrorStatus( - status_code=404) - group_logic_mock.get_group_roles_by_criteria.side_effect = ErrorStatus( - status_code=404) - - return group_logic_mock - - -class ResponseMock: - def __init__(self, status_code=200): - self.status_code = status_code - - -GROUPS_ROLE_JSON = [ - { - "customer": "customer-id", - "roles": [ - "role1", - "role2" - ] - } -] diff --git a/orm/tests/unit/cms/test_groups_users.py b/orm/tests/unit/cms/test_groups_users.py deleted file mode 100644 index 62cbaa54..00000000 --- a/orm/tests/unit/cms/test_groups_users.py +++ /dev/null @@ -1,151 +0,0 @@ -import mock -import requests -from wsme.exc import ClientSideError - -from orm.services.customer_manager.cms_rest.controllers.v1.orm.group \ - import users -from orm.services.customer_manager.cms_rest.logic.error_base import ErrorStatus -from orm.services.customer_manager.cms_rest.model import GroupModels -from orm.tests.unit.cms import FunctionalTest - -group_logic_mock = None - - -class TestGroupsUserController(FunctionalTest): - def setUp(self): - FunctionalTest.setUp(self) - - users.authentication = mock.MagicMock() - - users.GroupLogic = get_mock_group_logic - users.GroupLogic.return_error = 0 - - users.utils = mock.MagicMock() - users.utils.audit_trail.return_value = None - - users.err_utils = mock.MagicMock() - - def tearDown(self): - FunctionalTest.tearDown(self) - - def test_add_default_users(self): - # given - requests.post = mock.MagicMock(return_value=ResponseMock(200)) - - # when - response = self.app.post_json('/v1/orm/groups/{group id}/users/', - GROUPS_USER_JSON) - - # assert - self.assertEqual(response.status_int, 200) - self.assertTrue(group_logic_mock.add_group_default_users.called) - - def test_add_default_users_fail(self): - # given - requests.post = mock.MagicMock() - - users.GroupLogic.return_error = 1 - users.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - # when - response = self.app.post_json('/v1/orm/groups/{group id}/users/', - GROUPS_USER_JSON, - expect_errors=True) - # assert - self.assertEqual(response.status_int, 500) - - def test_add_default_users_fail_bad_request(self): - # given - requests.post = mock.MagicMock() - - users.GroupLogic.return_error = 2 - users.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 404)) - # when - response = self.app.post_json('/v1/orm/groups/{group id}/users/', - GROUPS_USER_JSON, - expect_errors=True) - - # assert - self.assertEqual(response.status_int, 404) - - def test_delete_default_user(self): - # given - requests.delete = mock.MagicMock(return_value=ResponseMock(200)) - - # when - response = self.app.delete('/v1/orm/groups/{group id}/users/{user_id}/{domain}') - - # assert - self.assertEqual(response.status_int, 204) - # uncomment below line when delete_default_user is implemented - # self.assertTrue(users.utils.audit_trail.called) - self.assertTrue(group_logic_mock.delete_group_default_user.called) - - def test_delete_default_user_fail(self): - # given - requests.delete = mock.MagicMock() - - users.GroupLogic.return_error = 1 - users.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 500)) - - # when - response = self.app.delete('/v1/orm/groups/{group id}/users/{user_id}/{domain}', - expect_errors=True) - - # assert - self.assertEqual(response.status_int, 500) - - def test_delete_default_user_fail_bad_request(self): - # given - requests.delete = mock.MagicMock() - - users.GroupLogic.return_error = 2 - users.err_utils.get_error = mock.MagicMock( - return_value=ClientSideError("blabla", 404)) - - # when - response = self.app.delete('/v1/orm/groups/{group id}/users/{user_id}/{domain}', - expect_errors=True) - - # assert - self.assertEqual(response.status_int, 404) - - -def get_mock_group_logic(): - global group_logic_mock - group_logic_mock = mock.MagicMock() - - if users.GroupLogic.return_error == 0: - res = GroupModels.UserResultWrapper(transaction_id='1', - users=[], - links={}, - created='1') - - group_logic_mock.add_group_default_users.return_value = res - - elif users.GroupLogic.return_error == 1: - group_logic_mock.add_group_default_users.side_effect = SystemError() - group_logic_mock.delete_group_default_user.side_effect = SystemError() - - else: - group_logic_mock.add_group_default_users.side_effect = ErrorStatus( - status_code=404) - group_logic_mock.delete_group_default_user.side_effect = ErrorStatus( - status_code=404) - - return group_logic_mock - - -class ResponseMock: - def __init__(self, status_code=200): - self.status_code = status_code - - -GROUPS_USER_JSON = [ - { - "id": ["attuser1"], - "domain": "nc" - } -] diff --git a/orm/tests/unit/ormcli/test_cmscli.py b/orm/tests/unit/ormcli/test_cmscli.py index cb67a9b7..323e8231 100755 --- a/orm/tests/unit/ormcli/test_cmscli.py +++ b/orm/tests/unit/ormcli/test_cmscli.py @@ -33,7 +33,6 @@ class CmsTests(TestCase): # Set up the args parameter args = mock.MagicMock() args.custid = 'test_custid' - args.groupid = 'test_groupid' args.regionid = 'test_region' args.userid = 'test_userid' args.userdomain = 'test_userdomain' @@ -42,10 +41,6 @@ class CmsTests(TestCase): args.starts_with = 'test_startswith' args.contains = 'test_contains' args.force_delete is False - args.role = 'test_role_name' - args.assignment_value = 'test_role_assignment_value' - args.customer = 'test_customer' - args.domain = None subcmd_to_result = { 'create_customer': (requests.post, 'customers/',), @@ -86,43 +81,7 @@ class CmsTests(TestCase): 'customers/?region=%s&user=%s&starts_with=%s' '&contains=%s' % (args.region, args.user, args.starts_with, - args.contains)), - 'delete_group': ( - requests.delete, 'groups/%s' % args.groupid,), - 'delete_groups_region': ( - requests.delete, - 'groups/%s/regions/%s/%s' % (args.groupid, args.regionid, - args.force_delete),), - 'get_group': (requests.get, 'groups/%s' % args.groupid,), - 'list_groups': (requests.get, - 'groups/?region=%s&starts_with=%s' - '&contains=%s' % (args.region, - args.starts_with, - args.contains)), - 'list_group_roles': ( - requests.get, 'groups/%s/roles/?region=%s&customer=%s' % ( - args.groupid, args.region, args.customer)), - 'add_group_default_users': ( - requests.post, 'groups/%s/users' % args.groupid,), - 'delete_group_default_user': ( - requests.delete, 'groups/%s/users/%s/%s' % ( - args.groupid, args.userid, args.userdomain),), - 'add_group_region_users': ( - requests.post, - 'groups/%s/regions/%s/users' % (args.groupid, - args.regionid,)), - 'delete_group_region_user': ( - requests.delete, - 'groups/%s/regions/%s/users/%s/%s' % (args.groupid, - args.regionid, - args.userid, - args.userdomain,)), - - 'assign_group_roles': ( - requests.post, 'groups/%s/roles' % args.groupid,), - 'assign_group_region_roles': ( - requests.post, 'groups/%s/regions/%s/roles' % (args.groupid, - args.regionid)) + args.contains)) } # Assert that each subcommand returns the expected details @@ -131,24 +90,6 @@ class CmsTests(TestCase): self.assertEqual(subcmd_to_result[subcmd], cmscli.cmd_details(args)) - args.subcmd = 'unassign_group_role' - for assignment_type in ['customer', 'domain']: - if assignment_type == 'customer': - args.customer = args.assignment_value - args.domain = None - else: - args.domain = args.assignment_value - args.customer = None - cmd_to_result = (requests.delete, - 'groups/%s/roles/%s/%s/%s' % ( - args.groupid, - args.role, - assignment_type, - args.assignment_value)) - - self.assertEqual(cmd_to_result, - cmscli.cmd_details(args)) - @mock.patch.object(cmscli, 'validate_args') @mock.patch.object(cmscli.requests, 'post') @mock.patch.object(cmscli.requests, 'get') @@ -200,71 +141,3 @@ class CmsTests(TestCase): resp = self.respond('{"Hi, mom"}', 200, {'X-Subject-Token': 989}) mock_post.return_value = self.respond( {"access": {"token": {"id": 989}}}, 200) - - @mock.patch.object(cmscli, 'validate_args') - @mock.patch.object(cmscli.requests, 'post') - @mock.patch.object(cmscli.requests, 'get') - @mock.patch.object(cmscli.cli_common, 'get_token') - @mock.patch.object(cmscli, 'globals') - def test_list_groups(self, mock_globals, mock_get_token, - mock_get, mock_post, mock_validate_args): - mock_post.return_value = self.respond(TJ, 200) - mock_get.return_value = self.mock_response - args = ormcli.main('orm cms list_groups t'.split()) - sys.stdout.seek(0) - output = sys.stdout.read() - self.assertIn(json.dumps(TJ), output) - - @mock.patch.object(cmscli, 'validate_args') - @mock.patch.object(cmscli.requests, 'post') - @mock.patch.object(cmscli.requests, 'get') - @mock.patch.object(cmscli.cli_common, 'get_token') - @mock.patch.object(cmscli, 'globals') - def test_list_groups_a(self, mock_globals, mock_get_token, - mock_get, mock_post, mock_validate_args): - mock_post.return_value = self.respond(TJ, 200) - mock_get.return_value = self.mock_response - mock_get.__name__ = 'a' - args = ormcli.main('orm cms --verbose list_groups t'.split()) - sys.stdout.seek(0) - output = sys.stdout.read() - self.assertIn(json.dumps(TJ), output) - - @mock.patch.object(cmscli, 'validate_args') - @mock.patch.object(cmscli.requests, 'post') - @mock.patch.object(cmscli.requests, 'get') - def test_list_groups_e(self, mock_get, mock_post, mock_validate_args): - mock_post.return_value = self.respond(TJ, 200) - mock_get.side_effect = Exception('e') - with self.assertRaises(SystemExit) as cm: - args = ormcli.main('orm cms list_groups t'.split()) - self.assertEqual(cm.exception.code, 1) - sys.stdout.seek(0) - output = sys.stdout.read() - self.assertIn('e', output) - - @mock.patch.object(cmscli, 'validate_args') - @mock.patch.object(cmscli.requests, 'post') - @mock.patch.object(cmscli.requests, 'get') - @mock.patch.object(cmscli.cli_common, 'get_token') - @mock.patch.object(cmscli, 'globals') - def test_list_groups_errors(self, mock_globals, mock_get_token, - mock_get, mock_post, - mock_validate_args): - mock_post.return_value = self.respond(TJ, 200) - mock_get.return_value = self.respond(TJ, 204, oy=True) - mock_get.__name__ = 'test' - with self.assertRaises(SystemExit) as cm: - args = ormcli.main('orm cms list_groups t'.split()) - self.assertEqual(cm.exception.code, 0) - sys.stdout.seek(0) - output = sys.stdout.read() - self.assertEqual('', output) - - mock_get.return_value = self.respond(TJ, 404, oy=True) - with self.assertRaises(SystemExit) as cm: - args = ormcli.main('orm cms --faceless list_groups t'.split()) - self.assertEqual(cm.exception.code, 1) - sys.stdout.seek(0) - output = sys.stdout.read() - self.assertIn('API error:', output) diff --git a/orm/tests/unit/rds/services/test_group_yaml.py b/orm/tests/unit/rds/services/test_group_yaml.py deleted file mode 100644 index 5fbeb7d2..00000000 --- a/orm/tests/unit/rds/services/test_group_yaml.py +++ /dev/null @@ -1,75 +0,0 @@ -"""unittests create group yaml module.""" -from mock import patch -import unittest -import yaml - -from orm.services.resource_distributor.rds.services import\ - yaml_group_builder as GroupBuild - -alldata = { - 'description': 'this is a description', 'enabled': 1, - 'domain_name': 'groupdomain', - "regions": [{ - "action": "modify", - "name": "local", - "users": []}], - 'name': 'test_group'} - -region_nousers = {'name': 'regionname', - 'rangerAgentVersion': 1.0, - 'users': []} - -region_users = {'name': 'regionname', - 'rangerAgentVersion': 1.0, - 'users': [{'id': ['user1', 'user2'], 'domain': 'domain1'}]} - -yaml_group_nousers = \ - 'heat_template_version: 2015-1-1\n\ndescription: yaml file for region - ' \ - 'regionname\n\nresources:\n' \ - ' test_group:\n properties:\n' \ - ' description: "this is a description"\n' \ - ' domain: groupdomain\n' \ - ' name: test_group\n' \ - ' type: OS::Keystone::Group\n\n\n' \ - 'outputs:\n test_group_id:\n value: {get_resource: test_group}\n' - -yaml_group_users = \ - 'heat_template_version: 2015-1-1\n\ndescription: yaml file for region - ' \ - 'regionname\n\nresources:\n' \ - ' test_group:\n properties:\n' \ - ' description: "this is a description"\n' \ - ' domain: groupdomain\n' \ - ' name: test_group\n' \ - ' type: OS::Keystone::Group\n\n\n' \ - ' test_group_user_assignments:\n properties:\n' \ - ' group: {get_resource: test_group}\n' \ - ' group_domain: groupdomain\n' \ - ' users:\n - name: user1\n user_domain: domain1\n' \ - ' - name: user2\n user_domain: domain1\n' \ - ' group_domain: groupdomain\n' \ - ' type: OS::Keystone::GroupUserAssignment\n'\ - '\n\noutputs:\n test_group_id:\n value: {get_resource: test_group}\n' \ - ' test_group_user_assignments_id:\n value: {get_resource: test_group_user_assignments}\n' - - -class CreateResource(unittest.TestCase): - """class metohd.""" - maxDiff = None - - @patch.object(GroupBuild, 'conf') - def test_create_group_yaml_nousers(self, mock_conf): - """test valid dict to yaml output as expected without users.""" - ver = mock_conf.yaml_configs.group_yaml.yaml_version = '2015-1-1' - yamlfile = GroupBuild.yamlbuilder(alldata, region_nousers) - yamlfile_as_json = yaml.safe_load(yamlfile) - self.assertEqual(yamlfile_as_json['heat_template_version'], ver) - self.assertEqual(yaml.safe_load(yamlfile), yaml.safe_load(yaml_group_nousers)) - - @patch.object(GroupBuild, 'conf') - def test_create_group_yaml_users(self, mock_conf): - """test valid dict to yaml output as expected without users.""" - ver = mock_conf.yaml_configs.group_yaml.yaml_version = '2015-1-1' - yamlfile = GroupBuild.yamlbuilder(alldata, region_users) - yamlfile_as_json = yaml.safe_load(yamlfile) - self.assertEqual(yamlfile_as_json['heat_template_version'], ver) - self.assertEqual(yaml.safe_load(yamlfile), yaml.safe_load(yaml_group_users))