From: tierno Date: Mon, 23 Mar 2020 14:42:10 +0000 (+0000) Subject: Fix 1025. Do not remove last project reference if it will be deleted X-Git-Tag: v7.1.0rc1~14 X-Git-Url: https://osm.etsi.org/gitweb/?a=commitdiff_plain;h=f5f2e3f9cecc38647a437af7812323a1da7d3f60;p=osm%2FNBI.git Fix 1025. Do not remove last project reference if it will be deleted Change-Id: Ib42405a46cf48f99780da37896830d68feb5a2da Signed-off-by: tierno --- diff --git a/osm_nbi/admin_topics.py b/osm_nbi/admin_topics.py index 9f591fc..9ecb61b 100644 --- a/osm_nbi/admin_topics.py +++ b/osm_nbi/admin_topics.py @@ -320,26 +320,25 @@ class CommonVimWimSdn(BaseTopic): if dry_run: return None - # remove reference from project_read. If not last delete - if session["project_id"]: - for project_id in session["project_id"]: - if project_id in db_content["_admin"]["projects_read"]: - db_content["_admin"]["projects_read"].remove(project_id) - if project_id in db_content["_admin"]["projects_write"]: - db_content["_admin"]["projects_write"].remove(project_id) - else: - db_content["_admin"]["projects_read"].clear() - db_content["_admin"]["projects_write"].clear() - - update_dict = {"_admin.projects_read": db_content["_admin"]["projects_read"], - "_admin.projects_write": db_content["_admin"]["projects_write"] - } - - # check if there are projects referencing it (apart from ANY that means public).... - if db_content["_admin"]["projects_read"] and (len(db_content["_admin"]["projects_read"]) > 1 or - db_content["_admin"]["projects_read"][0] != "ANY"): - self.db.set_one(self.topic, filter_q, update_dict=update_dict) # remove references but not delete - return None + # remove reference from project_read if there are more projects referencing it. If it last one, + # do not remove reference, but order via kafka to delete it + if session["project_id"] and session["project_id"]: + other_projects_referencing = next((p for p in db_content["_admin"]["projects_read"] + if p not in session["project_id"]), None) + + # check if there are projects referencing it (apart from ANY, that means, public).... + if other_projects_referencing: + # remove references but not delete + update_dict_pull = {"_admin.projects_read.{}".format(p): None for p in session["project_id"]} + update_dict_pull.update({"_admin.projects_write.{}".format(p): None for p in session["project_id"]}) + self.db.set_one(self.topic, filter_q, update_dict=None, pull=update_dict_pull) + return None + else: + can_write = next((p for p in db_content["_admin"]["projects_write"] if p == "ANY" or + p in session["project_id"]), None) + if not can_write: + raise EngineException("You have not write permission to delete it", + http_code=HTTPStatus.UNAUTHORIZED) # It must be deleted if session["force"]: @@ -347,7 +346,7 @@ class CommonVimWimSdn(BaseTopic): op_id = None self._send_msg("deleted", {"_id": _id, "op_id": op_id}, not_send_msg=not_send_msg) else: - update_dict["_admin.to_delete"] = True + update_dict = {"_admin.to_delete": True} self.db.set_one(self.topic, {"_id": _id}, update_dict=update_dict, push={"_admin.operations": self._create_operation("delete")} diff --git a/osm_nbi/base_topic.py b/osm_nbi/base_topic.py index 228e29d..e849bb8 100644 --- a/osm_nbi/base_topic.py +++ b/osm_nbi/base_topic.py @@ -152,7 +152,7 @@ class BaseTopic: def _get_project_filter(session): """ Generates a filter dictionary for querying database, so that only allowed items for this project can be - addressed. Only propietary or public can be used. Allowed projects are at _admin.project_read/write. If it is + addressed. Only proprietary or public can be used. Allowed projects are at _admin.project_read/write. If it is not present or contains ANY mean public. :param session: contains: project_id: project list this session has rights to access. Can be empty, one or several @@ -468,30 +468,39 @@ class BaseTopic: """ # To allow addressing projects and users by name AS WELL AS by _id - filter_q = {BaseTopic.id_field(self.topic, _id): _id} + if not self.multiproject: + filter_q = {} + else: + filter_q = self._get_project_filter(session) + filter_q[self.id_field(self.topic, _id)] = _id item_content = self.db.get_one(self.topic, filter_q) - # TODO add admin to filter, validate rights - # data = self.get_item(topic, _id) self.check_conflict_on_del(session, _id, item_content) if dry_run: return None - if self.multiproject: - filter_q.update(self._get_project_filter(session)) if self.multiproject and session["project_id"]: - # remove reference from project_read. If not last delete - # if this topic is not part of session["project_id"] no midification at database is done and an exception - # is raised - self.db.set_one(self.topic, filter_q, update_dict=None, - pull={"_admin.projects_read": {"$in": session["project_id"]}}) - # try to delete if there is not any more reference from projects. Ignore if it is not deleted - filter_q = {'_id': _id, '_admin.projects_read': [[], ["ANY"]]} - v = self.db.del_one(self.topic, filter_q, fail_on_empty=False) - if not v or not v["deleted"]: + # remove reference from project_read if there are more projects referencing it. If it last one, + # do not remove reference, but delete + other_projects_referencing = next((p for p in item_content["_admin"]["projects_read"] + if p not in session["project_id"]), None) + + # check if there are projects referencing it (apart from ANY, that means, public).... + if other_projects_referencing: + # remove references but not delete + update_dict_pull = {"_admin.projects_read.{}".format(p): None for p in session["project_id"]} + update_dict_pull.update({"_admin.projects_write.{}".format(p): None for p in session["project_id"]}) + self.db.set_one(self.topic, filter_q, update_dict=None, pull=update_dict_pull) return None - else: - self.db.del_one(self.topic, filter_q) + else: + can_write = next((p for p in item_content["_admin"]["projects_write"] if p == "ANY" or + p in session["project_id"]), None) + if not can_write: + raise EngineException("You have not write permission to delete it", + http_code=HTTPStatus.UNAUTHORIZED) + + # delete + self.db.del_one(self.topic, filter_q) self.delete_extra(session, _id, item_content, not_send_msg=not_send_msg) self._send_msg("deleted", {"_id": _id}, not_send_msg=not_send_msg) return None diff --git a/osm_nbi/tests/test_admin_topics.py b/osm_nbi/tests/test_admin_topics.py index 596faeb..d8a4e63 100755 --- a/osm_nbi/tests/test_admin_topics.py +++ b/osm_nbi/tests/test_admin_topics.py @@ -592,6 +592,7 @@ 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 @@ -727,10 +728,12 @@ 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") + 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": []} @@ -741,10 +744,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") @@ -752,16 +753,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__': diff --git a/osm_nbi/tests/test_descriptor_topics.py b/osm_nbi/tests/test_descriptor_topics.py index 773131c..d012ac9 100755 --- a/osm_nbi/tests/test_descriptor_topics.py +++ b/osm_nbi/tests/test_descriptor_topics.py @@ -33,13 +33,12 @@ from osm_common.dbbase import DbException import yaml -test_pid = str(uuid4()) test_name = "test-user" -fake_session = {"username": test_name, "project_id": (test_pid,), "method": None, - "admin": True, "force": False, "public": False, "allow_show_user_project_role": True} - db_vnfd_content = yaml.load(db_vnfds_text, Loader=yaml.Loader)[0] db_nsd_content = yaml.load(db_nsds_text, Loader=yaml.Loader)[0] +test_pid = db_vnfd_content["_admin"]["projects_read"][0] +fake_session = {"username": test_name, "project_id": (test_pid,), "method": None, + "admin": True, "force": False, "public": False, "allow_show_user_project_role": True} def norm(str): @@ -443,6 +442,7 @@ class Test_VnfdTopic(TestCase): def test_delete_vnfd(self): did = db_vnfd_content["_id"] self.db.get_one.return_value = db_vnfd_content + p_id = db_vnfd_content["_admin"]["projects_read"][0] with self.subTest(i=1, t='Normal Deletion'): self.db.get_list.return_value = [] self.db.del_one.return_value = {"deleted": 1} @@ -454,7 +454,7 @@ class Test_VnfdTopic(TestCase): self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content") self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID") - self.assertEqual(db_args[1]["_admin.projects_read"], [[], ['ANY']], "Wrong DB filter") + self.assertEqual(db_args[1]["_admin.projects_write.cont"], [p_id, 'ANY'], "Wrong DB filter") db_g1_args = self.db.get_one.call_args[0] self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic") self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID") @@ -464,13 +464,8 @@ class Test_VnfdTopic(TestCase): self.assertEqual(db_gl_calls[1][0][0], "nsds", "Wrong DB topic") self.assertEqual(db_gl_calls[1][0][1]["constituent-vnfd.ANYINDEX.vnfd-id-ref"], db_vnfd_content["id"], "Wrong DB NSD constituent-vnfd id-ref") - db_s1_args = self.db.set_one.call_args - self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic") - self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID") - self.assertIn(test_pid, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter") - self.assertIsNone(db_s1_args[1]["update_dict"], "Wrong DB update dictionary") - self.assertEqual(db_s1_args[1]["pull"]["_admin.projects_read"]["$in"], fake_session["project_id"], - "Wrong DB pull dictionary") + + self.db.set_one.assert_not_called() fs_del_calls = self.fs.file_delete.call_args_list self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id") self.assertEqual(fs_del_calls[1][0][0], did+'_', "Wrong FS folder id") @@ -495,6 +490,28 @@ class Test_VnfdTopic(TestCase): self.topic.delete(fake_session, did) self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code") self.assertIn(norm(excp_msg), norm(str(e.exception)), "Wrong exception text") + with self.subTest(i=5, t='No delete because referenced by other project'): + db_vnfd_content["_admin"]["projects_read"].append("other_project") + self.db.get_one = Mock(return_value=db_vnfd_content) + self.db.get_list = Mock(return_value=[]) + self.msg.write.reset_mock() + self.db.del_one.reset_mock() + self.fs.file_delete.reset_mock() + + self.topic.delete(fake_session, did) + self.db.del_one.assert_not_called() + self.msg.write.assert_not_called() + db_g1_args = self.db.get_one.call_args[0] + self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic") + self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID") + db_s1_args = self.db.set_one.call_args + self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic") + self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID") + self.assertIn(p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter") + self.assertIsNone(db_s1_args[1]["update_dict"], "Wrong DB update dictionary") + self.assertIn("_admin.projects_read." + p_id, db_s1_args[1]["pull"], "Wrong DB pull dictionary") + self.assertIn("_admin.projects_write." + p_id, db_s1_args[1]["pull"], "Wrong DB pull dictionary") + self.fs.file_delete.assert_not_called() return @@ -756,6 +773,7 @@ class Test_NsdTopic(TestCase): def test_delete_nsd(self): did = db_nsd_content["_id"] self.db.get_one.return_value = db_nsd_content + p_id = db_nsd_content["_admin"]["projects_read"][0] with self.subTest(i=1, t='Normal Deletion'): self.db.get_list.return_value = [] self.db.del_one.return_value = {"deleted": 1} @@ -767,7 +785,7 @@ class Test_NsdTopic(TestCase): self.assertEqual(msg_args[2], {"_id": did}, "Wrong message content") self.assertEqual(db_args[0], self.topic.topic, "Wrong DB topic") self.assertEqual(db_args[1]["_id"], did, "Wrong DB ID") - self.assertEqual(db_args[1]["_admin.projects_read"], [[], ['ANY']], "Wrong DB filter") + self.assertEqual(db_args[1]["_admin.projects_write.cont"], [p_id, 'ANY'], "Wrong DB filter") db_g1_args = self.db.get_one.call_args[0] self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic") self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB NSD ID") @@ -777,17 +795,10 @@ class Test_NsdTopic(TestCase): self.assertEqual(db_gl_calls[1][0][0], "nsts", "Wrong DB topic") self.assertEqual(db_gl_calls[1][0][1]["netslice-subnet.ANYINDEX.nsd-ref"], db_nsd_content["id"], "Wrong DB NSD netslice-subnet nsd-ref") - db_s1_args = self.db.set_one.call_args - self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic") - self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID") - self.assertIn(test_pid, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter") - self.assertIsNone(db_s1_args[1]["update_dict"], "Wrong DB update dictionary") - self.assertEqual(db_s1_args[1]["pull"]["_admin.projects_read"]["$in"], fake_session["project_id"], - "Wrong DB pull dictionary") + self.db.set_one.assert_not_called() fs_del_calls = self.fs.file_delete.call_args_list self.assertEqual(fs_del_calls[0][0][0], did, "Wrong FS file id") self.assertEqual(fs_del_calls[1][0][0], did+'_', "Wrong FS folder id") - return # TO REMOVE with self.subTest(i=2, t='Conflict on Delete - NSD in use by nsr'): self.db.get_list.return_value = [{"_id": str(uuid4()), "name": "fake-nsr"}] with self.assertRaises(EngineException, msg="Accepted NSD in use by NSR") as e: @@ -809,6 +820,28 @@ class Test_NsdTopic(TestCase): self.topic.delete(fake_session, did) self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code") self.assertIn(norm(excp_msg), norm(str(e.exception)), "Wrong exception text") + with self.subTest(i=5, t='No delete because referenced by other project'): + db_nsd_content["_admin"]["projects_read"].append("other_project") + self.db.get_one = Mock(return_value=db_nsd_content) + self.db.get_list = Mock(return_value=[]) + self.msg.write.reset_mock() + self.db.del_one.reset_mock() + self.fs.file_delete.reset_mock() + + self.topic.delete(fake_session, did) + self.db.del_one.assert_not_called() + self.msg.write.assert_not_called() + db_g1_args = self.db.get_one.call_args[0] + self.assertEqual(db_g1_args[0], self.topic.topic, "Wrong DB topic") + self.assertEqual(db_g1_args[1]["_id"], did, "Wrong DB VNFD ID") + db_s1_args = self.db.set_one.call_args + self.assertEqual(db_s1_args[0][0], self.topic.topic, "Wrong DB topic") + self.assertEqual(db_s1_args[0][1]["_id"], did, "Wrong DB ID") + self.assertIn(p_id, db_s1_args[0][1]["_admin.projects_write.cont"], "Wrong DB filter") + self.assertIsNone(db_s1_args[1]["update_dict"], "Wrong DB update dictionary") + self.assertIn("_admin.projects_read." + p_id, db_s1_args[1]["pull"], "Wrong DB pull dictionary") + self.assertIn("_admin.projects_write." + p_id, db_s1_args[1]["pull"], "Wrong DB pull dictionary") + self.fs.file_delete.assert_not_called() return diff --git a/osm_nbi/tests/test_instance_topics.py b/osm_nbi/tests/test_instance_topics.py index e6cc254..6d82b0f 100644 --- a/osm_nbi/tests/test_instance_topics.py +++ b/osm_nbi/tests/test_instance_topics.py @@ -233,3 +233,92 @@ class TestNsrTopic(unittest.TestCase): for expect_text in expect_text_list: self.assertIn(expect_text, str(e.exception).lower(), "Expected '{}' at exception text".format(expect_text)) + + def test_delete_ns(self): + self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader)) + self.nsr = self.db.get_list("nsrs")[0] + self.nsr_id = self.nsr["_id"] + self.db_set_one = self.db.set_one + p_id = self.nsd_project + p_other = "other_p" + + session = {"force": False, "admin": False, "public": None, "project_id": [p_id], "method": "delete"} + session2 = {"force": False, "admin": False, "public": None, "project_id": [p_other], "method": "delete"} + session_force = {"force": True, "admin": True, "public": None, "project_id": [], "method": "delete"} + with self.subTest(i=1, t='Normal Deletion'): + self.db.del_one = Mock() + self.db.set_one = Mock() + self.nsr_topic.delete(session, self.nsr_id) + + db_args = self.db.del_one.call_args[0] + msg_args = self.msg.write.call_args[0] + self.assertEqual(msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic") + self.assertEqual(msg_args[1], "deleted", "Wrong message action") + self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content") + self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic") + self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID") + self.assertEqual(db_args[1]["_admin.projects_read.cont"], [p_id], "Wrong DB filter") + self.db.set_one.assert_not_called() + fs_del_calls = self.fs.file_delete.call_args_list + self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id") + with self.subTest(i=2, t='No delete because referenced by other project'): + self.db_set_one("nsrs", {"_id": self.nsr_id}, update_dict=None, push={"_admin.projects_read": p_other, + "_admin.projects_write": p_other}) + self.db.del_one.reset_mock() + self.db.set_one.reset_mock() + self.msg.write.reset_mock() + self.fs.file_delete.reset_mock() + + self.nsr_topic.delete(session2, self.nsr_id) + self.db.del_one.assert_not_called() + self.msg.write.assert_not_called() + db_s1_args = self.db.set_one.call_args + self.assertEqual(db_s1_args[0][0], self.nsr_topic.topic, "Wrong DB topic") + self.assertEqual(db_s1_args[0][1]["_id"], self.nsr_id, "Wrong DB ID") + self.assertIsNone(db_s1_args[1]["update_dict"], "Wrong DB update dictionary") + self.assertIn("_admin.projects_read." + p_other, db_s1_args[1]["pull"], "Wrong DB pull dictionary") + self.assertIn("_admin.projects_write." + p_other, db_s1_args[1]["pull"], "Wrong DB pull dictionary") + self.fs.file_delete.assert_not_called() + with self.subTest(i=4, t='Delete with force and admin'): + self.db.del_one.reset_mock() + self.db.set_one.reset_mock() + self.msg.write.reset_mock() + self.fs.file_delete.reset_mock() + self.nsr_topic.delete(session_force, self.nsr_id) + + db_args = self.db.del_one.call_args[0] + msg_args = self.msg.write.call_args[0] + self.assertEqual(msg_args[0], self.nsr_topic.topic_msg, "Wrong message topic") + self.assertEqual(msg_args[1], "deleted", "Wrong message action") + self.assertEqual(msg_args[2], {"_id": self.nsr_id}, "Wrong message content") + self.assertEqual(db_args[0], self.nsr_topic.topic, "Wrong DB topic") + self.assertEqual(db_args[1]["_id"], self.nsr_id, "Wrong DB ID") + self.db.set_one.assert_not_called() + fs_del_calls = self.fs.file_delete.call_args_list + self.assertEqual(fs_del_calls[0][0][0], self.nsr_id, "Wrong FS file id") + with self.subTest(i=3, t='Conflict on Delete - NS in INSTANTIATED state'): + self.db_set_one("nsrs", {"_id": self.nsr_id}, {"_admin.nsState": "INSTANTIATED"}, + pull={"_admin.projects_read": p_other, "_admin.projects_write": p_other}) + self.db.del_one.reset_mock() + self.db.set_one.reset_mock() + self.msg.write.reset_mock() + self.fs.file_delete.reset_mock() + + with self.assertRaises(EngineException, msg="Accepted NSR with nsState INSTANTIATED") as e: + self.nsr_topic.delete(session, self.nsr_id) + self.assertEqual(e.exception.http_code, HTTPStatus.CONFLICT, "Wrong HTTP status code") + self.assertIn("INSTANTIATED", str(e.exception), "Wrong exception text") + # TODOD with self.subTest(i=3, t='Conflict on Delete - NS in use by NSI'): + + with self.subTest(i=4, t='Non-existent NS'): + self.db.del_one.reset_mock() + self.db.set_one.reset_mock() + self.msg.write.reset_mock() + self.fs.file_delete.reset_mock() + excp_msg = "Not found" + with self.assertRaises(DbException, msg="Accepted non-existent NSD ID") as e: + self.nsr_topic.delete(session2, "other_id") + self.assertEqual(e.exception.http_code, HTTPStatus.NOT_FOUND, "Wrong HTTP status code") + self.assertIn(excp_msg, str(e.exception), "Wrong exception text") + self.assertIn("other_id", str(e.exception), "Wrong exception text") + return