Disable Groups controller for Keystone Group feature
This patchset disabled Groups controller and removed related API/CLI unit test cases. Change-Id: I6bbd3615c85cc9db98e57fd6fa9ccfeb80c572c9
This commit is contained in:
Chi Lo
parent
b249c3a675
commit
921179d5f6
@ -278,198 +278,6 @@ def add_to_parser(service_sub):
|
||||
parser_list_customer.add_argument('--metadata', action='append', nargs="+",
|
||||
type=str, help='<key:value>')
|
||||
|
||||
# create group
|
||||
parser_create_group = subparsers.add_parser('create_group',
|
||||
help='[<"X-RANGER-Client" '
|
||||
'header>] <data file '
|
||||
'with new group '
|
||||
'JSON>')
|
||||
parser_create_group.add_argument('client',
|
||||
**cli_common.ORM_CLIENT_KWARGS)
|
||||
parser_create_group.add_argument('datafile',
|
||||
type=argparse.FileType('r'),
|
||||
help='<data file with new group '
|
||||
'JSON>')
|
||||
# delete group
|
||||
parser_delete_group = subparsers.add_parser('delete_group',
|
||||
help='[<"X-RANGER-Client" '
|
||||
'header>] <group id>')
|
||||
parser_delete_group.add_argument('client', **cli_common.ORM_CLIENT_KWARGS)
|
||||
parser_delete_group.add_argument('groupid', type=str, help='<group id>')
|
||||
|
||||
# get group
|
||||
h1, h2 = '[<"X-RANGER-Client" header>]', '<group id or group name>'
|
||||
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 <name>] [--starts_with <name>] [--contains <name>]'
|
||||
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>] <group id> '
|
||||
'<data file with region(s) JSON>')
|
||||
parser_add_groups_region.add_argument(
|
||||
'client', **cli_common.ORM_CLIENT_KWARGS)
|
||||
parser_add_groups_region.add_argument(
|
||||
'groupid', type=str, help='<groupid id>')
|
||||
parser_add_groups_region.add_argument(
|
||||
'datafile', type=argparse.FileType('r'),
|
||||
help='<data file with region(s) JSON>')
|
||||
|
||||
parser_delete_groups_region = subparsers.add_parser(
|
||||
'delete_groups_region',
|
||||
help='[<"X-RANGER-Client" header>] [--force_delete] '
|
||||
'<group id> <region id>')
|
||||
parser_delete_groups_region.add_argument('client',
|
||||
**cli_common.ORM_CLIENT_KWARGS)
|
||||
parser_delete_groups_region.add_argument('groupid',
|
||||
type=str,
|
||||
help='<group id>')
|
||||
parser_delete_groups_region.add_argument('regionid',
|
||||
type=str,
|
||||
help='<region id>')
|
||||
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>] <group id> '
|
||||
'<data file with group role(s) assignment JSON>')
|
||||
parser_assign_group_roles.add_argument(
|
||||
'client', **cli_common.ORM_CLIENT_KWARGS)
|
||||
parser_assign_group_roles.add_argument(
|
||||
'groupid', type=str, help='<groupid id>')
|
||||
parser_assign_group_roles.add_argument(
|
||||
'datafile', type=argparse.FileType('r'),
|
||||
help='<data file with group role(s) assignment JSON>')
|
||||
|
||||
# assign group region roles
|
||||
parser_assign_group_region_roles = subparsers.add_parser(
|
||||
'assign_group_region_roles',
|
||||
help='[<"X-RANGER-Client" '
|
||||
'header>] <group id> <region_id> '
|
||||
'<data file with group role(s) assignment JSON>')
|
||||
parser_assign_group_region_roles.add_argument(
|
||||
'client', **cli_common.ORM_CLIENT_KWARGS)
|
||||
parser_assign_group_region_roles.add_argument(
|
||||
'groupid', type=str, help='<groupid id>')
|
||||
parser_assign_group_region_roles.add_argument(
|
||||
'regionid', type=str, help='<regionid id>')
|
||||
parser_assign_group_region_roles.add_argument(
|
||||
'datafile', type=argparse.FileType('r'),
|
||||
help='<data file with group role(s) assignment JSON>')
|
||||
|
||||
# unassign group roles
|
||||
parser_unassign_group_role = subparsers.add_parser(
|
||||
'unassign_group_role',
|
||||
help='[<"X-RANGER-Client" '
|
||||
'header>] <group id> <role> <--customer <customer id>> '
|
||||
'or <--domain <domain_name>>')
|
||||
parser_unassign_group_role.add_argument(
|
||||
'client', **cli_common.ORM_CLIENT_KWARGS)
|
||||
parser_unassign_group_role.add_argument(
|
||||
'groupid', type=str, help='<groupid id>')
|
||||
parser_unassign_group_role.add_argument(
|
||||
'role', type=str, help='<role>')
|
||||
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 = '<group id> <--region <name>> <--customer <customer id>> ' \
|
||||
'or <--domain <domain_name>>'
|
||||
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='<groupid id>')
|
||||
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>] <group id> '
|
||||
'<data file with group user(s) to be added JSON>')
|
||||
parser_add_group_default_users.add_argument(
|
||||
'client', **cli_common.ORM_CLIENT_KWARGS)
|
||||
parser_add_group_default_users.add_argument(
|
||||
'groupid', type=str, help='<groupid id>')
|
||||
parser_add_group_default_users.add_argument(
|
||||
'datafile', type=argparse.FileType('r'),
|
||||
help='<data file with group user(s) to be added JSON>')
|
||||
|
||||
# groups - delete group default user
|
||||
parser_delete_group_default_user = \
|
||||
subparsers.add_parser('delete_group_default_user',
|
||||
help='[<"X-RANGER-Client" header>] <group '
|
||||
'id> <user id>')
|
||||
parser_delete_group_default_user.add_argument(
|
||||
'client', **cli_common.ORM_CLIENT_KWARGS)
|
||||
parser_delete_group_default_user.add_argument(
|
||||
'groupid', type=str, help='<group id>')
|
||||
parser_delete_group_default_user.add_argument(
|
||||
'userid', type=str, help='<user id>')
|
||||
parser_delete_group_default_user.add_argument('userdomain', type=str,
|
||||
help='<user domain>')
|
||||
|
||||
# groups - add_group_region_users
|
||||
parser_add_group_region_user = subparsers.add_parser(
|
||||
'add_group_region_users',
|
||||
help='[<"X-RANGER-Client" header>] <group id> '
|
||||
'<region id> <data file with user(s) JSON>')
|
||||
parser_add_group_region_user.add_argument(
|
||||
'client', **cli_common.ORM_CLIENT_KWARGS)
|
||||
parser_add_group_region_user.add_argument(
|
||||
'groupid', type=str, help='<group id>')
|
||||
parser_add_group_region_user.add_argument(
|
||||
'regionid', type=str, help='<region id>')
|
||||
parser_add_group_region_user.add_argument(
|
||||
'datafile', type=argparse.FileType('r'),
|
||||
help='<data file with user(s) JSON>')
|
||||
|
||||
# groups - delete group region user
|
||||
parser_delete_group_region_user = subparsers.add_parser(
|
||||
'delete_group_region_user',
|
||||
help='[<"X-RANGER-Client" header>] '
|
||||
'<group id> <region id> <user id> <user domain>')
|
||||
parser_delete_group_region_user.add_argument(
|
||||
'client', **cli_common.ORM_CLIENT_KWARGS)
|
||||
parser_delete_group_region_user.add_argument('groupid', type=str,
|
||||
help='<group id>')
|
||||
parser_delete_group_region_user.add_argument('regionid', type=str,
|
||||
help='<region id>')
|
||||
parser_delete_group_region_user.add_argument('userid', type=str,
|
||||
help='<user id>')
|
||||
parser_delete_group_region_user.add_argument('userdomain', type=str,
|
||||
help='<user domain>')
|
||||
|
||||
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',
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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
|
||||
@ -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",
|
||||
}
|
||||
]
|
||||
@ -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()]
|
||||
}
|
||||
@ -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"
|
||||
]
|
||||
}
|
||||
]
|
||||
@ -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"
|
||||
}
|
||||
]
|
||||
@ -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"
|
||||
]
|
||||
}
|
||||
]
|
||||
@ -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"
|
||||
}
|
||||
]
|
||||
@ -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)
|
||||
|
||||
@ -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))
|
||||
Loading…
x
Reference in New Issue
Block a user