From: Patricia Reinoso Date: Wed, 26 Oct 2022 08:56:47 +0000 (+0000) Subject: Adding PaaS Service Creation UTs X-Git-Url: https://osm.etsi.org/gitweb/?a=commitdiff_plain;h=refs%2Fchanges%2F20%2F12620%2F2;p=osm%2FNBI.git Adding PaaS Service Creation UTs Change-Id: I33660a3f08679d902fc1165eb6f0913d5f1f7a5b Signed-off-by: Patricia Reinoso --- diff --git a/osm_nbi/tests/test_db_descriptors.py b/osm_nbi/tests/test_db_descriptors.py index 1914190..3c11a61 100644 --- a/osm_nbi/tests/test_db_descriptors.py +++ b/osm_nbi/tests/test_db_descriptors.py @@ -53,6 +53,62 @@ db_vim_accounts_text = """ vim_user: osm """ +db_paas_accounts_text = """ +--- +- _id: 3d93ca4e-b007-4d94-bbdc-61911078b864 + name: my_new_paas1 + paas_type: juju + endpoints: + - 127.0.0.1 + user: admin + secret: SCgY5peOGf98oqc00TNxfQ== + _admin: + created: 1664533097.4940042 + modified: 1664533116.9955606 + projects_read: + - 25b5aebf-3da1-49ed-99de-1d2b4a86d6e4 + projects_write: + - 25b5aebf-3da1-49ed-99de-1d2b4a86d6e4 + operationalState: ENABLED + operations: + - lcmOperationType: create + operationState: COMPLETED + startTime: 1664533097.4943297 + statusEnteredTime: 1664533097.4943297 + detailed-status: PaaS validated + operationParams: null + worker: null + current_operation: null + detailed-status: "Connectivity: ok" + schema_version: 1.11 +- _id: c2538499-4c30-41c0-acd5-80cb92f45555 + name: my_new_paas2 + paas_type: juju + endpoints: + - 127.0.0.1 + user: admin + secret: 5g0yGX86qIhprX86YTMcpg== + _admin: + created: 1664533097.4940042 + modified: 1664533116.9955606 + projects_read: + - 25b5aebf-3da1-49ed-99de-1d2b4a86d6e4 + projects_write: + - 25b5aebf-3da1-49ed-99de-1d2b4a86d6e4 + operationalState: ENABLED + operations: + - lcmOperationType: create + operationState: COMPLETED + startTime: 1664533097.4943297 + statusEnteredTime: 1664533097.4943297 + detailed-status: PaaS validated + operationParams: null + worker: null + current_operation: null + detailed-status: "Connectivity: ok" + schema_version: 1.11 +""" + db_vnfds_text = """ --- - _admin: diff --git a/osm_nbi/tests/test_instance_topics.py b/osm_nbi/tests/test_instance_topics.py index 2c10632..908eab6 100644 --- a/osm_nbi/tests/test_instance_topics.py +++ b/osm_nbi/tests/test_instance_topics.py @@ -18,7 +18,7 @@ from contextlib import contextmanager import unittest from time import time -from unittest.mock import Mock, mock_open # patch, MagicMock +from unittest.mock import Mock, mock_open from osm_common.dbbase import DbException from osm_nbi.engine import EngineException from osm_common.dbmemory import DbMemory @@ -29,6 +29,7 @@ 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_paas_accounts_text, db_nsds_text, db_vnfds_text, db_nsrs_text, @@ -65,16 +66,33 @@ class TestNsLcmOpTopic(unittest.TestCase): self.vim = self.db.get_list("vim_accounts")[0] self.vim_id = self.vim["_id"] - - def test_create_instantiate(self): - self.db.set_one = Mock(return_value={"updated": 1}) - session = { + self.db.create_list( + "paas", yaml.load(db_paas_accounts_text, Loader=yaml.Loader) + ) + self.paas_ids = [ + paas_element["_id"] for paas_element in self.db.get_list("paas") + ] + self.paas_id = self.paas_ids[0] + self.vnfrs = { + vnfr["_id"]: vnfr["member-vnf-index-ref"] + for vnfr in self.db.get_list("vnfrs") + } + self.number_of_vnfr = len(self.vnfrs) + self.new_session = { "force": False, "admin": False, "public": False, "project_id": [self.nsr_project], "method": "write", } + + def check_operation_params(self, operation_params): + self.assertEqual(self.nsd_id, operation_params["nsdId"]) + self.assertEqual(self.nsr_id, operation_params["nsInstanceId"]) + self.assertEqual("name", operation_params["nsName"]) + + def test_create_instantiate_with_vim_account(self): + self.db.set_one = Mock(return_value={"updated": 1}) indata = { "nsdId": self.nsd_id, "nsInstanceId": self.nsr_id, @@ -116,7 +134,11 @@ class TestNsLcmOpTopic(unittest.TestCase): headers = {} nslcmop_id, _ = self.nslcmop_topic.new( - rollback, session, indata=deepcopy(indata), kwargs=None, headers=headers + rollback, + self.new_session, + indata=deepcopy(indata), + kwargs=None, + headers=headers, ) # check nslcmop is created at database @@ -161,6 +183,20 @@ class TestNsLcmOpTopic(unittest.TestCase): "rollback mismatch with created/set items at database", ) + operation_params = created_nslcmop["operationParams"] + self.check_operation_params(operation_params) + self.assertEqual(self.vim_id, operation_params["vimAccountId"]) + self.assertNotIn("paasAccountId", operation_params) + + self.assertEqual(len(self.db.set_one.call_args_list), self.number_of_vnfr) + + for call in self.db.set_one.call_args_list: + topic, vnfr_id, update = call[0] + self.assertEqual(topic, "vnfrs") + self.assertIn(vnfr_id["_id"], self.vnfrs) + self.assertEqual(self.vim_id, update["vim-account-id"]) + self.assertNotIn("paas-account-id", update) + # test parameters with error bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa" test_set = ( @@ -185,7 +221,7 @@ class TestNsLcmOpTopic(unittest.TestCase): with self.assertRaises(expect_exc, msg=message) as e: self.nslcmop_topic.new( rollback, - session, + self.new_session, indata=deepcopy(indata), kwargs=kwargs_, headers=headers, @@ -200,6 +236,141 @@ class TestNsLcmOpTopic(unittest.TestCase): "Expected '{}' at exception text".format(expect_text), ) + def test_create_instantiate_with_paas_account(self): + self.db.set_one = Mock(return_value={"updated": 1}) + indata = { + "nsdId": self.nsd_id, + "nsInstanceId": self.nsr_id, + "nsName": "name", + "paasAccountId": self.paas_id, + "lcmOperationType": "instantiate", + } + + nslcmop_id, _ = self.nslcmop_topic.new( + [], self.new_session, indata=deepcopy(indata), kwargs=None, headers={} + ) + + # check nslcmop is created at database + 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" + ) + + created_nslcmop = _call[0][1] + self.assertEqual( + nslcmop_id, + created_nslcmop["_id"], + "mismatch between return id and database '_id'", + ) + + operation_params = created_nslcmop["operationParams"] + self.check_operation_params(operation_params) + self.assertEqual(self.paas_id, operation_params["paasAccountId"]) + self.assertNotIn("vimAccountId", operation_params) + + self.assertEqual(len(self.db.set_one.call_args_list), self.number_of_vnfr) + + for call in self.db.set_one.call_args_list: + topic, vnfr_id, update = call[0] + self.assertEqual(topic, "vnfrs") + self.assertIn(vnfr_id["_id"], self.vnfrs) + self.assertEqual(self.paas_id, update["paas-account-id"]) + self.assertNotIn("vim-account-id", update) + + def test_create_instantiate_invalid_paas_account_raises_exception(self): + self.db.set_one = Mock(return_value={"updated": 1}) + invalid_paas_id = "88d90b0c-faff-4b9f-bccd-017f33985984" + indata = { + "nsdId": self.nsd_id, + "nsInstanceId": self.nsr_id, + "nsName": "name", + "paasAccountId": invalid_paas_id, + "lcmOperationType": "instantiate", + } + + with self.assertRaises(EngineException): + nslcmop_id, _ = self.nslcmop_topic.new( + [], self.new_session, indata=deepcopy(indata), kwargs=None, headers={} + ) + self.db.set_one.assert_not_called() + + def test_create_instantiate_with_paas_account_in_vnf(self): + self.db.set_one = Mock(return_value={"updated": 1}) + indata = { + "nsdId": self.nsd_id, + "nsInstanceId": self.nsr_id, + "nsName": "name", + "paasAccountId": self.paas_ids[0], + "lcmOperationType": "instantiate", + "vnf": [{"member-vnf-index": "1", "paasAccountId": self.paas_ids[1]}], + } + + nslcmop_id, _ = self.nslcmop_topic.new( + [], self.new_session, indata=deepcopy(indata), kwargs=None, headers={} + ) + + # check nslcmop is created at database + 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" + ) + + created_nslcmop = _call[0][1] + self.assertEqual( + nslcmop_id, + created_nslcmop["_id"], + "mismatch between return id and database '_id'", + ) + + operation_params = created_nslcmop["operationParams"] + self.check_operation_params(operation_params) + self.assertEqual(self.paas_id, operation_params["paasAccountId"]) + self.assertNotIn("vimAccountId", operation_params) + expected_paas_id = "" + + self.assertEqual(len(self.db.set_one.call_args_list), self.number_of_vnfr) + for call in self.db.set_one.call_args_list: + topic, vnfr_id, update = call[0] + self.assertEqual(topic, "vnfrs") + self.assertIn(vnfr_id["_id"], self.vnfrs) + vnf_index_ref = self.vnfrs[vnfr_id["_id"]] + if vnf_index_ref == "1": + expected_paas_id = self.paas_ids[1] + elif vnf_index_ref == "2": + expected_paas_id = self.paas_ids[0] + else: + assert False + self.assertEqual(expected_paas_id, update["paas-account-id"]) + self.assertNotIn("vim-account-id", update) + + def test_create_instantiate_invalid_paas_account_in_vnf_raises_exception(self): + self.db.set_one = Mock(return_value={"updated": 1}) + invalid_paas_id = "88d90b0c-faff-4b9f-bccd-017f33985984" + indata = { + "nsdId": self.nsd_id, + "nsInstanceId": self.nsr_id, + "nsName": "name", + "paasAccountId": self.paas_ids[0], + "lcmOperationType": "instantiate", + "vnf": [{"member-vnf-index": "1", "paasAccountId": invalid_paas_id}], + } + + with self.assertRaises(EngineException): + self.nslcmop_topic.new( + [], self.new_session, indata=deepcopy(indata), kwargs=None, headers={} + ) + self.db.set_one.assert_not_called() + def test_check_ns_operation_action(self): nsrs = self.db.get_list("nsrs")[0] session = {} @@ -232,15 +403,13 @@ class TestNsLcmOpTopic(unittest.TestCase): vnfr_id = self.db.get_list("vnfrs")[0]["_id"] session = {} self.db.set_one( - "nsrs", - {"_id": self.nsr_id}, - {"_admin.nsState": "INSTANTIATED"}, + "nsrs", {"_id": self.nsr_id}, {"_admin.nsState": "INSTANTIATED"} ) indata = { "lcmOperationType": "update", "updateType": "REMOVE_VNF", "nsInstanceId": self.nsr_id, - "removeVnfInstanceId": vnfr_id + "removeVnfInstanceId": vnfr_id, } session = { @@ -282,12 +451,9 @@ class TestNsLcmOpTopic(unittest.TestCase): ) def test_migrate(self): - vnfr_id = self.db.get_list("vnfrs")[0]["_id"] session = {} self.db.set_one( - "nsrs", - {"_id": self.nsr_id}, - {"_admin.nsState": "INSTANTIATED"}, + "nsrs", {"_id": self.nsr_id}, {"_admin.nsState": "INSTANTIATED"} ) session = { "force": False, @@ -303,12 +469,9 @@ class TestNsLcmOpTopic(unittest.TestCase): indata = { "lcmOperationType": "migrate", "nsInstanceId": self.nsr_id, - "migrateToHost":"sample02", - "vdu": { - "vduCountIndex": 0, - "vduId": "mgmtVM" - }, - "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f" + "migrateToHost": "sample02", + "vdu": {"vduCountIndex": 0, "vduId": "mgmtVM"}, + "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f", } nslcmop_id, _ = self.nslcmop_topic.new( rollback, session, indata, kwargs=None, headers=headers @@ -337,7 +500,7 @@ class TestNsLcmOpTopic(unittest.TestCase): indata = { "lcmOperationType": "migrate", "nsInstanceId": self.nsr_id, - "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f" + "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f", } nslcmop_id, _ = self.nslcmop_topic.new( rollback, session, indata, kwargs=None, headers=headers @@ -366,18 +529,19 @@ class TestNsLcmOpTopic(unittest.TestCase): indata = { "lcmOperationType": "migrate", "nsInstanceId": self.nsr_id, - "migrateToHost":"sample02", - "vdu": { - "vduCountIndex": 0 - }, - "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f" + "migrateToHost": "sample02", + "vdu": {"vduCountIndex": 0}, + "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f", } with self.assertRaises(Exception) as e: nslcmop_id, _ = self.nslcmop_topic.new( - rollback, session, indata, kwargs=None, headers=headers + rollback, session, indata, kwargs=None, headers=headers + ) + self.assertTrue( + "Format error at 'vdu' ''vduId' is a required property'" + in str(e.exception) ) - self.assertTrue("Format error at 'vdu' ''vduId' is a required property'" in str(e.exception)) class TestNsLcmOpTopicWithMock(unittest.TestCase): @@ -394,18 +558,34 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0] test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader) self.db.get_one.side_effect = [test_vnfr, test_vnfd] - vnfr = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr['_id']) - self.assertEqual(self.db.get_one.call_args_list[0][0][0], 'vnfrs', "Incorrect first DB lookup") - self.assertEqual(self.db.get_one.call_args_list[1][0][0], 'vnfds', "Incorrect second DB lookup") + self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr["_id"]) + self.assertEqual( + self.db.get_one.call_args_list[0][0][0], + "vnfrs", + "Incorrect first DB lookup", + ) + self.assertEqual( + self.db.get_one.call_args_list[1][0][0], + "vnfds", + "Incorrect second DB lookup", + ) def test_get_vnfd_from_vnf_member_no_revision(self): test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0] - test_vnfr['revision'] = 3 + test_vnfr["revision"] = 3 test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader) self.db.get_one.side_effect = [test_vnfr, test_vnfd] - vnfr = self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr['_id']) - self.assertEqual(self.db.get_one.call_args_list[0][0][0], 'vnfrs', "Incorrect first DB lookup") - self.assertEqual(self.db.get_one.call_args_list[1][0][0], 'vnfds_revisions', "Incorrect second DB lookup") + self.nslcmop_topic._get_vnfd_from_vnf_member_index("1", test_vnfr["_id"]) + self.assertEqual( + self.db.get_one.call_args_list[0][0][0], + "vnfrs", + "Incorrect first DB lookup", + ) + self.assertEqual( + self.db.get_one.call_args_list[1][0][0], + "vnfds_revisions", + "Incorrect second DB lookup", + ) @contextmanager def assertNotRaises(self, exception_type): @@ -507,7 +687,9 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): "VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984", ) - with self.subTest(i=5, t="Ns update REMOVE_VNF request validated with no exception"): + with self.subTest( + i=5, t="Ns update REMOVE_VNF request validated with no exception" + ): test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader) test_vnfr[0]["revision"] = 2 test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) @@ -522,6 +704,7 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): with self.assertNotRaises(EngineException): self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) + class TestNsrTopic(unittest.TestCase): def setUp(self): self.db = DbMemory() @@ -546,113 +729,165 @@ class TestNsrTopic(unittest.TestCase): self.vim = self.db.get_list("vim_accounts")[0] self.vim_id = self.vim["_id"] - def test_create(self): - 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": "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.paas_id = "3d93ca4e-b007-4d94-bbdc-61911078b864" - # check vnfrs and nsrs created in whatever order - created_vnfrs = [] - created_nsrs = [] - nsr_id = None + def stock_db_creations(self, created_vnfrs, created_nsrs): for _call in self.db.create.call_args_list: assert len(_call[0]) >= 2, "called db.create with few parameters" 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", - ) - if nsr_id: - 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" - ) - else: - nsr_id = created_item["_id"] else: - assert True, "created an unknown record {} at database".format( + assert False, "created an unknown record {} at database".format( _call[0][0] ) + self.check_admin_field_in_db_call(created_item) - 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'", - ) + def check_admin_field_in_db_call(self, created_item): + 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'", + ) + def check_vnfr_content(self, vnfr, nsr_id): + self.assertEqual(vnfr["vim-account-id"], None) + self.assertEqual(vnfr["paas-account-id"], None) + self.assertIn( + "member-vnf-index-ref", + vnfr, + "Created item must contain member-vnf-index-ref section", + ) self.assertEqual( - len(created_vnfrs), 2, "created a mismatch number of vnfr at database" + nsr_id, vnfr["nsr-id-ref"], "bad reference id from vnfr to nsr" ) + self.check_vdur_interfaces(vnfr) + def check_vdur_interfaces(self, vnfr): self.assertEqual( - created_vnfrs[0]["vdur"][0]["interfaces"][0]["position"], + vnfr["vdur"][0]["interfaces"][0]["position"], 1, "vdur first interface position does not match", ) - self.assertEqual( - created_vnfrs[0]["vdur"][0]["interfaces"][1]["position"], + vnfr["vdur"][0]["interfaces"][1]["position"], 2, "vdur second interface position does not match", ) + def check_vnfrs(self, created_vnfrs, nsrs_id): + self.assertEqual( + len(created_vnfrs), 2, "created a mismatch number of vnfr at database" + ) + for vnfr in created_vnfrs: + self.check_vnfr_content(vnfr, nsrs_id) + + def check_vnfrs_ref_in_nsr(self, nsrs, created_vnfrs): + self.assertEqual(len(nsrs["constituent-vnfr-ref"]), len(created_vnfrs)) + self.assertNotEqual(created_vnfrs[0]["_id"], created_vnfrs[1]["_id"]) + for vnfr in created_vnfrs: + vnfr_id = vnfr["_id"] + self.assertIn(vnfr_id, nsrs["constituent-vnfr-ref"]) + + def check_nsrs(self, created_nsrs, created_vnfrs, vim_id, paas_id): self.assertEqual( len(created_nsrs), 1, "Only one nsrs must be created at database" ) + nsrs = created_nsrs[0] + self.assertEqual(nsrs["vimdatacenter"], vim_id) + self.assertEqual(nsrs["paasdatacenter"], paas_id) + self.check_vnfrs_ref_in_nsr(nsrs, created_vnfrs) + + def check_created_nsrs_and_vnfrs( + self, created_nsrs, created_vnfrs, rollback, vim_id, paas_id + ): + + nsrs_id = created_nsrs[0]["_id"] + self.assertEqual( len(rollback), len(created_vnfrs) + 1, "rollback mismatch with created items at database", ) + self.check_nsrs(created_nsrs, created_vnfrs, vim_id, paas_id) + self.check_vnfrs(created_vnfrs, nsrs_id) + + def test_create_with_vim_account(self): + 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": "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 + ) + + # check vnfrs and nsrs created in whatever order + created_vnfrs = [] + created_nsrs = [] + self.stock_db_creations(created_vnfrs, created_nsrs) + self.check_created_nsrs_and_vnfrs( + created_nsrs, created_vnfrs, rollback, self.vim_id, None + ) + + def test_create_with_vim_account_raise_exception(self): # test parameters with error bad_id = "88d90b0c-faff-4b9f-bccd-aaaaaaaaaaaa" + 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": "hackfest_vnf1", + "additionalParams": {"touch_filename": "file"}, + }, + { + "member-vnf-index": "hackfest_vnf2", + "additionalParams": {"touch_filename": "file"}, + }, + ], + } + rollback = [] + headers = {} test_set = ( # TODO add "nsd" ( @@ -684,54 +919,114 @@ class TestNsrTopic(unittest.TestCase): 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_create_with_paas_account(self): + session = { + "force": False, + "admin": False, + "public": False, + "project_id": [self.nsd_project], + "method": "write", + } + indata = { + "nsdId": self.nsd_id, + "nsName": "name", + "paasAccountId": self.paas_id, + "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 + ) + + created_vnfrs = [] + created_nsrs = [] + self.stock_db_creations(created_vnfrs, created_nsrs) + self.check_created_nsrs_and_vnfrs( + created_nsrs, created_vnfrs, rollback, None, self.paas_id + ) def test_show_instance(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", + } filter_q = {} for refresh_status in ("true", "false"): self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader)) actual_nsr = self.db.get_list("nsrs")[0] nsr_id = actual_nsr["_id"] - filter_q['vcaStatus-refresh'] = refresh_status + filter_q["vcaStatus-refresh"] = refresh_status expected_nsr = self.nsr_topic.show(session, nsr_id, filter_q=filter_q) self.nsr_topic.delete(session, nsr_id) actual_nsr.pop("_admin") expected_nsr.pop("_admin") - self.assertEqual(expected_nsr, actual_nsr, "Database nsr and show() nsr do not match.") + self.assertEqual( + expected_nsr, actual_nsr, "Database nsr and show() nsr do not match." + ) def test_vca_status_refresh(self): - session = {"force": False, "admin": False, "public": False, "project_id": [self.nsd_project], "method": "write"} - filter_q = {'vcaStatus-refresh': 'true'} + session = { + "force": False, + "admin": False, + "public": False, + "project_id": [self.nsd_project], + "method": "write", + } + filter_q = {"vcaStatus-refresh": "true"} time_delta = 120 self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader)) nsr = self.db.get_list("nsrs")[0] # When vcaStatus-refresh is true - filter_q['vcaStatus-refresh'] = "true" + filter_q["vcaStatus-refresh"] = "true" self.nsr_topic.vca_status_refresh(session, nsr, filter_q) msg_args = self.msg.write.call_args[0] self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action") self.assertGreater(nsr["_admin"]["modified"], time() - time_delta) # When vcaStatus-refresh is false but modified time is within threshold - filter_q['vcaStatus-refresh'] = "false" + filter_q["vcaStatus-refresh"] = "false" time_now = time() nsr["_admin"]["modified"] = time_now self.nsr_topic.vca_status_refresh(session, nsr, filter_q) msg_args = self.msg.write.call_args[1] self.assertEqual(msg_args, {}, "Message should not be sent.") - self.assertEqual(nsr["_admin"]["modified"], time_now, "Modified time should not be changed.") + self.assertEqual( + nsr["_admin"]["modified"], time_now, "Modified time should not be changed." + ) # When vcaStatus-refresh is false but modified time is less than threshold - filter_q['vcaStatus-refresh'] = "false" - nsr["_admin"]["modified"] = time() - (2*time_delta) + filter_q["vcaStatus-refresh"] = "false" + nsr["_admin"]["modified"] = time() - (2 * time_delta) self.nsr_topic.vca_status_refresh(session, nsr, filter_q) msg_args = self.msg.write.call_args[0] self.assertEqual(msg_args[1], "vca_status_refresh", "Wrong message action") self.nsr_topic.delete(session, nsr["_id"]) - self.assertGreater(nsr["_admin"]["modified"], time() - time_delta, "Modified time is not changed.") + self.assertGreater( + nsr["_admin"]["modified"], + time() - time_delta, + "Modified time is not changed.", + ) def test_delete_ns(self): self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader))