X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FNBI.git;a=blobdiff_plain;f=osm_nbi%2Ftests%2Ftest_admin_topics.py;h=dce871058e9dc65ec01e1c506f083d0933edf5d0;hp=5597979b4607c3736b18b35e7b109ce04e97398c;hb=refs%2Fchanges%2F30%2F9030%2F2;hpb=b3d0a0e94cc4572d81477754baf56c910e892be3 diff --git a/osm_nbi/tests/test_admin_topics.py b/osm_nbi/tests/test_admin_topics.py index 5597979..dce8710 100755 --- a/osm_nbi/tests/test_admin_topics.py +++ b/osm_nbi/tests/test_admin_topics.py @@ -54,6 +54,7 @@ class Test_ProjectTopicAuth(TestCase): self.topic = ProjectTopicAuth(self.db, self.fs, self.msg, self.auth) self.fake_session = {"username": self.test_name, "project_id": (test_pid,), "method": None, "admin": True, "force": False, "public": False, "allow_show_user_project_role": True} + self.topic.check_quota = Mock(return_value=None) # skip quota def test_new_project(self): with self.subTest(i=1): @@ -211,9 +212,11 @@ class Test_RoleTopicAuth(TestCase): self.fs = Mock(fsbase.FsBase()) self.msg = Mock(msgbase.MsgBase()) self.auth = Mock(authconn.Authconn(None, None, None)) - self.topic = RoleTopicAuth(self.db, self.fs, self.msg, self.auth, self.test_operations) + self.auth.role_permissions = self.test_operations + self.topic = RoleTopicAuth(self.db, self.fs, self.msg, self.auth) self.fake_session = {"username": test_name, "project_id": (test_pid,), "method": None, "admin": True, "force": False, "public": False, "allow_show_user_project_role": True} + self.topic.check_quota = Mock(return_value=None) # skip quota def test_new_role(self): with self.subTest(i=1): @@ -374,6 +377,7 @@ class Test_UserTopicAuth(TestCase): self.topic = UserTopicAuth(self.db, self.fs, self.msg, self.auth) self.fake_session = {"username": test_name, "project_id": (test_pid,), "method": None, "admin": True, "force": False, "public": False, "allow_show_user_project_role": True} + self.topic.check_quota = Mock(return_value=None) # skip quota def test_new_user(self): uid1 = str(uuid4()) @@ -591,11 +595,13 @@ class Test_CommonVimWimSdn(TestCase): self.auth = Mock(authconn.Authconn(None, None, None)) self.topic = CommonVimWimSdn(self.db, self.fs, self.msg, self.auth) # Use WIM schemas for testing because they are the simplest + self.topic._send_msg = Mock() self.topic.topic = "wims" self.topic.schema_new = validation.wim_account_new_schema self.topic.schema_edit = validation.wim_account_edit_schema self.fake_session = {"username": test_name, "project_id": (test_pid,), "method": None, "admin": True, "force": False, "public": False, "allow_show_user_project_role": True} + self.topic.check_quota = Mock(return_value=None) # skip quota def test_new_cvws(self): test_url = "http://0.0.0.0:0" @@ -716,6 +722,7 @@ class Test_CommonVimWimSdn(TestCase): ro_pid = str(uuid4()) rw_pid = str(uuid4()) cvws = {"_id": cid, "name": self.test_name} + self.db.get_list.return_value = [] with self.subTest(i=1): cvws["_admin"] = {"projects_read": [test_pid, ro_pid, rw_pid], "projects_write": [test_pid, rw_pid]} self.db.get_one.return_value = cvws @@ -725,11 +732,13 @@ class Test_CommonVimWimSdn(TestCase): self.assertEqual(self.db.get_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier") self.assertEqual(self.db.set_one.call_args[0][0], self.topic.topic, "Wrong topic") self.assertEqual(self.db.set_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier") - self.assertEqual(self.db.set_one.call_args[1]["update_dict"]["_admin.projects_read"], [ro_pid, rw_pid], + self.assertEqual(self.db.set_one.call_args[1]["update_dict"], None, "Wrong read-only projects update") - self.assertEqual(self.db.set_one.call_args[1]["update_dict"]["_admin.projects_write"], [rw_pid], + self.assertEqual(self.db.set_one.call_args[1]["pull"], {"_admin.projects_read." + test_pid: None, + "_admin.projects_write." + test_pid: None}, "Wrong read/write projects update") - with self.subTest(i=3): + self.topic._send_msg.assert_not_called() + with self.subTest(i=2): now = time() cvws["_admin"] = {"projects_read": [test_pid], "projects_write": [test_pid], "operations": []} self.db.get_one.return_value = cvws @@ -739,10 +748,8 @@ class Test_CommonVimWimSdn(TestCase): self.assertEqual(self.db.get_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier") self.assertEqual(self.db.set_one.call_args[0][0], self.topic.topic, "Wrong topic") self.assertEqual(self.db.set_one.call_args[0][1]["_id"], cid, "Wrong user identifier") - update_dict = self.db.set_one.call_args[1]["update_dict"] - self.assertEqual(update_dict["_admin.projects_read"], [], "Wrong read-only projects update") - self.assertEqual(update_dict["_admin.projects_write"], [], "Wrong read/write projects update") - self.assertEqual(update_dict["_admin.to_delete"], True, "Wrong deletion mark") + self.assertEqual(self.db.set_one.call_args[1]["update_dict"], {"_admin.to_delete": True}, + "Wrong _admin.to_delete update") operation = self.db.set_one.call_args[1]["push"]["_admin.operations"] self.assertEqual(operation["lcmOperationType"], "delete", "Wrong operation type") self.assertEqual(operation["operationState"], "PROCESSING", "Wrong operation state") @@ -750,16 +757,23 @@ class Test_CommonVimWimSdn(TestCase): self.assertIsNone(operation["operationParams"], "Wrong operation parameters") self.assertGreater(operation["startTime"], now, "Wrong operation start time") self.assertGreater(operation["statusEnteredTime"], now, "Wrong operation status enter time") + self.topic._send_msg.assert_called_once_with("delete", {"_id": cid, "op_id": cid + ":0"}, not_send_msg=None) with self.subTest(i=3): cvws["_admin"] = {"projects_read": [], "projects_write": [], "operations": []} self.db.get_one.return_value = cvws + self.topic._send_msg.reset_mock() + self.db.get_one.reset_mock() + self.db.del_one.reset_mock() self.fake_session["force"] = True # to force deletion + self.fake_session["admin"] = True # to force deletion + self.fake_session["project_id"] = [] # to force deletion oid = self.topic.delete(self.fake_session, cid) self.assertIsNone(oid, "Wrong operation identifier") self.assertEqual(self.db.get_one.call_args[0][0], self.topic.topic, "Wrong topic") self.assertEqual(self.db.get_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier") self.assertEqual(self.db.del_one.call_args[0][0], self.topic.topic, "Wrong topic") self.assertEqual(self.db.del_one.call_args[0][1]["_id"], cid, "Wrong CIM identifier") + self.topic._send_msg.assert_called_once_with("deleted", {"_id": cid, "op_id": None}, not_send_msg=None) if __name__ == '__main__':