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"]:
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")}
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
"""
# 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
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.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": []}
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")
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__':
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):
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}
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")
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")
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
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}
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")
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:
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
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
projects / osm / NBI.git / commitdiff