6f3fb47049
python2: https://review.opendev.org/#/c/693399 Change-Id: I0f68235685f03fc73fb848997d5eba5e8915b477
558 lines
22 KiB
Python
558 lines
22 KiB
Python
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
|