Reformat NBI to standardized format
[osm/NBI.git] / osm_nbi / tests / test_instance_topics.py
index e6cc254..5e86ae5 100644 (file)
@@ -16,7 +16,7 @@
 ##
 
 import unittest
-from unittest.mock import Mock, mock_open   # patch, MagicMock
+from unittest.mock import Mock, mock_open  # patch, MagicMock
 from osm_common.dbbase import DbException
 from osm_nbi.engine import EngineException
 from osm_common.dbmemory import DbMemory
@@ -24,14 +24,18 @@ from osm_common.fsbase import FsBase
 from osm_common.msgbase import MsgBase
 from http import HTTPStatus
 from osm_nbi.instance_topics import NsLcmOpTopic, NsrTopic
-from osm_nbi.tests.test_db_descriptors import db_vim_accounts_text, db_nsds_text, db_vnfds_text, db_nsrs_text,\
-    db_vnfrs_text
+from osm_nbi.tests.test_db_descriptors import (
+    db_vim_accounts_text,
+    db_nsds_text,
+    db_vnfds_text,
+    db_nsrs_text,
+    db_vnfrs_text,
+)
 from copy import deepcopy
 import yaml
 
 
 class TestNsLcmOpTopic(unittest.TestCase):
-
     def setUp(self):
         self.db = DbMemory()
         self.fs = Mock(FsBase())
@@ -42,7 +46,9 @@ class TestNsLcmOpTopic(unittest.TestCase):
         self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None)
         self.nslcmop_topic.check_quota = Mock(return_value=None)  # skip quota
 
-        self.db.create_list("vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader))
+        self.db.create_list(
+            "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
+        )
         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
         self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
         self.db.create_list("vnfrs", yaml.load(db_vnfrs_text, Loader=yaml.Loader))
@@ -59,64 +65,137 @@ class TestNsLcmOpTopic(unittest.TestCase):
         self.vim_id = self.vim["_id"]
 
     def test_create_instantiate(self):
-        session = {"force": False, "admin": False, "public": False, "project_id": [self.nsr_project], "method": "write"}
+        session = {
+            "force": False,
+            "admin": False,
+            "public": False,
+            "project_id": [self.nsr_project],
+            "method": "write",
+        }
         indata = {
             "nsdId": self.nsd_id,
             "nsInstanceId": self.nsr_id,
             "nsName": "name",
             "vimAccountId": self.vim_id,
-            "additionalParamsForVnf": [{"member-vnf-index": "1", "additionalParams": {"touch_filename": "file"}},
-                                       {"member-vnf-index": "2", "additionalParams": {"touch_filename": "file"}}],
-            "vnf": [{"member-vnf-index": "1",
-                     "vdu": [{"id": "dataVM", "interface": [{"name": "dataVM-eth0",
-                                                             "ip-address": "10.11.12.13",
-                                                             "floating-ip-required": True}]
-                              }],
-                     "internal-vld": [{"name": "internal", "vim-network-id": "vim-net-id"}]
-                     }],
+            "additionalParamsForVnf": [
+                {
+                    "member-vnf-index": "1",
+                    "additionalParams": {"touch_filename": "file"},
+                },
+                {
+                    "member-vnf-index": "2",
+                    "additionalParams": {"touch_filename": "file"},
+                },
+            ],
+            "vnf": [
+                {
+                    "member-vnf-index": "1",
+                    "vdu": [
+                        {
+                            "id": "dataVM",
+                            "interface": [
+                                {
+                                    "name": "dataVM-eth0",
+                                    "ip-address": "10.11.12.13",
+                                    "floating-ip-required": True,
+                                }
+                            ],
+                        }
+                    ],
+                    "internal-vld": [
+                        {"name": "internal", "vim-network-id": "vim-net-id"}
+                    ],
+                }
+            ],
             "lcmOperationType": "instantiate",
-
         }
         rollback = []
         headers = {}
 
-        nslcmop_id, _ = self.nslcmop_topic.new(rollback, session, indata=deepcopy(indata), kwargs=None, headers=headers)
+        nslcmop_id, _ = self.nslcmop_topic.new(
+            rollback, session, indata=deepcopy(indata), kwargs=None, headers=headers
+        )
 
         # check nslcmop is created at database
-        self.assertEqual(self.db.create.call_count, 1, "database create not called, or called more than once")
+        self.assertEqual(
+            self.db.create.call_count,
+            1,
+            "database create not called, or called more than once",
+        )
         _call = self.db.create.call_args_list[0]
-        self.assertEqual(_call[0][0], "nslcmops", "must be create a nslcmops entry at database")
+        self.assertEqual(
+            _call[0][0], "nslcmops", "must be create a nslcmops entry at database"
+        )
 
         created_nslcmop = _call[0][1]
-        self.assertEqual(nslcmop_id, created_nslcmop["_id"], "mismatch between return id and database '_id'")
-        self.assertEqual(self.nsr_id, created_nslcmop["nsInstanceId"], "bad reference id from nslcmop to nsr")
-        self.assertTrue(created_nslcmop["_admin"].get("projects_read"),
-                        "Database record must contain '_amdin.projects_read'")
-        self.assertIn("created", created_nslcmop["_admin"], "Database record must contain '_admin.created'")
-        self.assertTrue(created_nslcmop["lcmOperationType"] == "instantiate",
-                        "Database record must contain 'lcmOperationType=instantiate'")
+        self.assertEqual(
+            nslcmop_id,
+            created_nslcmop["_id"],
+            "mismatch between return id and database '_id'",
+        )
+        self.assertEqual(
+            self.nsr_id,
+            created_nslcmop["nsInstanceId"],
+            "bad reference id from nslcmop to nsr",
+        )
+        self.assertTrue(
+            created_nslcmop["_admin"].get("projects_read"),
+            "Database record must contain '_amdin.projects_read'",
+        )
+        self.assertIn(
+            "created",
+            created_nslcmop["_admin"],
+            "Database record must contain '_admin.created'",
+        )
+        self.assertTrue(
+            created_nslcmop["lcmOperationType"] == "instantiate",
+            "Database record must contain 'lcmOperationType=instantiate'",
+        )
 
-        self.assertEqual(len(rollback), len(self.db.set_one.call_args_list) + 1,
-                         "rollback mismatch with created/set items at database")
+        self.assertEqual(
+            len(rollback),
+            len(self.db.set_one.call_args_list) + 1,
+            "rollback mismatch with created/set items at database",
+        )
 
         # test parameters with error
         bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
         test_set = (
-            ("nsr not found", {"nsInstanceId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),
+            (
+                "nsr not found",
+                {"nsInstanceId": bad_id},
+                DbException,
+                HTTPStatus.NOT_FOUND,
+                ("not found", bad_id),
+            ),
             # TODO add "nsd"
             # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),  # TODO add "vim"
-            ("bad member-vnf-index", {"vnf.0.member-vnf-index": "k"}, EngineException, HTTPStatus.BAD_REQUEST,
-             ("k",)),
+            (
+                "bad member-vnf-index",
+                {"vnf.0.member-vnf-index": "k"},
+                EngineException,
+                HTTPStatus.BAD_REQUEST,
+                ("k",),
+            ),
         )
         for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
             with self.assertRaises(expect_exc, msg=message) as e:
-                self.nslcmop_topic.new(rollback, session, indata=deepcopy(indata), kwargs=kwargs_, headers=headers)
+                self.nslcmop_topic.new(
+                    rollback,
+                    session,
+                    indata=deepcopy(indata),
+                    kwargs=kwargs_,
+                    headers=headers,
+                )
             if expect_code:
                 self.assertTrue(e.exception.http_code == expect_code)
             if expect_text_list:
                 for expect_text in expect_text_list:
-                    self.assertIn(expect_text, str(e.exception).lower(),
-                                  "Expected '{}' at exception text".format(expect_text))
+                    self.assertIn(
+                        expect_text,
+                        str(e.exception).lower(),
+                        "Expected '{}' at exception text".format(expect_text),
+                    )
 
     def test_check_ns_operation_action(self):
         nsrs = self.db.get_list("nsrs")[0]
@@ -126,7 +205,7 @@ class TestNsLcmOpTopic(unittest.TestCase):
             "member_vnf_index": "1",
             "vdu_id": None,
             "primitive": "touch",
-            "primitive_params": {"filename": "file"}
+            "primitive_params": {"filename": "file"},
         }
 
         self.nslcmop_topic._check_ns_operation(session, nsrs, "action", indata)
@@ -136,14 +215,18 @@ class TestNsLcmOpTopic(unittest.TestCase):
                 continue
             indata_copy[k] = "non_existing"
             with self.assertRaises(EngineException) as exc_manager:
-                self.nslcmop_topic._check_ns_operation(session, nsrs, "action", indata_copy)
+                self.nslcmop_topic._check_ns_operation(
+                    session, nsrs, "action", indata_copy
+                )
             exc = exc_manager.exception
-            self.assertEqual(exc.http_code, HTTPStatus.BAD_REQUEST, "Engine exception bad http_code with {}".
-                             format(indata_copy))
+            self.assertEqual(
+                exc.http_code,
+                HTTPStatus.BAD_REQUEST,
+                "Engine exception bad http_code with {}".format(indata_copy),
+            )
 
 
 class TestNsrTopic(unittest.TestCase):
-
     def setUp(self):
         self.db = DbMemory()
         self.fs = Mock(FsBase())
@@ -154,7 +237,9 @@ class TestNsrTopic(unittest.TestCase):
         self.nsr_topic = NsrTopic(self.db, self.fs, self.msg, None)
         self.nsr_topic.check_quota = Mock(return_value=None)  # skip quota
 
-        self.db.create_list("vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader))
+        self.db.create_list(
+            "vim_accounts", yaml.load(db_vim_accounts_text, Loader=yaml.Loader)
+        )
         self.db.create_list("nsds", yaml.load(db_nsds_text, Loader=yaml.Loader))
         self.db.create_list("vnfds", yaml.load(db_vnfds_text, Loader=yaml.Loader))
         self.db.create = Mock(return_value="created_id")
@@ -166,18 +251,34 @@ class TestNsrTopic(unittest.TestCase):
         self.vim_id = self.vim["_id"]
 
     def test_create(self):
-        session = {"force": False, "admin": False, "public": False, "project_id": [self.nsd_project], "method": "write"}
+        session = {
+            "force": False,
+            "admin": False,
+            "public": False,
+            "project_id": [self.nsd_project],
+            "method": "write",
+        }
         indata = {
             "nsdId": self.nsd_id,
             "nsName": "name",
             "vimAccountId": self.vim_id,
-            "additionalParamsForVnf": [{"member-vnf-index": "1", "additionalParams": {"touch_filename": "file"}},
-                                       {"member-vnf-index": "2", "additionalParams": {"touch_filename": "file"}}]
+            "additionalParamsForVnf": [
+                {
+                    "member-vnf-index": "hackfest_vnf1",
+                    "additionalParams": {"touch_filename": "file"},
+                },
+                {
+                    "member-vnf-index": "hackfest_vnf2",
+                    "additionalParams": {"touch_filename": "file"},
+                },
+            ],
         }
         rollback = []
         headers = {}
 
-        self.nsr_topic.new(rollback, session, indata=indata, kwargs=None, headers=headers)
+        self.nsr_topic.new(
+            rollback, session, indata=indata, kwargs=None, headers=headers
+        )
 
         # check vnfrs and nsrs created in whatever order
         created_vnfrs = []
@@ -188,48 +289,238 @@ class TestNsrTopic(unittest.TestCase):
             created_item = _call[0][1]
             if _call[0][0] == "vnfrs":
                 created_vnfrs.append(created_item)
-                self.assertIn("member-vnf-index-ref", created_item,
-                              "Created item must contain member-vnf-index-ref section")
+                self.assertIn(
+                    "member-vnf-index-ref",
+                    created_item,
+                    "Created item must contain member-vnf-index-ref section",
+                )
                 if nsr_id:
-                    self.assertEqual(nsr_id, created_item["nsr-id-ref"], "bad reference id from vnfr to nsr")
+                    self.assertEqual(
+                        nsr_id,
+                        created_item["nsr-id-ref"],
+                        "bad reference id from vnfr to nsr",
+                    )
                 else:
                     nsr_id = created_item["nsr-id-ref"]
 
             elif _call[0][0] == "nsrs":
                 created_nsrs.append(created_item)
                 if nsr_id:
-                    self.assertEqual(nsr_id, created_item["_id"], "bad reference id from vnfr to nsr")
+                    self.assertEqual(
+                        nsr_id, created_item["_id"], "bad reference id from vnfr to nsr"
+                    )
                 else:
                     nsr_id = created_item["_id"]
             else:
-                assert True, "created an unknown record {} at database".format(_call[0][0])
+                assert True, "created an unknown record {} at database".format(
+                    _call[0][0]
+                )
 
-            self.assertTrue(created_item["_admin"].get("projects_read"),
-                            "Database record must contain '_amdin.projects_read'")
-            self.assertIn("created", created_item["_admin"], "Database record must contain '_admin.created'")
-            self.assertTrue(created_item["_admin"]["nsState"] == "NOT_INSTANTIATED",
-                            "Database record must contain '_admin.nstate=NOT INSTANTIATE'")
+            self.assertTrue(
+                created_item["_admin"].get("projects_read"),
+                "Database record must contain '_amdin.projects_read'",
+            )
+            self.assertIn(
+                "created",
+                created_item["_admin"],
+                "Database record must contain '_admin.created'",
+            )
+            self.assertTrue(
+                created_item["_admin"]["nsState"] == "NOT_INSTANTIATED",
+                "Database record must contain '_admin.nstate=NOT INSTANTIATE'",
+            )
 
-        self.assertEqual(len(created_vnfrs), len(self.nsd["constituent-vnfd"]),
-                         "created a mismatch number of vnfr at database")
-        self.assertEqual(len(created_nsrs), 1, "Only one nsrs must be created at database")
-        self.assertEqual(len(rollback), len(created_vnfrs) + 1, "rollback mismatch with created items at database")
+        self.assertEqual(
+            len(created_vnfrs), 2, "created a mismatch number of vnfr at database"
+        )
+        self.assertEqual(
+            len(created_nsrs), 1, "Only one nsrs must be created at database"
+        )
+        self.assertEqual(
+            len(rollback),
+            len(created_vnfrs) + 1,
+            "rollback mismatch with created items at database",
+        )
 
         # test parameters with error
         bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa"
         test_set = (
             # TODO add "nsd"
-            ("nsd not found", {"nsdId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),
+            (
+                "nsd not found",
+                {"nsdId": bad_id},
+                DbException,
+                HTTPStatus.NOT_FOUND,
+                ("not found", bad_id),
+            ),
             # ({"vimAccountId": bad_id}, DbException, HTTPStatus.NOT_FOUND, ("not found", bad_id)),  # TODO add "vim"
-            ("additional params not supply", {"additionalParamsForVnf.0.member-vnf-index": "k"}, EngineException,
-             HTTPStatus.BAD_REQUEST, None),
+            (
+                "additional params not supply",
+                {"additionalParamsForVnf.0.member-vnf-index": "k"},
+                EngineException,
+                HTTPStatus.BAD_REQUEST,
+                None,
+            ),
         )
         for message, kwargs_, expect_exc, expect_code, expect_text_list in test_set:
             with self.assertRaises(expect_exc, msg=message) as e:
-                self.nsr_topic.new(rollback, session, indata=deepcopy(indata), kwargs=kwargs_, headers=headers)
+                self.nsr_topic.new(
+                    rollback,
+                    session,
+                    indata=deepcopy(indata),
+                    kwargs=kwargs_,
+                    headers=headers,
+                )
             if expect_code:
                 self.assertTrue(e.exception.http_code == expect_code)
             if expect_text_list:
                 for expect_text in expect_text_list:
-                    self.assertIn(expect_text, str(e.exception).lower(),
-                                  "Expected '{}' at exception text".format(expect_text))
+                    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_ro_nsrs = self.db.del_one.call_args_list[1][0]
+            db_args = self.db.del_one.call_args_list[0][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_ro_nsrs[0], "ro_nsrs", "Wrong DB topic")
+            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.assertEqual(
+                db_s1_args[1]["pull_list"],
+                {"_admin.projects_read": [p_other], "_admin.projects_write": [p_other]},
+                "Wrong DB pull_list 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_ro_nsrs = self.db.del_one.call_args_list[1][0]
+            db_args = self.db.del_one.call_args_list[0][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_ro_nsrs[0], "ro_nsrs", "Wrong DB topic")
+            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