diff options
Diffstat (limited to 'keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_authz.py')
| -rw-r--r-- | keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_authz.py | 2749 |
1 files changed, 2010 insertions, 739 deletions
diff --git a/keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_authz.py b/keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_authz.py index 0e22a1b4..705aea8a 100644 --- a/keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_authz.py +++ b/keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_authz.py @@ -16,6 +16,7 @@ from keystone import resource from keystone.contrib.moon.exception import * from keystone.tests.unit import default_fixtures from keystone.contrib.moon.core import LogManager, TenantManager, ADMIN_ID +from keystone.tests.moon.unit import * CONF = cfg.CONF @@ -31,15 +32,22 @@ IE = { "description": "a simple description." } -class TestIntraExtensionAuthzManagerAuthz(tests.TestCase): +@dependency.requires('admin_api', 'authz_api', 'tenant_api', 'configuration_api', 'moonlog_api') +class TestIntraExtensionAuthzManagerAuthzOK(tests.TestCase): def setUp(self): self.useFixture(database.Database()) - super(TestIntraExtensionAuthzManagerAuthz, self).setUp() + super(TestIntraExtensionAuthzManagerAuthzOK, self).setUp() self.load_backends() self.load_fixtures(default_fixtures) - self.manager = IntraExtensionAuthzManager() - self.admin_manager = IntraExtensionAdminManager() + self.admin = create_user(self, username="admin") + self.demo = create_user(self, username="demo") + self.root_intra_extension = create_intra_extension(self, policy_model="policy_root") + # force re-initialization of the ADMIN_ID variable + from keystone.contrib.moon.core import ADMIN_ID + self.ADMIN_ID = ADMIN_ID + self.manager = self.authz_api + self.admin_manager = self.admin_api def __get_key_from_value(self, value, values_dict): return filter(lambda v: v[1] == value, values_dict.iteritems())[0][0] @@ -48,11 +56,13 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase): return { "moonlog_api": LogManager(), "tenant_api": TenantManager(), + "admin_api": IntraExtensionAdminManager(), + "authz_api": IntraExtensionAuthzManager(), # "resource_api": resource.Manager(), } def config_overrides(self): - super(TestIntraExtensionAuthzManagerAuthz, self).config_overrides() + super(TestIntraExtensionAuthzManagerAuthzOK, self).config_overrides() self.policy_directory = 'examples/moon/policies' self.config_fixture.config( group='moon', @@ -61,37 +71,942 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase): group='moon', policy_directory=self.policy_directory) - def create_tenant(self): - tenant = { - "id": uuid.uuid4().hex, - "name": "TestIntraExtensionAuthzManager", - "enabled": True, - "description": "", - "domain_id": "default" + def delete_admin_intra_extension(self): + self.manager.del_intra_extension(self.ref["id"]) + + def test_subjects(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + subjects = self.manager.get_subjects_dict(admin_subject_id, authz_ie_dict["id"]) + self.assertIsInstance(subjects, dict) + for key, value in subjects.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + self.assertIn("keystone_name", value) + self.assertIn("keystone_id", value) + + create_user(self, "subject_test") + new_subject = {"name": "subject_test", "description": "subject_test"} + + subjects = self.admin_manager.add_subject_dict(admin_subject_id, authz_ie_dict["id"], new_subject) + _subjects = dict(subjects) + self.assertEqual(len(_subjects.keys()), 1) + new_subject["id"] = _subjects.keys()[0] + value = subjects[new_subject["id"]] + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertEqual(value["name"], new_subject["name"]) + self.assertIn("description", value) + self.assertEqual(value["description"], new_subject["description"]) + + # Delete the new subject + self.admin_manager.del_subject(admin_subject_id, authz_ie_dict["id"], new_subject["id"]) + subjects = self.manager.get_subjects_dict(admin_subject_id, authz_ie_dict["id"]) + for key, value in subjects.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIsNot(new_subject["name"], value["name"]) + self.assertIn("description", value) + + def test_objects(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + objects = self.manager.get_objects_dict(admin_subject_id, authz_ie_dict["id"]) + objects_id_list = [] + self.assertIsInstance(objects, dict) + for key, value in objects.iteritems(): + objects_id_list.append(key) + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + + def test_actions(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + actions = self.manager.get_actions_dict(admin_subject_id, authz_ie_dict["id"]) + actions_id_list = [] + self.assertIsInstance(actions, dict) + for key, value in actions.iteritems(): + actions_id_list.append(key) + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + + def test_subject_categories(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + subject_categories = self.manager.get_subject_categories_dict(admin_subject_id, authz_ie_dict["id"]) + self.assertIsInstance(subject_categories, dict) + for key, value in subject_categories.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + + new_subject_category = {"name": "subject_category_test", "description": "subject_category_test"} + + subject_categories = self.admin_manager.add_subject_category(admin_subject_id, authz_ie_dict["id"], new_subject_category) + _subject_categories = dict(subject_categories) + self.assertEqual(len(_subject_categories.keys()), 1) + new_subject_category["id"] = _subject_categories.keys()[0] + value = subject_categories[new_subject_category["id"]] + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertEqual(value["name"], new_subject_category["name"]) + self.assertIn("description", value) + self.assertEqual(value["description"], new_subject_category["description"]) + + # Delete the new subject_category + self.admin_manager.del_subject_category(admin_subject_id, authz_ie_dict["id"], new_subject_category["id"]) + subject_categories = self.manager.get_subject_categories_dict(admin_subject_id, authz_ie_dict["id"]) + for key, value in subject_categories.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIsNot(new_subject_category["name"], value["name"]) + self.assertIn("description", value) + + def test_object_categories(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + object_categories = self.manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"]) + self.assertIsInstance(object_categories, dict) + for key, value in object_categories.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + + new_object_category = {"name": "object_category_test", "description": "object_category_test"} + + object_categories = self.admin_manager.add_object_category(admin_subject_id, authz_ie_dict["id"], new_object_category) + _object_categories = dict(object_categories) + self.assertEqual(len(_object_categories.keys()), 1) + new_object_category["id"] = _object_categories.keys()[0] + value = object_categories[new_object_category["id"]] + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertEqual(value["name"], new_object_category["name"]) + self.assertIn("description", value) + self.assertEqual(value["description"], new_object_category["description"]) + + # Delete the new object_category + + self.admin_manager.del_object_category(admin_subject_id, authz_ie_dict["id"], new_object_category["id"]) + object_categories = self.manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"]) + for key, value in object_categories.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIsNot(new_object_category["name"], value["name"]) + self.assertIn("description", value) + + def test_action_categories(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + action_categories = self.manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"]) + self.assertIsInstance(action_categories, dict) + for key, value in action_categories.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + + new_action_category = {"name": "action_category_test", "description": "action_category_test"} + + action_categories = self.admin_manager.add_action_category(admin_subject_id, authz_ie_dict["id"], new_action_category) + _action_categories = dict(action_categories) + self.assertEqual(len(_action_categories.keys()), 1) + new_action_category["id"] = _action_categories.keys()[0] + value = action_categories[new_action_category["id"]] + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertEqual(value["name"], new_action_category["name"]) + self.assertIn("description", value) + self.assertEqual(value["description"], new_action_category["description"]) + + # Delete the new action_category + + self.admin_manager.del_action_category(admin_subject_id, authz_ie_dict["id"], new_action_category["id"]) + action_categories = self.manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"]) + for key, value in action_categories.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIsNot(new_action_category["name"], value["name"]) + self.assertIn("description", value) + + def test_subject_category_scope(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + subject_categories = self.admin_manager.add_subject_category( + admin_subject_id, + authz_ie_dict["id"], + { + "name": "country", + "description": "country", + } + ) + + for subject_category_id in subject_categories: + + subject_category_scope = self.manager.get_subject_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + subject_category_id) + self.assertIsInstance(subject_category_scope, dict) + self.assertEqual({}, subject_category_scope) + + new_subject_category_scope = { + "name": "france", + "description": "france", + } + + subject_category_scope = self.admin_manager.add_subject_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + subject_category_id, + new_subject_category_scope) + self.assertIsInstance(subject_category_scope, dict) + self.assertEqual(len(subject_category_scope.keys()), 1) + subject_category_scope_id = subject_category_scope.keys()[0] + subject_category_scope_value = subject_category_scope[subject_category_scope_id] + self.assertIn("name", subject_category_scope_value) + self.assertEqual(new_subject_category_scope["name"], subject_category_scope_value["name"]) + self.assertIn("description", subject_category_scope_value) + self.assertEqual(new_subject_category_scope["description"], subject_category_scope_value["description"]) + + # Delete the new subject_category_scope + + self.admin_manager.del_subject_scope( + admin_subject_id, + authz_ie_dict["id"], + subject_category_id, + subject_category_scope_id) + subject_category_scope = self.admin_manager.get_subject_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + subject_category_id) + self.assertIsInstance(subject_category_scope, dict) + self.assertNotIn(subject_category_scope_id, subject_category_scope.keys()) + + def test_object_category_scope(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + object_categories = self.admin_manager.add_object_category( + admin_subject_id, + authz_ie_dict["id"], + { + "name": "country", + "description": "country", + } + ) + + for object_category_id in object_categories: + + object_category_scope = self.manager.get_object_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + object_category_id) + self.assertIsInstance(object_category_scope, dict) + self.assertEqual({}, object_category_scope) + + new_object_category_scope = { + "name": "france", + "description": "france", + } + + object_category_scope = self.admin_manager.add_object_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + object_category_id, + new_object_category_scope) + self.assertIsInstance(object_category_scope, dict) + self.assertEqual(len(object_category_scope.keys()), 1) + object_category_scope_id = object_category_scope.keys()[0] + object_category_scope_value = object_category_scope[object_category_scope_id] + self.assertIn("name", object_category_scope_value) + self.assertEqual(new_object_category_scope["name"], object_category_scope_value["name"]) + self.assertIn("description", object_category_scope_value) + self.assertEqual(new_object_category_scope["description"], object_category_scope_value["description"]) + + # Delete the new object_category_scope + + self.admin_manager.del_object_scope( + admin_subject_id, + authz_ie_dict["id"], + object_category_id, + object_category_scope_id) + object_category_scope = self.admin_manager.get_object_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + object_category_id) + self.assertIsInstance(object_category_scope, dict) + self.assertNotIn(object_category_scope_id, object_category_scope.keys()) + + def test_action_category_scope(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + action_categories = self.admin_manager.add_action_category( + admin_subject_id, + authz_ie_dict["id"], + { + "name": "swift", + "description": "swift actions", + } + ) + + for action_category_id in action_categories: + + action_category_scope = self.manager.get_action_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + action_category_id) + self.assertIsInstance(action_category_scope, dict) + self.assertEqual({}, action_category_scope) + + new_action_category_scope = { + "name": "get", + "description": "get swift files", + } + + action_category_scope = self.admin_manager.add_action_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + action_category_id, + new_action_category_scope) + self.assertIsInstance(action_category_scope, dict) + self.assertEqual(len(action_category_scope.keys()), 1) + action_category_scope_id = action_category_scope.keys()[0] + action_category_scope_value = action_category_scope[action_category_scope_id] + self.assertIn("name", action_category_scope_value) + self.assertEqual(new_action_category_scope["name"], action_category_scope_value["name"]) + self.assertIn("description", action_category_scope_value) + self.assertEqual(new_action_category_scope["description"], action_category_scope_value["description"]) + + # Delete the new action_category_scope + + self.admin_manager.del_action_scope( + admin_subject_id, + authz_ie_dict["id"], + action_category_id, + action_category_scope_id) + action_category_scope = self.admin_manager.get_action_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + action_category_id) + self.assertIsInstance(action_category_scope, dict) + self.assertNotIn(action_category_scope_id, action_category_scope.keys()) + + def test_subject_category_assignment(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + admin_authz_subject_id, admin_authz_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], authz_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + demo_authz_subject_id, demo_authz_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], authz_ie_dict['id'], 'demo').iteritems().next() + + subjects_dict = self.manager.get_subjects_dict(admin_subject_id, authz_ie_dict["id"]) + + subject_categories = self.admin_manager.add_subject_category( + admin_subject_id, + authz_ie_dict["id"], + { + "name": "country", + "description": "country", + } + ) + + for subject_category_id in subject_categories: + subject_category_scope = self.manager.get_subject_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + subject_category_id) + self.assertIsInstance(subject_category_scope, dict) + self.assertEqual({}, subject_category_scope) + + new_subject_category_scope_1 = { + "name": "france", + "description": "france", + } + + subject_category_scope_1 = self.admin_manager.add_subject_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + subject_category_id, + new_subject_category_scope_1) + subject_category_scope_1_id = subject_category_scope_1.keys()[0] + + new_subject_category_scope_2 = { + "name": "china", + "description": "china", + } + + subject_category_scope_2 = self.admin_manager.add_subject_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + subject_category_id, + new_subject_category_scope_2) + subject_category_scope_2_id = subject_category_scope_2.keys()[0] + + subject_category_assignments = self.manager.get_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + admin_authz_subject_id, + subject_category_id + ) + self.assertIsInstance(subject_category_assignments, list) + self.assertEqual([], subject_category_assignments) + + subject_category_assignments = self.manager.get_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + demo_authz_subject_id, + subject_category_id + ) + self.assertIsInstance(subject_category_assignments, list) + self.assertEqual([], subject_category_assignments) + + subject_category_assignments = self.admin_manager.add_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + admin_authz_subject_id, subject_category_id, subject_category_scope_1_id + ) + self.assertIsInstance(subject_category_assignments, list) + + self.assertEqual(len(subject_category_assignments), 1) + + subject_category_assignments = self.admin_manager.add_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + admin_authz_subject_id, subject_category_id, subject_category_scope_2_id + ) + self.assertIsInstance(subject_category_assignments, list) + self.assertEqual(len(subject_category_assignments), 2) + + subject_category_assignments = self.admin_manager.add_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + demo_authz_subject_id, subject_category_id, subject_category_scope_2_id + ) + self.assertIsInstance(subject_category_assignments, list) + self.assertEqual(len(subject_category_assignments), 1) + + subject_category_assignments = self.admin_manager.get_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + admin_authz_subject_id, subject_category_id + ) + self.assertIsInstance(subject_category_assignments, list) + self.assertEqual(len(subject_category_assignments), 2) + + self.admin_manager.del_subject_assignment( + admin_subject_id, + authz_ie_dict["id"], + admin_authz_subject_id, subject_category_id, subject_category_scope_2_id + ) + subject_category_assignments = self.admin_manager.get_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + admin_authz_subject_id, subject_category_id + ) + self.assertIsInstance(subject_category_assignments, list) + self.assertEqual(len(subject_category_assignments), 1) + + def test_object_category_assignment(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + objects_dict = self.manager.get_objects_dict(admin_subject_id, authz_ie_dict["id"]) + + object_vm1_id = None + object_vm2_id = None + for _object_id in objects_dict: + if objects_dict[_object_id]['name'] == 'vm1': + object_vm1_id = _object_id + if objects_dict[_object_id]['name'] == 'vm2': + object_vm2_id = _object_id + if not object_vm1_id or not object_vm2_id: + raise Exception("Cannot run tests, database is corrupted ? (need upload and list in objects)") + + object_categories = self.admin_manager.add_object_category( + admin_subject_id, + authz_ie_dict["id"], + { + "name": "location", + "description": "location", + } + ) + + for object_category_id in object_categories: + object_category_scope = self.manager.get_object_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + object_category_id) + self.assertIsInstance(object_category_scope, dict) + self.assertEqual({}, object_category_scope) + + new_object_category_scope_1 = { + "name": "france", + "description": "france", + } + + object_category_scope_1 = self.admin_manager.add_object_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + object_category_id, + new_object_category_scope_1) + object_category_scope_1_id = object_category_scope_1.keys()[0] + + new_object_category_scope_2 = { + "name": "china", + "description": "china", + } + + object_category_scope_2 = self.admin_manager.add_object_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + object_category_id, + new_object_category_scope_2) + object_category_scope_2_id = object_category_scope_2.keys()[0] + + object_category_assignments = self.manager.get_object_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + object_vm1_id, + object_category_id + ) + self.assertIsInstance(object_category_assignments, list) + self.assertEqual([], object_category_assignments) + + object_category_assignments = self.manager.get_object_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + object_vm2_id, + object_category_id + ) + self.assertIsInstance(object_category_assignments, list) + self.assertEqual([], object_category_assignments) + + object_category_assignments = self.admin_manager.add_object_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + object_vm1_id, object_category_id, object_category_scope_1_id + ) + self.assertIsInstance(object_category_assignments, list) + + self.assertEqual(len(object_category_assignments), 1) + + object_category_assignments = self.admin_manager.add_object_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + object_vm1_id, object_category_id, object_category_scope_2_id + ) + self.assertIsInstance(object_category_assignments, list) + self.assertEqual(len(object_category_assignments), 2) + + object_category_assignments = self.admin_manager.add_object_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + object_vm2_id, object_category_id, object_category_scope_2_id + ) + self.assertIsInstance(object_category_assignments, list) + self.assertEqual(len(object_category_assignments), 1) + + object_category_assignments = self.admin_manager.get_object_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + object_vm1_id, object_category_id + ) + self.assertIsInstance(object_category_assignments, list) + self.assertEqual(len(object_category_assignments), 2) + + self.admin_manager.del_object_assignment( + admin_subject_id, + authz_ie_dict["id"], + object_vm1_id, object_category_id, object_category_scope_2_id + ) + object_category_assignments = self.admin_manager.get_object_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + object_vm1_id, object_category_id + ) + self.assertIsInstance(object_category_assignments, list) + self.assertEqual(len(object_category_assignments), 1) + + def test_action_category_assignment(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + actions_dict = self.manager.get_actions_dict(admin_subject_id, authz_ie_dict["id"]) + + action_upload_id = None + action_list_id = None + for _action_id in actions_dict: + if actions_dict[_action_id]['name'] == 'upload': + action_upload_id = _action_id + if actions_dict[_action_id]['name'] == 'list': + action_list_id = _action_id + if not action_upload_id or not action_list_id: + raise Exception("Cannot run tests, database is corrupted ? (need upload and list in actions)") + + action_categories = self.admin_manager.add_action_category( + admin_subject_id, + authz_ie_dict["id"], + { + "name": "swift", + "description": "swift actions", + } + ) + + for action_category_id in action_categories: + action_category_scope = self.manager.get_action_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + action_category_id) + self.assertIsInstance(action_category_scope, dict) + self.assertEqual({}, action_category_scope) + + new_action_category_scope_1 = { + "name": "swift_admin", + "description": "action require admin rights", + } + + action_category_scope_1 = self.admin_manager.add_action_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + action_category_id, + new_action_category_scope_1) + action_category_scope_1_id = action_category_scope_1.keys()[0] + + new_action_category_scope_2 = { + "name": "swift_anonymous", + "description": "action require no right", + } + + action_category_scope_2 = self.admin_manager.add_action_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + action_category_id, + new_action_category_scope_2) + action_category_scope_2_id = action_category_scope_2.keys()[0] + + action_category_assignments = self.manager.get_action_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + action_upload_id, + action_category_id + ) + self.assertIsInstance(action_category_assignments, list) + self.assertEqual([], action_category_assignments) + + action_category_assignments = self.manager.get_action_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + action_list_id, + action_category_id + ) + self.assertIsInstance(action_category_assignments, list) + self.assertEqual([], action_category_assignments) + + action_category_assignments = self.admin_manager.add_action_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + action_upload_id, action_category_id, action_category_scope_1_id + ) + self.assertIsInstance(action_category_assignments, list) + + self.assertEqual(len(action_category_assignments), 1) + + action_category_assignments = self.admin_manager.add_action_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + action_upload_id, action_category_id, action_category_scope_2_id + ) + self.assertIsInstance(action_category_assignments, list) + self.assertEqual(len(action_category_assignments), 2) + + action_category_assignments = self.admin_manager.add_action_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + action_list_id, action_category_id, action_category_scope_2_id + ) + self.assertIsInstance(action_category_assignments, list) + self.assertEqual(len(action_category_assignments), 1) + + action_category_assignments = self.admin_manager.get_action_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + action_upload_id, action_category_id + ) + self.assertIsInstance(action_category_assignments, list) + self.assertEqual(len(action_category_assignments), 2) + + self.admin_manager.del_action_assignment( + admin_subject_id, + authz_ie_dict["id"], + action_upload_id, action_category_id, action_category_scope_2_id + ) + action_category_assignments = self.admin_manager.get_action_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + action_upload_id, action_category_id + ) + self.assertIsInstance(action_category_assignments, list) + self.assertEqual(len(action_category_assignments), 1) + + def test_sub_meta_rules(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.admin_manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + aggregation_algorithms = self.admin_manager.get_aggregation_algorithm_dict(admin_subject_id, authz_ie_dict["id"]) + for key, value in aggregation_algorithms.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + + # TODO: need more tests on aggregation_algorithms (set and del) + + sub_meta_rules = self.admin_manager.get_sub_meta_rules_dict(admin_subject_id, authz_ie_dict["id"]) + self.assertIsInstance(sub_meta_rules, dict) + categories = { + "subject_categories": self.admin_manager.get_subject_categories_dict(admin_subject_id, authz_ie_dict["id"]), + "object_categories": self.admin_manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"]), + "action_categories": self.admin_manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"]) } - return self.resource_api.create_project(tenant["id"], tenant) + for key, value in sub_meta_rules.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("action_categories", value) + self.assertIn("object_categories", value) + self.assertIn("subject_categories", value) + self.assertIn("algorithm", value) + self.assertIn("name", value) + for action_category_id in value["action_categories"]: + self.assertIn(action_category_id, categories["action_categories"]) + for object_category_id in value["object_categories"]: + self.assertIn(object_category_id, categories["object_categories"]) + for subject_category_id in value["subject_categories"]: + self.assertIn(subject_category_id, categories["subject_categories"]) + # TODO: need more tests (set and del) - def create_mapping(self, tenant, authz_uuid=None, admin_uuid=None): + def test_sub_rules(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.admin_manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + sub_meta_rules = self.admin_manager.get_sub_meta_rules_dict(admin_subject_id, authz_ie_dict["id"]) + self.assertIsInstance(sub_meta_rules, dict) + + for relation_id in sub_meta_rules: + rules = self.admin_manager.get_rules_dict(admin_subject_id, authz_ie_dict["id"], relation_id) + rule_length = len(sub_meta_rules[relation_id]["subject_categories"]) + \ + len(sub_meta_rules[relation_id]["object_categories"]) + \ + len(sub_meta_rules[relation_id]["action_categories"]) + 1 + for rule_id in rules: + self.assertEqual(rule_length, len(rules[rule_id])) + rule = list(rules[rule_id]) + for cat, cat_func, func_name in ( + ("subject_categories", self.admin_manager.get_subject_scopes_dict, "subject_scope"), + ("action_categories", self.admin_manager.get_action_scopes_dict, "action_scope"), + ("object_categories", self.admin_manager.get_object_scopes_dict, "object_scope"), + ): + for cat_value in sub_meta_rules[relation_id][cat]: + scope = cat_func( + admin_subject_id, + authz_ie_dict["id"], + cat_value + ) + a_scope = rule.pop(0) + if type(a_scope) is not bool: + self.assertIn(a_scope, scope.keys()) + + # add a new subrule + + sub_rule = [] + for cat, cat_func, func_name in ( + ("subject_categories", self.admin_manager.get_subject_scopes_dict, "subject_scope"), + ("action_categories", self.admin_manager.get_action_scopes_dict, "action_scope"), + ("object_categories", self.admin_manager.get_object_scopes_dict, "object_scope"), + ): + for cat_value in sub_meta_rules[relation_id][cat]: + scope = cat_func( + admin_subject_id, + authz_ie_dict["id"], + cat_value + ) + sub_rule.append(scope.keys()[0]) + + sub_rule.append(False) + + sub_rules = self.admin_manager.add_rule_dict(admin_subject_id, authz_ie_dict["id"], relation_id, sub_rule) + self.assertIsInstance(sub_rules, dict) + self.assertIn(sub_rule, sub_rules.values()) + + for rule_id, rule_value in sub_rules.iteritems(): + for cat, cat_func, func_name in ( + ("subject_categories", self.admin_manager.get_subject_scopes_dict, "subject_category_scope"), + ("action_categories", self.admin_manager.get_action_scopes_dict, "action_category_scope"), + ("object_categories", self.admin_manager.get_object_scopes_dict, "object_category_scope"), + ): + for cat_value in sub_meta_rules[relation_id][cat]: + scope = cat_func( + admin_subject_id, + authz_ie_dict["id"], + cat_value + ) + a_scope = rule_value.pop(0) + self.assertIn(a_scope, scope.keys()) + + # TODO: add test for the delete function - mapping = self.tenant_api.set_tenant_dict(tenant["id"], tenant["name"], authz_uuid, admin_uuid) - self.assertIsInstance(mapping, dict) - self.assertIn("authz", mapping) - self.assertEqual(mapping["authz"], authz_uuid) - return mapping - def create_user(self, username="admin"): +@dependency.requires('admin_api', 'authz_api', 'tenant_api', 'configuration_api', 'moonlog_api') +class TestIntraExtensionAuthzManagerAuthzKO(tests.TestCase): + + def setUp(self): + self.useFixture(database.Database()) + super(TestIntraExtensionAuthzManagerAuthzKO, self).setUp() + self.load_backends() + self.load_fixtures(default_fixtures) + self.admin = create_user(self, username="admin") + self.demo = create_user(self, username="demo") + self.root_intra_extension = create_intra_extension(self, policy_model="policy_root") + # force re-initialization of the ADMIN_ID variable + from keystone.contrib.moon.core import ADMIN_ID + self.ADMIN_ID = ADMIN_ID + self.manager = self.authz_api + self.admin_manager = self.admin_api - _USER = dict(USER) - _USER["name"] = username - return self.identity_api.create_user(_USER) + def __get_key_from_value(self, value, values_dict): + return filter(lambda v: v[1] == value, values_dict.iteritems())[0][0] - def create_intra_extension(self, policy_model="policy_authz"): + def load_extra_backends(self): + return { + "moonlog_api": LogManager(), + "tenant_api": TenantManager(), + "admin_api": IntraExtensionAdminManager(), + "authz_api": IntraExtensionAuthzManager(), + # "resource_api": resource.Manager(), + } - IE["model"] = policy_model - IE["name"] = uuid.uuid4().hex - ref = self.admin_manager.load_intra_extension_dict(ADMIN_ID, intra_extension_dict=IE) - self.assertIsInstance(ref, dict) - return ref + def config_overrides(self): + super(TestIntraExtensionAuthzManagerAuthzKO, self).config_overrides() + self.policy_directory = 'examples/moon/policies' + self.config_fixture.config( + group='moon', + intraextension_driver='keystone.contrib.moon.backends.sql.IntraExtensionConnector') + self.config_fixture.config( + group='moon', + policy_directory=self.policy_directory) def test_tenant_exceptions(self): self.assertRaises( @@ -142,137 +1057,149 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase): def test_authz_exceptions(self): self.assertRaises( - IntraExtensionUnknown, + TenantUnknown, self.manager.authz, uuid.uuid4().hex, uuid.uuid4().hex, uuid.uuid4().hex, uuid.uuid4().hex ) - admin_user = self.create_user() - tenant = self.create_tenant() - ie_authz = self.create_intra_extension("policy_authz") - ie_admin = self.create_intra_extension("policy_admin") - mapping = self.create_mapping(tenant, ie_authz["id"], ie_admin["id"]) + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() # Test when subject is unknown self.assertRaises( - AuthzException, + SubjectUnknown, self.manager.authz, - ie_authz["id"], uuid.uuid4().hex, uuid.uuid4().hex, uuid.uuid4().hex + tenant["name"], uuid.uuid4().hex, uuid.uuid4().hex, uuid.uuid4().hex ) # Test when subject is known but not the object - demo_user = self.create_user("demo") - self.manager.add_subject_dict( - admin_user['id'], - self.ref["id"], - demo_user["id"] - ) + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], authz_ie_dict['id'], 'demo').iteritems().next() + + # self.manager.add_subject_dict( + # admin_subject_id, + # ie_authz["id"], + # demo_user["id"] + # ) self.assertRaises( - AuthzException, + ObjectUnknown, self.manager.authz, - ie_authz["id"], demo_user["id"], uuid.uuid4().hex, uuid.uuid4().hex + tenant["name"], demo_subject_dict["name"], uuid.uuid4().hex, uuid.uuid4().hex ) # Test when subject and object are known but not the action + my_object = {"name": "my_object", "description": "my_object description"} _tmp = self.manager.add_object_dict( - admin_user['id'], - self.ref["id"], - "my_object" - ).items()[0] - my_object = {"id": _tmp[0], "name": _tmp[1]} + admin_subject_id, + authz_ie_dict["id"], + my_object + ) + my_object["id"] = _tmp.keys()[0] self.assertRaises( - AuthzException, + ActionUnknown, self.manager.authz, - ie_authz["id"], demo_user["id"], my_object["id"], uuid.uuid4().hex + tenant["name"], demo_subject_dict["name"], my_object["name"], uuid.uuid4().hex ) # Test when subject and object and action are known + my_action = {"name": "my_action", "description": "my_action description"} _tmp = self.manager.add_action_dict( - admin_user['id'], - self.ref["id"], - "my_action" - ).items()[0] - my_action = {"id": _tmp[0], "name": _tmp[1]} + admin_subject_id, + authz_ie_dict["id"], + my_action + ) + my_action["id"] = _tmp.keys()[0] self.assertRaises( AuthzException, self.manager.authz, - ie_authz["id"], demo_user["id"], my_object["id"], my_action["id"] + tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"] ) # Add a subject scope and test ObjectCategoryAssignmentOutOfScope + my_subject_category = {"name": "my_subject_category", "description": "my_subject_category description"} _tmp = self.manager.add_subject_category( - admin_user['id'], - self.ref["id"], - "my_subject_category" + admin_subject_id, + authz_ie_dict["id"], + my_subject_category ) - my_subject_category = {"id": _tmp[0], "name": _tmp[1]} + my_subject_category["id"] = _tmp.keys()[0] + my_subject_scope = {"name": "my_subject_scope", "description": "my_subject_scope description"} _tmp = self.manager.add_subject_scope_dict( - admin_user['id'], - self.ref["id"], + admin_subject_id, + authz_ie_dict["id"], my_subject_category["id"], - "my_subject_scope", + my_subject_scope ) - my_subject_scope = {"id": _tmp[0], "name": _tmp[1]} + my_subject_scope["id"] = _tmp.keys()[0] self.assertRaises( AuthzException, self.manager.authz, - ie_authz["id"], demo_user["id"], my_object["id"], my_action["id"] + tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"] ) # Add an object scope and test ActionCategoryAssignmentOutOfScope + my_object_category = {"name": "my_object_category", "description": "my_object_category description"} _tmp = self.manager.add_object_category( - admin_user['id'], - self.ref["id"], - "my_object_category" + admin_subject_id, + authz_ie_dict["id"], + my_object_category ) - my_object_category = {"id": _tmp[0], "name": _tmp[1]} + my_object_category["id"] = _tmp.keys()[0] + my_object_scope = {"name": "my_object_scope", "description": "my_object_scope description"} _tmp = self.manager.add_object_scope_dict( - admin_user['id'], - self.ref["id"], + admin_subject_id, + authz_ie_dict["id"], my_object_category["id"], - "my_object_scope", + my_object_scope ) - my_object_scope = {"id": _tmp[0], "name": _tmp[1]} + my_object_scope["id"] = _tmp.keys()[0] self.assertRaises( AuthzException, self.manager.authz, - ie_authz["id"], demo_user["id"], my_object["id"], my_action["id"] + tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"] ) # Add an action scope and test SubjectCategoryAssignmentUnknown + my_action_category = {"name": "my_action_category", "description": "my_action_category description"} _tmp = self.manager.add_action_category( - admin_user['id'], - self.ref["id"], - "my_action_category" + admin_subject_id, + authz_ie_dict["id"], + my_action_category ) - my_action_category = {"id": _tmp[0], "name": _tmp[1]} + my_action_category["id"] = _tmp.keys()[0] + my_action_scope = {"name": "my_action_scope", "description": "my_action_scope description"} _tmp = self.manager.add_action_scope_dict( - admin_user['id'], - self.ref["id"], + admin_subject_id, + authz_ie_dict["id"], my_action_category["id"], - "my_action_scope", + my_action_scope ) - my_action_scope = {"id": _tmp[0], "name": _tmp[1]} + my_action_scope["id"] = _tmp.keys()[0] self.assertRaises( AuthzException, self.manager.authz, - ie_authz["id"], demo_user["id"], my_object["id"], my_action["id"] + tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"] ) # Add a subject assignment and test ObjectCategoryAssignmentUnknown self.manager.add_subject_assignment_list( - admin_user['id'], - self.ref["id"], - demo_user["id"], + admin_subject_id, + authz_ie_dict["id"], + demo_subject_id, my_subject_category["id"], my_subject_scope["id"] ) @@ -280,14 +1207,14 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase): self.assertRaises( AuthzException, self.manager.authz, - ie_authz["id"], demo_user["id"], my_object["id"], my_action["id"] + tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"] ) # Add an object assignment and test ActionCategoryAssignmentUnknown self.manager.add_object_assignment_list( - admin_user['id'], - self.ref["id"], - demo_user["id"], + admin_subject_id, + authz_ie_dict["id"], + my_object["id"], my_object_category["id"], my_object_scope["id"] ) @@ -295,14 +1222,14 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase): self.assertRaises( AuthzException, self.manager.authz, - ie_authz["id"], demo_user["id"], my_object["id"], my_action["id"] + tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"] ) # Add an action assignment and test RuleUnknown self.manager.add_action_assignment_list( - admin_user['id'], - self.ref["id"], - demo_user["id"], + admin_subject_id, + authz_ie_dict["id"], + my_action["id"], my_action_category["id"], my_action_scope["id"] ) @@ -310,828 +1237,1172 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase): self.assertRaises( AuthzException, self.manager.authz, - ie_authz["id"], demo_user["id"], my_object["id"], my_action["id"] + tenant["name"], admin_subject_dict["name"], my_object["name"], my_action["name"] ) # Add the correct rule and test that no exception is raised my_meta_rule = { - "relation_super": { - "subject_categories": [my_subject_category["id"], ], - "action_categories": [my_object_category["id"], ], - "object_categories": [my_action_category["id"], ], - "relation": "relation_super" - } + "name": "my_meta_rule", + "algorithm": "test", + "subject_categories": [my_subject_category["id"], ], + "action_categories": [my_action_category["id"], ], + "object_categories": [my_object_category["id"], ] } - self.manager.get_sub_meta_rule_dict( - admin_user['id'], - self.ref["id"], + print("my_meta_rule", my_meta_rule) + sub_meta_rules_dict = self.manager.get_sub_meta_rules_dict( + admin_subject_id, + authz_ie_dict["id"] + ) + + self.assertRaises( + SubMetaRuleAlgorithmNotExisting, + self.manager.add_sub_meta_rule_dict, + admin_subject_id, + authz_ie_dict["id"], my_meta_rule ) - self.manager.set_sub_rule( - admin_user['id'], - self.ref["id"], - "relation_super", - [my_subject_scope, my_object_scope, my_action_scope] + + # TODO: the next request should be called with demo_subject_id + # but the demo user has no right in the root intra_extension + algorithms = self.configuration_api.get_sub_meta_rule_algorithms_dict(admin_subject_id) + for algorithm_id in algorithms: + if algorithms[algorithm_id]["name"] == "inclusion": + my_meta_rule["algorithm"] = algorithm_id + + sub_meta_rule = self.manager.add_sub_meta_rule_dict( + admin_subject_id, + authz_ie_dict["id"], + my_meta_rule ) + sub_meta_rule_id, sub_meta_rule_dict = sub_meta_rule.iteritems().next() - result = self.manager.authz(ie_authz["id"], demo_user["id"], my_object["id"], my_action["id"]) + rule = self.manager.add_rule_dict( + admin_subject_id, + authz_ie_dict["id"], + sub_meta_rule_id, + [my_subject_scope["id"], my_action_scope["id"], my_object_scope["id"], True] + ) + + self.assertRaises( + AuthzException, + self.manager.authz, + tenant["name"], admin_subject_dict["name"], my_object["name"], my_action["name"] + ) + + result = self.manager.authz(tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"]) self.assertEqual(True, result) def test_subjects(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) - - subjects = self.manager.get_subjects_dict(admin_user["id"], tenant["id"]) + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + subjects = self.manager.get_subjects_dict(admin_subject_id, authz_ie_dict["id"]) self.assertIsInstance(subjects, dict) - self.assertIn("subjects", subjects) - self.assertIn("id", subjects) - self.assertIn("intra_extension_uuid", subjects) - self.assertEqual(ref["id"], subjects["intra_extension_uuid"]) - self.assertIsInstance(subjects["subjects"], dict) - - new_subject = self.create_user("my_user") - new_subjects = dict() - new_subjects[new_subject["id"]] = new_subject["name"] + for key, value in subjects.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + self.assertIn("keystone_name", value) + self.assertIn("keystone_id", value) + + create_user(self, "subject_test") + new_subject = {"name": "subject_test", "description": "subject_test"} self.assertRaises( AuthzException, - self.manager.set_subject_dict, - admin_user["id"], ref["id"], new_subjects) + self.manager.add_subject_dict, + demo_subject_id, admin_ie_dict["id"], new_subject) + + subjects = self.admin_manager.add_subject_dict(admin_subject_id, authz_ie_dict["id"], new_subject) + _subjects = dict(subjects) + self.assertEqual(len(_subjects.keys()), 1) + new_subject["id"] = _subjects.keys()[0] + value = subjects[new_subject["id"]] + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertEqual(value["name"], new_subject["name"]) + self.assertIn("description", value) + self.assertEqual(value["description"], new_subject["description"]) # Delete the new subject self.assertRaises( AuthzException, self.manager.del_subject, - admin_user["id"], ref["id"], new_subject["id"]) + demo_subject_id, authz_ie_dict["id"], new_subject["id"]) - # Add a particular subject - self.assertRaises( - AuthzException, - self.manager.add_subject_dict, - admin_user["id"], ref["id"], new_subject["id"]) + self.admin_manager.del_subject(admin_subject_id, authz_ie_dict["id"], new_subject["id"]) + subjects = self.manager.get_subjects_dict(admin_subject_id, authz_ie_dict["id"]) + for key, value in subjects.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIsNot(new_subject["name"], value["name"]) + self.assertIn("description", value) def test_objects(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) - - objects = self.manager.get_objects_dict(admin_user["id"], tenant["id"]) + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + objects = self.manager.get_objects_dict(admin_subject_id, authz_ie_dict["id"]) + objects_id_list = [] self.assertIsInstance(objects, dict) - self.assertIn("objects", objects) - self.assertIn("id", objects) - self.assertIn("intra_extension_uuid", objects) - self.assertEqual(ref["id"], objects["intra_extension_uuid"]) - self.assertIsInstance(objects["objects"], dict) - - new_object = {"id": uuid.uuid4().hex, "name": "my_object"} - new_objects = dict() - new_objects[new_object["id"]] = new_object["name"] + for key, value in objects.iteritems(): + objects_id_list.append(key) + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + + create_user(self, "subject_test") + new_object = {"name": "object_test", "description": "object_test"} self.assertRaises( AuthzException, - self.manager.set_object_dict, - admin_user["id"], ref["id"], new_object["id"]) + self.manager.add_object_dict, + demo_subject_id, admin_ie_dict["id"], new_object) - # Delete the new object self.assertRaises( - AuthzException, - self.manager.del_object, - admin_user["id"], ref["id"], new_object["id"]) + ObjectsWriteNoAuthorized, + self.admin_manager.add_object_dict, + admin_subject_id, authz_ie_dict["id"], new_object + ) - # Add a particular object - self.assertRaises( - AuthzException, - self.manager.add_object_dict, - admin_user["id"], ref["id"], new_object["name"]) + # Delete the new object + for key in objects_id_list: + self.assertRaises( + AuthzException, + self.manager.del_object, + demo_subject_id, authz_ie_dict["id"], key) + self.assertRaises( + AuthzException, + self.manager.del_object, + admin_subject_id, authz_ie_dict["id"], key) - def test_actions(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) + for key in objects_id_list: + self.assertRaises( + ObjectsWriteNoAuthorized, + self.admin_manager.del_object, + demo_subject_id, authz_ie_dict["id"], key) + self.assertRaises( + ObjectsWriteNoAuthorized, + self.admin_manager.del_object, + admin_subject_id, authz_ie_dict["id"], key) - actions = self.manager.get_actions_dict(admin_user["id"], tenant["id"]) + def test_actions(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + actions = self.manager.get_actions_dict(admin_subject_id, authz_ie_dict["id"]) + actions_id_list = [] self.assertIsInstance(actions, dict) - self.assertIn("actions", actions) - self.assertIn("id", actions) - self.assertIn("intra_extension_uuid", actions) - self.assertEqual(ref["id"], actions["intra_extension_uuid"]) - self.assertIsInstance(actions["actions"], dict) - - new_action = {"id": uuid.uuid4().hex, "name": "my_action"} - new_actions = dict() - new_actions[new_action["id"]] = new_action["name"] + for key, value in actions.iteritems(): + actions_id_list.append(key) + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + + create_user(self, "subject_test") + new_action = {"name": "action_test", "description": "action_test"} self.assertRaises( AuthzException, - self.manager.set_action_dict, - admin_user["id"], ref["id"], new_actions) + self.manager.add_action_dict, + demo_subject_id, admin_ie_dict["id"], new_action) - # Delete the new action self.assertRaises( - AuthzException, - self.manager.del_action, - admin_user["id"], ref["id"], new_action["id"]) + ActionsWriteNoAuthorized, + self.admin_manager.add_action_dict, + admin_subject_id, authz_ie_dict["id"], new_action + ) - # Add a particular action - self.assertRaises( - AuthzException, - self.manager.add_action_dict, - admin_user["id"], ref["id"], new_action["id"]) + # Delete all actions + for key in actions_id_list: + self.assertRaises( + AuthzException, + self.manager.del_action, + demo_subject_id, authz_ie_dict["id"], key) + self.assertRaises( + AuthzException, + self.manager.del_action, + admin_subject_id, authz_ie_dict["id"], key) - def test_subject_categories(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) + for key in actions_id_list: + self.assertRaises( + ActionsWriteNoAuthorized, + self.admin_manager.del_action, + demo_subject_id, authz_ie_dict["id"], key) + self.assertRaises( + ActionsWriteNoAuthorized, + self.admin_manager.del_action, + admin_subject_id, authz_ie_dict["id"], key) - subject_categories = self.manager.get_subject_categories_dict(admin_user["id"], ref["id"]) + def test_subject_categories(self): + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + subject_categories = self.manager.get_subject_categories_dict(admin_subject_id, authz_ie_dict["id"]) self.assertIsInstance(subject_categories, dict) - self.assertIn("subject_categories", subject_categories) - self.assertIn("id", subject_categories) - self.assertIn("intra_extension_uuid", subject_categories) - self.assertEqual(ref["id"], subject_categories["intra_extension_uuid"]) - self.assertIsInstance(subject_categories["subject_categories"], dict) - - new_subject_category = {"id": uuid.uuid4().hex, "name": "subject_category_test"} - new_subject_categories = dict() - new_subject_categories[new_subject_category["id"]] = new_subject_category["name"] + for key, value in subject_categories.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + + new_subject_category = {"name": "subject_category_test", "description": "subject_category_test"} self.assertRaises( AuthzException, - self.manager.set_subject_category_dict, - admin_user["id"], ref["id"], new_subject_categories) + self.manager.add_subject_category, + demo_subject_id, admin_ie_dict["id"], new_subject_category) + + subject_categories = self.admin_manager.add_subject_category(admin_subject_id, authz_ie_dict["id"], new_subject_category) + _subject_categories = dict(subject_categories) + self.assertEqual(len(_subject_categories.keys()), 1) + new_subject_category["id"] = _subject_categories.keys()[0] + value = subject_categories[new_subject_category["id"]] + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertEqual(value["name"], new_subject_category["name"]) + self.assertIn("description", value) + self.assertEqual(value["description"], new_subject_category["description"]) # Delete the new subject_category self.assertRaises( AuthzException, self.manager.del_subject_category, - admin_user["id"], ref["id"], new_subject_category["id"]) + demo_subject_id, authz_ie_dict["id"], new_subject_category["id"]) - # Add a particular subject_category - self.assertRaises( - AuthzException, - self.manager.add_subject_category, - admin_user["id"], ref["id"], new_subject_category["name"]) + self.admin_manager.del_subject_category(admin_subject_id, authz_ie_dict["id"], new_subject_category["id"]) + subject_categories = self.manager.get_subject_categories_dict(admin_subject_id, authz_ie_dict["id"]) + for key, value in subject_categories.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIsNot(new_subject_category["name"], value["name"]) + self.assertIn("description", value) def test_object_categories(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) - - object_categories = self.manager.get_object_category_dict(admin_user["id"], ref["id"]) + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + object_categories = self.manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"]) self.assertIsInstance(object_categories, dict) - self.assertIn("object_categories", object_categories) - self.assertIn("id", object_categories) - self.assertIn("intra_extension_uuid", object_categories) - self.assertEqual(ref["id"], object_categories["intra_extension_uuid"]) - self.assertIsInstance(object_categories["object_categories"], dict) - - new_object_category = {"id": uuid.uuid4().hex, "name": "object_category_test"} - new_object_categories = dict() - new_object_categories[new_object_category["id"]] = new_object_category["name"] + for key, value in object_categories.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + + new_object_category = {"name": "object_category_test", "description": "object_category_test"} self.assertRaises( AuthzException, - self.manager.set_object_category_dict, - admin_user["id"], ref["id"], new_object_categories) + self.manager.add_object_category, + demo_subject_id, admin_ie_dict["id"], new_object_category) + + object_categories = self.admin_manager.add_object_category(admin_subject_id, authz_ie_dict["id"], new_object_category) + _object_categories = dict(object_categories) + self.assertEqual(len(_object_categories.keys()), 1) + new_object_category["id"] = _object_categories.keys()[0] + value = object_categories[new_object_category["id"]] + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertEqual(value["name"], new_object_category["name"]) + self.assertIn("description", value) + self.assertEqual(value["description"], new_object_category["description"]) # Delete the new object_category self.assertRaises( AuthzException, self.manager.del_object_category, - admin_user["id"], ref["id"], new_object_category["id"]) + demo_subject_id, authz_ie_dict["id"], new_object_category["id"]) - # Add a particular object_category - self.assertRaises( - AuthzException, - self.manager.add_object_category, - admin_user["id"], ref["id"], new_object_category["name"]) + self.admin_manager.del_object_category(admin_subject_id, authz_ie_dict["id"], new_object_category["id"]) + object_categories = self.manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"]) + for key, value in object_categories.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIsNot(new_object_category["name"], value["name"]) + self.assertIn("description", value) def test_action_categories(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) - - action_categories = self.manager.get_action_category_dict(admin_user["id"], ref["id"]) + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + action_categories = self.manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"]) self.assertIsInstance(action_categories, dict) - self.assertIn("action_categories", action_categories) - self.assertIn("id", action_categories) - self.assertIn("intra_extension_uuid", action_categories) - self.assertEqual(ref["id"], action_categories["intra_extension_uuid"]) - self.assertIsInstance(action_categories["action_categories"], dict) - - new_action_category = {"id": uuid.uuid4().hex, "name": "action_category_test"} - new_action_categories = dict() - new_action_categories[new_action_category["id"]] = new_action_category["name"] + for key, value in action_categories.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + + new_action_category = {"name": "action_category_test", "description": "action_category_test"} self.assertRaises( AuthzException, - self.manager.set_action_category_dict, - admin_user["id"], ref["id"], new_action_categories) + self.manager.add_action_category, + demo_subject_id, admin_ie_dict["id"], new_action_category) + + action_categories = self.admin_manager.add_action_category(admin_subject_id, authz_ie_dict["id"], new_action_category) + _action_categories = dict(action_categories) + self.assertEqual(len(_action_categories.keys()), 1) + new_action_category["id"] = _action_categories.keys()[0] + value = action_categories[new_action_category["id"]] + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertEqual(value["name"], new_action_category["name"]) + self.assertIn("description", value) + self.assertEqual(value["description"], new_action_category["description"]) # Delete the new action_category self.assertRaises( AuthzException, self.manager.del_action_category, - admin_user["id"], ref["id"], new_action_category["id"]) + demo_subject_id, authz_ie_dict["id"], new_action_category["id"]) - # Add a particular action_category - self.assertRaises( - AuthzException, - self.manager.add_action_category, - admin_user["id"], ref["id"], new_action_category["name"]) + self.admin_manager.del_action_category(admin_subject_id, authz_ie_dict["id"], new_action_category["id"]) + action_categories = self.manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"]) + for key, value in action_categories.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIsNot(new_action_category["name"], value["name"]) + self.assertIn("description", value) def test_subject_category_scope(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) - - subject_categories = self.admin_manager.set_subject_category_dict( - admin_user["id"], - ref["id"], + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + subject_categories = self.admin_manager.add_subject_category( + admin_subject_id, + authz_ie_dict["id"], { - uuid.uuid4().hex: admin_user["id"], - uuid.uuid4().hex: "dev", + "name": "country", + "description": "country", } ) - for subject_category in subject_categories["subject_categories"]: + for subject_category_id in subject_categories: + subject_category_scope = self.manager.get_subject_scopes_dict( - admin_user["id"], - ref["id"], - subject_category) + admin_subject_id, + authz_ie_dict["id"], + subject_category_id) self.assertIsInstance(subject_category_scope, dict) - self.assertIn("subject_category_scope", subject_category_scope) - self.assertIn("id", subject_category_scope) - self.assertIn("intra_extension_uuid", subject_category_scope) - self.assertEqual(ref["id"], subject_category_scope["intra_extension_uuid"]) - self.assertIsInstance(subject_category_scope["subject_category_scope"], dict) - - new_subject_category_scope = dict() - new_subject_category_scope_uuid = uuid.uuid4().hex - new_subject_category_scope[new_subject_category_scope_uuid] = "new_subject_category_scope" - self.assertRaises( - AuthzException, - self.manager.set_subject_scope_dict, - admin_user["id"], ref["id"], subject_category, new_subject_category_scope) + self.assertEqual({}, subject_category_scope) + + new_subject_category_scope = { + "name": "france", + "description": "france", + } - # Delete the new subject_category_scope self.assertRaises( AuthzException, - self.manager.del_subject_scope, - admin_user["id"], ref["id"], subject_category, new_subject_category_scope_uuid) + self.admin_manager.add_subject_scope_dict, + demo_subject_id, authz_ie_dict["id"], subject_category_id, new_subject_category_scope) - # Add a particular subject_category_scope + subject_category_scope = self.admin_manager.add_subject_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + subject_category_id, + new_subject_category_scope) + self.assertIsInstance(subject_category_scope, dict) + self.assertEqual(len(subject_category_scope.keys()), 1) + subject_category_scope_id = subject_category_scope.keys()[0] + subject_category_scope_value = subject_category_scope[subject_category_scope_id] + self.assertIn("name", subject_category_scope_value) + self.assertEqual(new_subject_category_scope["name"], subject_category_scope_value["name"]) + self.assertIn("description", subject_category_scope_value) + self.assertEqual(new_subject_category_scope["description"], subject_category_scope_value["description"]) + + # Delete the new subject_category_scope self.assertRaises( AuthzException, - self.manager.add_subject_scope_dict, - admin_user["id"], ref["id"], subject_category, new_subject_category_scope[new_subject_category_scope_uuid]) + self.admin_manager.del_subject_scope, + demo_subject_id, authz_ie_dict["id"], subject_category_id, subject_category_scope_id) + + self.admin_manager.del_subject_scope( + admin_subject_id, + authz_ie_dict["id"], + subject_category_id, + subject_category_scope_id) + subject_category_scope = self.admin_manager.get_subject_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + subject_category_id) + self.assertIsInstance(subject_category_scope, dict) + self.assertNotIn(subject_category_scope_id, subject_category_scope.keys()) def test_object_category_scope(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) - - object_categories = self.admin_manager.set_object_category_dict( - admin_user["id"], - ref["id"], + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + object_categories = self.admin_manager.add_object_category( + admin_subject_id, + authz_ie_dict["id"], { - uuid.uuid4().hex: "id", - uuid.uuid4().hex: "domain", + "name": "country", + "description": "country", } ) - for object_category in object_categories["object_categories"]: + for object_category_id in object_categories: + object_category_scope = self.manager.get_object_scopes_dict( - admin_user["id"], - ref["id"], - object_category) + admin_subject_id, + authz_ie_dict["id"], + object_category_id) self.assertIsInstance(object_category_scope, dict) - self.assertIn("object_category_scope", object_category_scope) - self.assertIn("id", object_category_scope) - self.assertIn("intra_extension_uuid", object_category_scope) - self.assertEqual(ref["id"], object_category_scope["intra_extension_uuid"]) - self.assertIsInstance(object_category_scope["object_category_scope"], dict) - - new_object_category_scope = dict() - new_object_category_scope_uuid = uuid.uuid4().hex - new_object_category_scope[new_object_category_scope_uuid] = "new_object_category_scope" - self.assertRaises( - AuthzException, - self.manager.set_object_scope_dict, - admin_user["id"], ref["id"], object_category, new_object_category_scope) + self.assertEqual({}, object_category_scope) + + new_object_category_scope = { + "name": "france", + "description": "france", + } - # Delete the new object_category_scope self.assertRaises( AuthzException, - self.manager.del_object_scope, - admin_user["id"], ref["id"], object_category, new_object_category_scope_uuid) + self.admin_manager.add_object_scope_dict, + demo_subject_id, authz_ie_dict["id"], object_category_id, new_object_category_scope) + + object_category_scope = self.admin_manager.add_object_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + object_category_id, + new_object_category_scope) + self.assertIsInstance(object_category_scope, dict) + self.assertEqual(len(object_category_scope.keys()), 1) + object_category_scope_id = object_category_scope.keys()[0] + object_category_scope_value = object_category_scope[object_category_scope_id] + self.assertIn("name", object_category_scope_value) + self.assertEqual(new_object_category_scope["name"], object_category_scope_value["name"]) + self.assertIn("description", object_category_scope_value) + self.assertEqual(new_object_category_scope["description"], object_category_scope_value["description"]) - # Add a particular object_category_scope + # Delete the new object_category_scope self.assertRaises( AuthzException, - self.manager.add_object_scope_dict, - admin_user["id"], ref["id"], object_category, new_object_category_scope[new_object_category_scope_uuid]) + self.admin_manager.del_object_scope, + demo_subject_id, authz_ie_dict["id"], object_category_id, object_category_scope_id) + + self.admin_manager.del_object_scope( + admin_subject_id, + authz_ie_dict["id"], + object_category_id, + object_category_scope_id) + object_category_scope = self.admin_manager.get_object_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + object_category_id) + self.assertIsInstance(object_category_scope, dict) + self.assertNotIn(object_category_scope_id, object_category_scope.keys()) def test_action_category_scope(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) - - action_categories = self.admin_manager.set_action_category_dict( - admin_user["id"], - ref["id"], + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + action_categories = self.admin_manager.add_action_category( + admin_subject_id, + authz_ie_dict["id"], { - uuid.uuid4().hex: "compute", - uuid.uuid4().hex: "identity", + "name": "swift", + "description": "swift actions", } ) - for action_category in action_categories["action_categories"]: + for action_category_id in action_categories: + action_category_scope = self.manager.get_action_scopes_dict( - admin_user["id"], - ref["id"], - action_category) + admin_subject_id, + authz_ie_dict["id"], + action_category_id) self.assertIsInstance(action_category_scope, dict) - self.assertIn("action_category_scope", action_category_scope) - self.assertIn("id", action_category_scope) - self.assertIn("intra_extension_uuid", action_category_scope) - self.assertEqual(ref["id"], action_category_scope["intra_extension_uuid"]) - self.assertIsInstance(action_category_scope["action_category_scope"], dict) - - new_action_category_scope = dict() - new_action_category_scope_uuid = uuid.uuid4().hex - new_action_category_scope[new_action_category_scope_uuid] = "new_action_category_scope" - self.assertRaises( - AuthzException, - self.manager.set_action_scope_dict, - admin_user["id"], ref["id"], action_category, new_action_category_scope) + self.assertEqual({}, action_category_scope) + + new_action_category_scope = { + "name": "get", + "description": "get swift files", + } - # Delete the new action_category_scope self.assertRaises( AuthzException, - self.manager.del_action_scope, - admin_user["id"], ref["id"], action_category, new_action_category_scope_uuid) + self.admin_manager.add_action_scope_dict, + demo_subject_id, authz_ie_dict["id"], action_category_id, new_action_category_scope) - # Add a particular action_category_scope + action_category_scope = self.admin_manager.add_action_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + action_category_id, + new_action_category_scope) + self.assertIsInstance(action_category_scope, dict) + self.assertEqual(len(action_category_scope.keys()), 1) + action_category_scope_id = action_category_scope.keys()[0] + action_category_scope_value = action_category_scope[action_category_scope_id] + self.assertIn("name", action_category_scope_value) + self.assertEqual(new_action_category_scope["name"], action_category_scope_value["name"]) + self.assertIn("description", action_category_scope_value) + self.assertEqual(new_action_category_scope["description"], action_category_scope_value["description"]) + + # Delete the new action_category_scope self.assertRaises( AuthzException, - self.manager.add_action_scope_dict, - admin_user["id"], ref["id"], action_category, new_action_category_scope[new_action_category_scope_uuid]) + self.admin_manager.del_action_scope, + demo_subject_id, authz_ie_dict["id"], action_category_id, action_category_scope_id) + + self.admin_manager.del_action_scope( + admin_subject_id, + authz_ie_dict["id"], + action_category_id, + action_category_scope_id) + action_category_scope = self.admin_manager.get_action_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + action_category_id) + self.assertIsInstance(action_category_scope, dict) + self.assertNotIn(action_category_scope_id, action_category_scope.keys()) def test_subject_category_assignment(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) - - new_subject = self.create_user() - new_subjects = dict() - new_subjects[new_subject["id"]] = new_subject["name"] - subjects = self.admin_manager.set_subject_dict(admin_user["id"], ref["id"], new_subjects) - - new_subject_category_uuid = uuid.uuid4().hex - new_subject_category_value = "role" - subject_categories = self.admin_manager.set_subject_category_dict( - admin_user["id"], - ref["id"], + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + admin_authz_subject_id, admin_authz_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], authz_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + demo_authz_subject_id, demo_authz_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], authz_ie_dict['id'], 'demo').iteritems().next() + + subjects_dict = self.manager.get_subjects_dict(admin_subject_id, authz_ie_dict["id"]) + + subject_categories = self.admin_manager.add_subject_category( + admin_subject_id, + authz_ie_dict["id"], { - new_subject_category_uuid: new_subject_category_value + "name": "country", + "description": "country", } ) - for subject_category in subject_categories["subject_categories"]: - subject_category_scope = self.admin_manager.get_subject_scopes_dict( - admin_user["id"], - ref["id"], - subject_category) - self.assertIsInstance(subject_category_scope, dict) - self.assertIn("subject_category_scope", subject_category_scope) - self.assertIn("id", subject_category_scope) - self.assertIn("intra_extension_uuid", subject_category_scope) - self.assertEqual(ref["id"], subject_category_scope["intra_extension_uuid"]) - self.assertIsInstance(subject_category_scope["subject_category_scope"], dict) - - new_subject_category_scope = dict() - new_subject_category_scope_uuid = uuid.uuid4().hex - new_subject_category_scope[new_subject_category_scope_uuid] = admin_user["id"] - subject_category_scope = self.admin_manager.set_subject_scope_dict( - admin_user["id"], - ref["id"], - subject_category, - new_subject_category_scope) - self.assertIsInstance(subject_category_scope, dict) - self.assertIn("subject_category_scope", subject_category_scope) - self.assertIn("id", subject_category_scope) - self.assertIn("intra_extension_uuid", subject_category_scope) - self.assertEqual(ref["id"], subject_category_scope["intra_extension_uuid"]) - self.assertIn(new_subject_category_scope[new_subject_category_scope_uuid], - subject_category_scope["subject_category_scope"][subject_category].values()) - - new_subject_category_scope2 = dict() - new_subject_category_scope2_uuid = uuid.uuid4().hex - new_subject_category_scope2[new_subject_category_scope2_uuid] = "dev" - subject_category_scope = self.admin_manager.set_subject_scope_dict( - admin_user["id"], - ref["id"], - subject_category, - new_subject_category_scope2) + for subject_category_id in subject_categories: + subject_category_scope = self.manager.get_subject_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + subject_category_id) self.assertIsInstance(subject_category_scope, dict) - self.assertIn("subject_category_scope", subject_category_scope) - self.assertIn("id", subject_category_scope) - self.assertIn("intra_extension_uuid", subject_category_scope) - self.assertEqual(ref["id"], subject_category_scope["intra_extension_uuid"]) - self.assertIn(new_subject_category_scope2[new_subject_category_scope2_uuid], - subject_category_scope["subject_category_scope"][subject_category].values()) - - subject_category_assignments = self.manager.get_subject_assignment_dict( - admin_user["id"], - ref["id"], - new_subject["id"] + self.assertEqual({}, subject_category_scope) + + new_subject_category_scope_1 = { + "name": "france", + "description": "france", + } + + subject_category_scope_1 = self.admin_manager.add_subject_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + subject_category_id, + new_subject_category_scope_1) + subject_category_scope_1_id = subject_category_scope_1.keys()[0] + + new_subject_category_scope_2 = { + "name": "china", + "description": "china", + } + + subject_category_scope_2 = self.admin_manager.add_subject_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + subject_category_id, + new_subject_category_scope_2) + subject_category_scope_2_id = subject_category_scope_2.keys()[0] + + subject_category_assignments = self.manager.get_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + admin_authz_subject_id, + subject_category_id ) - self.assertIsInstance(subject_category_assignments, dict) - self.assertIn("subject_category_assignments", subject_category_assignments) - self.assertIn("id", subject_category_assignments) - self.assertIn("intra_extension_uuid", subject_category_assignments) - self.assertEqual(ref["id"], subject_category_assignments["intra_extension_uuid"]) - self.assertEqual({}, subject_category_assignments["subject_category_assignments"][new_subject["id"]]) + self.assertIsInstance(subject_category_assignments, list) + self.assertEqual([], subject_category_assignments) + + subject_category_assignments = self.manager.get_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + demo_authz_subject_id, + subject_category_id + ) + self.assertIsInstance(subject_category_assignments, list) + self.assertEqual([], subject_category_assignments) self.assertRaises( AuthzException, - self.manager.set_subject_assignment_dict, - admin_user["id"], ref["id"], new_subject["id"], - { - new_subject_category_uuid: [new_subject_category_scope_uuid, new_subject_category_scope2_uuid], - }) + self.manager.add_subject_assignment_list, + demo_subject_id, authz_ie_dict["id"], + admin_authz_subject_id, subject_category_id, subject_category_scope_1_id + ) self.assertRaises( AuthzException, - self.manager.del_subject_assignment, - admin_user["id"], ref["id"], new_subject["id"], - new_subject_category_uuid, - new_subject_category_scope_uuid) + self.manager.add_subject_assignment_list, + demo_subject_id, authz_ie_dict["id"], + demo_authz_subject_id, subject_category_id, subject_category_scope_2_id + ) + + subject_category_assignments = self.admin_manager.add_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + admin_authz_subject_id, subject_category_id, subject_category_scope_1_id + ) + self.assertIsInstance(subject_category_assignments, list) + + self.assertEqual(len(subject_category_assignments), 1) + + subject_category_assignments = self.admin_manager.add_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + admin_authz_subject_id, subject_category_id, subject_category_scope_2_id + ) + self.assertIsInstance(subject_category_assignments, list) + self.assertEqual(len(subject_category_assignments), 2) + + subject_category_assignments = self.admin_manager.add_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + demo_authz_subject_id, subject_category_id, subject_category_scope_2_id + ) + self.assertIsInstance(subject_category_assignments, list) + self.assertEqual(len(subject_category_assignments), 1) + + subject_category_assignments = self.admin_manager.get_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + admin_authz_subject_id, subject_category_id + ) + self.assertIsInstance(subject_category_assignments, list) + self.assertEqual(len(subject_category_assignments), 2) self.assertRaises( AuthzException, - self.manager.add_subject_assignment_list, - admin_user["id"], ref["id"], new_subject["id"], - new_subject_category_uuid, - new_subject_category_scope_uuid) + self.admin_manager.del_subject_assignment, + demo_subject_id, authz_ie_dict["id"], + demo_authz_subject_id, subject_category_id, subject_category_scope_2_id + ) + + self.admin_manager.del_subject_assignment( + admin_subject_id, + authz_ie_dict["id"], + admin_authz_subject_id, subject_category_id, subject_category_scope_2_id + ) + subject_category_assignments = self.admin_manager.get_subject_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + admin_authz_subject_id, subject_category_id + ) + self.assertIsInstance(subject_category_assignments, list) + self.assertEqual(len(subject_category_assignments), 1) + + self.assertRaises( + SubjectAssignmentUnknown, + self.admin_manager.del_subject_assignment, + admin_subject_id, + authz_ie_dict["id"], + admin_authz_subject_id, subject_category_id, subject_category_scope_2_id + ) def test_object_category_assignment(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) - - new_object = {"id": uuid.uuid4().hex, "name": "my_object"} - new_objects = dict() - new_objects[new_object["id"]] = new_object["name"] - objects = self.admin_manager.set_object_dict(admin_user["id"], ref["id"], new_objects) - - new_object_category_uuid = uuid.uuid4().hex - new_object_category_value = "role" - object_categories = self.admin_manager.set_object_category_dict( - admin_user["id"], - ref["id"], + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + objects_dict = self.manager.get_objects_dict(admin_subject_id, authz_ie_dict["id"]) + + object_vm1_id = None + object_vm2_id = None + for _object_id in objects_dict: + if objects_dict[_object_id]['name'] == 'vm1': + object_vm1_id = _object_id + if objects_dict[_object_id]['name'] == 'vm2': + object_vm2_id = _object_id + if not object_vm1_id or not object_vm2_id: + raise Exception("Cannot run tests, database is corrupted ? (need upload and list in objects)") + + object_categories = self.admin_manager.add_object_category( + admin_subject_id, + authz_ie_dict["id"], { - new_object_category_uuid: new_object_category_value + "name": "location", + "description": "location", } ) - for object_category in object_categories["object_categories"]: - object_category_scope = self.admin_manager.get_object_scopes_dict( - admin_user["id"], - ref["id"], - object_category) - self.assertIsInstance(object_category_scope, dict) - self.assertIn("object_category_scope", object_category_scope) - self.assertIn("id", object_category_scope) - self.assertIn("intra_extension_uuid", object_category_scope) - self.assertEqual(ref["id"], object_category_scope["intra_extension_uuid"]) - self.assertIsInstance(object_category_scope["object_category_scope"], dict) - - new_object_category_scope = dict() - new_object_category_scope_uuid = uuid.uuid4().hex - new_object_category_scope[new_object_category_scope_uuid] = admin_user["id"] - object_category_scope = self.admin_manager.set_object_scope_dict( - admin_user["id"], - ref["id"], - object_category, - new_object_category_scope) - self.assertIsInstance(object_category_scope, dict) - self.assertIn("object_category_scope", object_category_scope) - self.assertIn("id", object_category_scope) - self.assertIn("intra_extension_uuid", object_category_scope) - self.assertEqual(ref["id"], object_category_scope["intra_extension_uuid"]) - self.assertIn(new_object_category_scope[new_object_category_scope_uuid], - object_category_scope["object_category_scope"][object_category].values()) - - new_object_category_scope2 = dict() - new_object_category_scope2_uuid = uuid.uuid4().hex - new_object_category_scope2[new_object_category_scope2_uuid] = "dev" - object_category_scope = self.admin_manager.set_object_scope_dict( - admin_user["id"], - ref["id"], - object_category, - new_object_category_scope2) + for object_category_id in object_categories: + object_category_scope = self.manager.get_object_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + object_category_id) self.assertIsInstance(object_category_scope, dict) - self.assertIn("object_category_scope", object_category_scope) - self.assertIn("id", object_category_scope) - self.assertIn("intra_extension_uuid", object_category_scope) - self.assertEqual(ref["id"], object_category_scope["intra_extension_uuid"]) - self.assertIn(new_object_category_scope2[new_object_category_scope2_uuid], - object_category_scope["object_category_scope"][object_category].values()) + self.assertEqual({}, object_category_scope) + + new_object_category_scope_1 = { + "name": "france", + "description": "france", + } + + object_category_scope_1 = self.admin_manager.add_object_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + object_category_id, + new_object_category_scope_1) + object_category_scope_1_id = object_category_scope_1.keys()[0] + + new_object_category_scope_2 = { + "name": "china", + "description": "china", + } + + object_category_scope_2 = self.admin_manager.add_object_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + object_category_id, + new_object_category_scope_2) + object_category_scope_2_id = object_category_scope_2.keys()[0] object_category_assignments = self.manager.get_object_assignment_list( - admin_user["id"], - ref["id"], - new_object["id"] + admin_subject_id, + authz_ie_dict["id"], + object_vm1_id, + object_category_id ) - self.assertIsInstance(object_category_assignments, dict) - self.assertIn("object_category_assignments", object_category_assignments) - self.assertIn("id", object_category_assignments) - self.assertIn("intra_extension_uuid", object_category_assignments) - self.assertEqual(ref["id"], object_category_assignments["intra_extension_uuid"]) - self.assertEqual({}, object_category_assignments["object_category_assignments"][new_object["id"]]) + self.assertIsInstance(object_category_assignments, list) + self.assertEqual([], object_category_assignments) + + object_category_assignments = self.manager.get_object_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + object_vm2_id, + object_category_id + ) + self.assertIsInstance(object_category_assignments, list) + self.assertEqual([], object_category_assignments) self.assertRaises( AuthzException, - self.manager.set_object_category_assignment_dict, - admin_user["id"], ref["id"], new_object["id"], - { - new_object_category_uuid: [new_object_category_scope_uuid, new_object_category_scope2_uuid], - }) + self.manager.add_object_assignment_list, + demo_subject_id, authz_ie_dict["id"], + object_vm1_id, object_category_id, object_category_scope_1_id + ) self.assertRaises( AuthzException, - self.manager.del_object_assignment, - admin_user["id"], ref["id"], new_object["id"], - new_object_category_uuid, - new_object_category_scope_uuid) + self.manager.add_object_assignment_list, + demo_subject_id, authz_ie_dict["id"], + object_vm2_id, object_category_id, object_category_scope_2_id + ) + + object_category_assignments = self.admin_manager.add_object_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + object_vm1_id, object_category_id, object_category_scope_1_id + ) + self.assertIsInstance(object_category_assignments, list) + + self.assertEqual(len(object_category_assignments), 1) + + object_category_assignments = self.admin_manager.add_object_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + object_vm1_id, object_category_id, object_category_scope_2_id + ) + self.assertIsInstance(object_category_assignments, list) + self.assertEqual(len(object_category_assignments), 2) + + object_category_assignments = self.admin_manager.add_object_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + object_vm2_id, object_category_id, object_category_scope_2_id + ) + self.assertIsInstance(object_category_assignments, list) + self.assertEqual(len(object_category_assignments), 1) + + object_category_assignments = self.admin_manager.get_object_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + object_vm1_id, object_category_id + ) + self.assertIsInstance(object_category_assignments, list) + self.assertEqual(len(object_category_assignments), 2) self.assertRaises( AuthzException, - self.manager.add_object_assignment_list, - admin_user["id"], ref["id"], new_object["id"], - new_object_category_uuid, - new_object_category_scope_uuid) + self.admin_manager.del_object_assignment, + demo_subject_id, authz_ie_dict["id"], + object_vm2_id, object_category_id, object_category_scope_2_id + ) + + self.admin_manager.del_object_assignment( + admin_subject_id, + authz_ie_dict["id"], + object_vm1_id, object_category_id, object_category_scope_2_id + ) + object_category_assignments = self.admin_manager.get_object_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + object_vm1_id, object_category_id + ) + self.assertIsInstance(object_category_assignments, list) + self.assertEqual(len(object_category_assignments), 1) + + self.assertRaises( + ObjectAssignmentUnknown, + self.admin_manager.del_object_assignment, + admin_subject_id, + authz_ie_dict["id"], + object_vm1_id, object_category_id, object_category_scope_2_id + ) def test_action_category_assignment(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) - - new_action = {"id": uuid.uuid4().hex, "name": "my_action"} - new_actions = dict() - new_actions[new_action["id"]] = new_action["name"] - actions = self.admin_manager.set_action_dict(admin_user["id"], ref["id"], new_actions) - - new_action_category_uuid = uuid.uuid4().hex - new_action_category_value = "role" - action_categories = self.admin_manager.set_action_category_dict( - admin_user["id"], - ref["id"], + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + actions_dict = self.manager.get_actions_dict(admin_subject_id, authz_ie_dict["id"]) + + action_upload_id = None + action_list_id = None + for _action_id in actions_dict: + if actions_dict[_action_id]['name'] == 'upload': + action_upload_id = _action_id + if actions_dict[_action_id]['name'] == 'list': + action_list_id = _action_id + if not action_upload_id or not action_list_id: + raise Exception("Cannot run tests, database is corrupted ? (need upload and list in actions)") + + action_categories = self.admin_manager.add_action_category( + admin_subject_id, + authz_ie_dict["id"], { - new_action_category_uuid: new_action_category_value + "name": "swift", + "description": "swift actions", } ) - for action_category in action_categories["action_categories"]: - action_category_scope = self.admin_manager.get_action_scopes_dict( - admin_user["id"], - ref["id"], - action_category) - self.assertIsInstance(action_category_scope, dict) - self.assertIn("action_category_scope", action_category_scope) - self.assertIn("id", action_category_scope) - self.assertIn("intra_extension_uuid", action_category_scope) - self.assertEqual(ref["id"], action_category_scope["intra_extension_uuid"]) - self.assertIsInstance(action_category_scope["action_category_scope"], dict) - - new_action_category_scope = dict() - new_action_category_scope_uuid = uuid.uuid4().hex - new_action_category_scope[new_action_category_scope_uuid] = admin_user["id"] - action_category_scope = self.admin_manager.set_action_scope_dict( - admin_user["id"], - ref["id"], - action_category, - new_action_category_scope) - self.assertIsInstance(action_category_scope, dict) - self.assertIn("action_category_scope", action_category_scope) - self.assertIn("id", action_category_scope) - self.assertIn("intra_extension_uuid", action_category_scope) - self.assertEqual(ref["id"], action_category_scope["intra_extension_uuid"]) - self.assertIn(new_action_category_scope[new_action_category_scope_uuid], - action_category_scope["action_category_scope"][action_category].values()) - - new_action_category_scope2 = dict() - new_action_category_scope2_uuid = uuid.uuid4().hex - new_action_category_scope2[new_action_category_scope2_uuid] = "dev" - action_category_scope = self.admin_manager.set_action_scope_dict( - admin_user["id"], - ref["id"], - action_category, - new_action_category_scope2) + for action_category_id in action_categories: + action_category_scope = self.manager.get_action_scopes_dict( + admin_subject_id, + authz_ie_dict["id"], + action_category_id) self.assertIsInstance(action_category_scope, dict) - self.assertIn("action_category_scope", action_category_scope) - self.assertIn("id", action_category_scope) - self.assertIn("intra_extension_uuid", action_category_scope) - self.assertEqual(ref["id"], action_category_scope["intra_extension_uuid"]) - self.assertIn(new_action_category_scope2[new_action_category_scope2_uuid], - action_category_scope["action_category_scope"][action_category].values()) + self.assertEqual({}, action_category_scope) + + new_action_category_scope_1 = { + "name": "swift_admin", + "description": "action require admin rights", + } + + action_category_scope_1 = self.admin_manager.add_action_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + action_category_id, + new_action_category_scope_1) + action_category_scope_1_id = action_category_scope_1.keys()[0] + + new_action_category_scope_2 = { + "name": "swift_anonymous", + "description": "action require no right", + } + + action_category_scope_2 = self.admin_manager.add_action_scope_dict( + admin_subject_id, + authz_ie_dict["id"], + action_category_id, + new_action_category_scope_2) + action_category_scope_2_id = action_category_scope_2.keys()[0] action_category_assignments = self.manager.get_action_assignment_list( - admin_user["id"], - ref["id"], - new_action["id"] + admin_subject_id, + authz_ie_dict["id"], + action_upload_id, + action_category_id ) - self.assertIsInstance(action_category_assignments, dict) - self.assertIn("action_category_assignments", action_category_assignments) - self.assertIn("id", action_category_assignments) - self.assertIn("intra_extension_uuid", action_category_assignments) - self.assertEqual(ref["id"], action_category_assignments["intra_extension_uuid"]) - self.assertEqual({}, action_category_assignments["action_category_assignments"][new_action["id"]]) + self.assertIsInstance(action_category_assignments, list) + self.assertEqual([], action_category_assignments) + + action_category_assignments = self.manager.get_action_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + action_list_id, + action_category_id + ) + self.assertIsInstance(action_category_assignments, list) + self.assertEqual([], action_category_assignments) self.assertRaises( AuthzException, - self.manager.set_action_assignment_dict, - admin_user["id"], ref["id"], new_action["id"], - { - new_action_category_uuid: [new_action_category_scope_uuid, new_action_category_scope2_uuid], - }) + self.manager.add_action_assignment_list, + demo_subject_id, authz_ie_dict["id"], + action_upload_id, action_category_id, action_category_scope_1_id + ) + + self.assertRaises( + AuthzException, + self.manager.add_action_assignment_list, + demo_subject_id, authz_ie_dict["id"], + action_list_id, action_category_id, action_category_scope_2_id + ) + + action_category_assignments = self.admin_manager.add_action_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + action_upload_id, action_category_id, action_category_scope_1_id + ) + self.assertIsInstance(action_category_assignments, list) + + self.assertEqual(len(action_category_assignments), 1) + + action_category_assignments = self.admin_manager.add_action_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + action_upload_id, action_category_id, action_category_scope_2_id + ) + self.assertIsInstance(action_category_assignments, list) + self.assertEqual(len(action_category_assignments), 2) + + action_category_assignments = self.admin_manager.add_action_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + action_list_id, action_category_id, action_category_scope_2_id + ) + self.assertIsInstance(action_category_assignments, list) + self.assertEqual(len(action_category_assignments), 1) + + action_category_assignments = self.admin_manager.get_action_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + action_upload_id, action_category_id + ) + self.assertIsInstance(action_category_assignments, list) + self.assertEqual(len(action_category_assignments), 2) self.assertRaises( AuthzException, - self.manager.del_action_assignment, - admin_user["id"], ref["id"], new_action["id"], - new_action_category_uuid, - new_action_category_scope_uuid) + self.admin_manager.del_action_assignment, + demo_subject_id, authz_ie_dict["id"], + action_list_id, action_category_id, action_category_scope_2_id + ) + + self.admin_manager.del_action_assignment( + admin_subject_id, + authz_ie_dict["id"], + action_upload_id, action_category_id, action_category_scope_2_id + ) + action_category_assignments = self.admin_manager.get_action_assignment_list( + admin_subject_id, + authz_ie_dict["id"], + action_upload_id, action_category_id + ) + self.assertIsInstance(action_category_assignments, list) + self.assertEqual(len(action_category_assignments), 1) self.assertRaises( - ActionCategoryAssignmentAddNotAuthorized, - self.manager.add_action_assignment_list, - admin_user["id"], ref["id"], new_action["id"], - new_action_category_uuid, - new_action_category_scope_uuid) + ActionAssignmentUnknown, + self.admin_manager.del_action_assignment, + admin_subject_id, + authz_ie_dict["id"], + action_upload_id, action_category_id, action_category_scope_2_id + ) def test_sub_meta_rules(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) - - aggregation_algorithms = self.manager.get_aggregation_algorithms(admin_user["id"], ref["id"]) - self.assertIsInstance(aggregation_algorithms, dict) - self.assertIsInstance(aggregation_algorithms["aggregation_algorithms"], list) - self.assertIn("and_true_aggregation", aggregation_algorithms["aggregation_algorithms"]) - self.assertIn("test_aggregation", aggregation_algorithms["aggregation_algorithms"]) - - aggregation_algorithm = self.manager.get_aggregation_algorithm_dict(admin_user["id"], ref["id"]) - self.assertIsInstance(aggregation_algorithm, dict) - self.assertIn("aggregation", aggregation_algorithm) - self.assertIn(aggregation_algorithm["aggregation"], aggregation_algorithms["aggregation_algorithms"]) - - _aggregation_algorithm = list(aggregation_algorithms["aggregation_algorithms"]) - _aggregation_algorithm.remove(aggregation_algorithm["aggregation"]) - self.assertRaises( - AuthzException, - self.manager.set_aggregation_algorithm_dict, - admin_user["id"], ref["id"], _aggregation_algorithm[0]) - - sub_meta_rules = self.manager.get_sub_meta_rules_dict(admin_user["id"], ref["id"]) + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.admin_manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + aggregation_algorithms = self.admin_manager.get_aggregation_algorithm_dict(admin_subject_id, authz_ie_dict["id"]) + for key, value in aggregation_algorithms.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("name", value) + self.assertIn("description", value) + + # TODO: need more tests on aggregation_algorithms (set and del) + + sub_meta_rules = self.admin_manager.get_sub_meta_rules_dict(admin_subject_id, authz_ie_dict["id"]) self.assertIsInstance(sub_meta_rules, dict) - self.assertIn("sub_meta_rules", sub_meta_rules) - sub_meta_rules_conf = json.load(open(os.path.join(self.policy_directory, ref["model"], "metarule.json"))) - metarule = dict() categories = { - "subject_categories": self.manager.get_subject_categories_dict(admin_user["id"], ref["id"]), - "object_categories": self.manager.get_object_category_dict(admin_user["id"], ref["id"]), - "action_categories": self.manager.get_action_category_dict(admin_user["id"], ref["id"]) + "subject_categories": self.admin_manager.get_subject_categories_dict(admin_subject_id, authz_ie_dict["id"]), + "object_categories": self.admin_manager.get_object_category_dict(admin_subject_id, authz_ie_dict["id"]), + "action_categories": self.admin_manager.get_action_category_dict(admin_subject_id, authz_ie_dict["id"]) } - for relation in sub_meta_rules_conf["sub_meta_rules"]: - metarule[relation] = dict() - for item in ("subject_categories", "object_categories", "action_categories"): - metarule[relation][item] = list() - for element in sub_meta_rules_conf["sub_meta_rules"][relation][item]: - metarule[relation][item].append(self.__get_key_from_value( - element, - categories[item][item] - )) - - for relation in sub_meta_rules["sub_meta_rules"]: - self.assertIn(relation, metarule) - for item in ("subject_categories", "object_categories", "action_categories"): - self.assertEqual( - sub_meta_rules["sub_meta_rules"][relation][item], - metarule[relation][item] - ) - - new_subject_category = {"id": uuid.uuid4().hex, "name": "subject_category_test"} - # Add a particular subject_category - data = self.admin_manager.add_subject_category( - admin_user["id"], - ref["id"], - new_subject_category["name"]) - new_subject_category["id"] = data["subject_category"]["uuid"] - subject_categories = self.manager.get_subject_categories_dict( - admin_user["id"], - ref["id"]) - self.assertIsInstance(subject_categories, dict) - self.assertIn("subject_categories", subject_categories) - self.assertIn("id", subject_categories) - self.assertIn("intra_extension_uuid", subject_categories) - self.assertEqual(ref["id"], subject_categories["intra_extension_uuid"]) - self.assertIn(new_subject_category["id"], subject_categories["subject_categories"]) - metarule[relation]["subject_categories"].append(new_subject_category["id"]) - self.assertRaises( - AuthzException, - self.manager.get_sub_meta_rule_dict, - admin_user["id"], ref["id"], metarule) + for key, value in sub_meta_rules.iteritems(): + self.assertIsInstance(value, dict) + self.assertIn("action_categories", value) + self.assertIn("object_categories", value) + self.assertIn("subject_categories", value) + self.assertIn("algorithm", value) + self.assertIn("name", value) + for action_category_id in value["action_categories"]: + self.assertIn(action_category_id, categories["action_categories"]) + for object_category_id in value["object_categories"]: + self.assertIn(object_category_id, categories["object_categories"]) + for subject_category_id in value["subject_categories"]: + self.assertIn(subject_category_id, categories["subject_categories"]) + # TODO: need more tests (set and del) def test_sub_rules(self): - demo_user = self.create_user("demo") - admin_user = self.create_user() - tenant = self.create_tenant() - ref = self.create_intra_extension("policy_authz") - ref_admin = self.create_intra_extension("policy_admin") - self.create_mapping(tenant, ref["id"], ref_admin["id"]) - - sub_meta_rules = self.manager.get_sub_meta_rules_dict(admin_user["id"], ref["id"]) + authz_ie_dict = create_intra_extension(self, "policy_authz") + admin_ie_dict = create_intra_extension(self, "policy_admin") + tenant, mapping = create_mapping(self, "demo", authz_ie_dict['id'], admin_ie_dict['id']) + + admin_subject_id, admin_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'admin').iteritems().next() + demo_subject_dict = self.admin_manager.add_subject_dict(admin_subject_id, admin_ie_dict["id"], + {"name": "demo", "description": "demo"}) + demo_subject_id, demo_subject_dict = \ + self.tenant_api.get_subject_from_keystone_name(tenant['id'], admin_ie_dict['id'], 'demo').iteritems().next() + + sub_meta_rules = self.admin_manager.get_sub_meta_rules_dict(admin_subject_id, authz_ie_dict["id"]) self.assertIsInstance(sub_meta_rules, dict) - self.assertIn("sub_meta_rules", sub_meta_rules) - - sub_rules = self.manager.get_sub_rules(admin_user["id"], ref["id"]) - self.assertIsInstance(sub_rules, dict) - self.assertIn("rules", sub_rules) - rules = dict() - for relation in sub_rules["rules"]: - self.assertIn(relation, self.manager.get_sub_meta_rule_relations(admin_user["id"], ref["id"])["sub_meta_rule_relations"]) - rules[relation] = list() - for rule in sub_rules["rules"][relation]: + + for relation_id in sub_meta_rules: + rules = self.admin_manager.get_rules_dict(admin_subject_id, authz_ie_dict["id"], relation_id) + rule_length = len(sub_meta_rules[relation_id]["subject_categories"]) + \ + len(sub_meta_rules[relation_id]["object_categories"]) + \ + len(sub_meta_rules[relation_id]["action_categories"]) + 1 + for rule_id in rules: + self.assertEqual(rule_length, len(rules[rule_id])) + rule = list(rules[rule_id]) for cat, cat_func, func_name in ( - ("subject_categories", self.manager.get_subject_scopes_dict, "subject_category_scope"), - ("action_categories", self.manager.get_action_scopes_dict, "action_category_scope"), - ("object_categories", self.manager.get_object_scopes_dict, "object_category_scope"), + ("subject_categories", self.admin_manager.get_subject_scopes_dict, "subject_scope"), + ("action_categories", self.admin_manager.get_action_scopes_dict, "action_scope"), + ("object_categories", self.admin_manager.get_object_scopes_dict, "object_scope"), ): - for cat_value in sub_meta_rules["sub_meta_rules"][relation][cat]: + for cat_value in sub_meta_rules[relation_id][cat]: scope = cat_func( - admin_user["id"], - ref["id"], + admin_subject_id, + authz_ie_dict["id"], cat_value ) a_scope = rule.pop(0) - self.assertIn(a_scope, scope[func_name][cat_value]) - - # add a new subrule - - relation = sub_rules["rules"].keys()[0] - sub_rule = [] - for cat, cat_func, func_name in ( - ("subject_categories", self.manager.get_subject_scopes_dict, "subject_category_scope"), - ("action_categories", self.manager.get_action_scopes_dict, "action_category_scope"), - ("object_categories", self.manager.get_object_scopes_dict, "object_category_scope"), - ): - for cat_value in sub_meta_rules["sub_meta_rules"][relation][cat]: - scope = cat_func( - admin_user["id"], - ref["id"], - cat_value - ) - sub_rule.append(scope[func_name][cat_value].keys()[0]) + if type(a_scope) is not bool: + self.assertIn(a_scope, scope.keys()) + + # add a new subrule + + sub_rule = [] + for cat, cat_func, func_name in ( + ("subject_categories", self.admin_manager.get_subject_scopes_dict, "subject_scope"), + ("action_categories", self.admin_manager.get_action_scopes_dict, "action_scope"), + ("object_categories", self.admin_manager.get_object_scopes_dict, "object_scope"), + ): + for cat_value in sub_meta_rules[relation_id][cat]: + scope = cat_func( + admin_subject_id, + authz_ie_dict["id"], + cat_value + ) + sub_rule.append(scope.keys()[0]) + + sub_rule.append(False) + self.assertRaises( + AuthzException, + self.admin_manager.add_rule_dict, + demo_subject_id, authz_ie_dict["id"], relation_id, sub_rule + ) - self.assertRaises( - AuthzException, - self.manager.set_sub_rule, - admin_user["id"], ref["id"], relation, sub_rule) + sub_rules = self.admin_manager.add_rule_dict(admin_subject_id, authz_ie_dict["id"], relation_id, sub_rule) + self.assertIsInstance(sub_rules, dict) + self.assertIn(sub_rule, sub_rules.values()) + + for rule_id, rule_value in sub_rules.iteritems(): + for cat, cat_func, func_name in ( + ("subject_categories", self.admin_manager.get_subject_scopes_dict, "subject_category_scope"), + ("action_categories", self.admin_manager.get_action_scopes_dict, "action_category_scope"), + ("object_categories", self.admin_manager.get_object_scopes_dict, "object_category_scope"), + ): + for cat_value in sub_meta_rules[relation_id][cat]: + scope = cat_func( + admin_subject_id, + authz_ie_dict["id"], + cat_value + ) + a_scope = rule_value.pop(0) + self.assertIn(a_scope, scope.keys()) + # TODO: add test for the delete function |