X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_nbi%2Ftests%2Ftest_instance_topics.py;h=b12a3304ea9d9abbed5ffd673c2ec5d72332bb05;hb=828f3f29a221f4415e0962e63a491462c1b02370;hp=2ad596d87b73f53c98b917b28376db356ed72715;hpb=c9c0339dee11c893b35e9eb05098554dbda751a3;p=osm%2FNBI.git diff --git a/osm_nbi/tests/test_instance_topics.py b/osm_nbi/tests/test_instance_topics.py index 2ad596d..b12a330 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 # patch, MagicMock from osm_common.dbbase import DbException from osm_nbi.engine import EngineException from osm_common.dbmemory import DbMemory @@ -49,13 +49,11 @@ 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("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)) - self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader)) + self.db.create_list("vim_accounts", yaml.safe_load(db_vim_accounts_text)) + self.db.create_list("nsds", yaml.safe_load(db_nsds_text)) + self.db.create_list("vnfds", yaml.safe_load(db_vnfds_text)) + self.db.create_list("vnfrs", yaml.safe_load(db_vnfrs_text)) + self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) self.db.create = Mock(return_value="created_id") self.nsd = self.db.get_list("nsds")[0] self.nsd_id = self.nsd["_id"] @@ -240,7 +238,7 @@ class TestNsLcmOpTopic(unittest.TestCase): "lcmOperationType": "update", "updateType": "REMOVE_VNF", "nsInstanceId": self.nsr_id, - "removeVnfInstanceId": vnfr_id + "removeVnfInstanceId": vnfr_id, } session = { @@ -281,6 +279,102 @@ class TestNsLcmOpTopic(unittest.TestCase): "Database record must contain 'updateType=REMOVE_VNF'", ) + def test_migrate(self): + _ = self.db.get_list("vnfrs")[0]["_id"] + session = {} + self.db.set_one( + "nsrs", + {"_id": self.nsr_id}, + {"_admin.nsState": "INSTANTIATED"}, + ) + session = { + "force": False, + "admin": False, + "public": False, + "project_id": [self.nsr_project], + "method": "write", + } + rollback = [] + headers = {} + + with self.subTest(i=1, t="Migration for Specific VM"): + indata = { + "lcmOperationType": "migrate", + "nsInstanceId": self.nsr_id, + "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 + ) + + 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", "nslcmops entry must be created at database" + ) + created_nslcmop = _call[0][1] + self.assertEqual( + self.nsr_id, + created_nslcmop["nsInstanceId"], + "mismatch between nsId '_id' in created nslcmop and database nsr", + ) + self.assertTrue( + created_nslcmop["lcmOperationType"] == "migrate", + "Database record must contain 'lcmOperationType=migrate'", + ) + with self.subTest(i=2, t="Migration of all VDUs in a VNF"): + indata = { + "lcmOperationType": "migrate", + "nsInstanceId": self.nsr_id, + "vnfInstanceId": "9e8006df-cdfa-4f63-bf6a-fce860d71c1f", + } + nslcmop_id, _ = self.nslcmop_topic.new( + rollback, session, indata, kwargs=None, headers=headers + ) + + self.assertEqual( + self.db.create.call_count, + 2, + "database create not called, or called more than once", + ) + _call = self.db.create.call_args_list[0] + self.assertEqual( + _call[0][0], "nslcmops", "nslcmops entry must be created at database" + ) + created_nslcmop = _call[0][1] + self.assertEqual( + self.nsr_id, + created_nslcmop["nsInstanceId"], + "mismatch between nsId '_id' in created nslcmop and database nsr", + ) + self.assertTrue( + created_nslcmop["lcmOperationType"] == "migrate", + "Database record must contain 'lcmOperationType=migrate'", + ) + with self.subTest(i=3, t="Migration failure - vduId not provided in vdu "): + indata = { + "lcmOperationType": "migrate", + "nsInstanceId": self.nsr_id, + "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 + ) + self.assertTrue( + "Format error at 'vdu' ''vduId' is a required property'" + in str(e.exception) + ) + class TestNsLcmOpTopicWithMock(unittest.TestCase): def setUp(self): @@ -293,21 +387,37 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None) def test_get_vnfd_from_vnf_member_revision(self): - test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader)[0] - test_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader) + test_vnfr = yaml.safe_load(db_vnfrs_text)[0] + test_vnfd = yaml.safe_load(db_vnfds_text) 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_vnfd = yaml.load(db_vnfds_text, Loader=yaml.Loader) + test_vnfr = yaml.safe_load(db_vnfrs_text)[0] + test_vnfr["revision"] = 3 + test_vnfd = yaml.safe_load(db_vnfds_text) 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): @@ -322,9 +432,9 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): session = {} with self.subTest(i=1, t="VNF instance does not belong to NS"): - test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader) + test_vnfr = yaml.safe_load(db_vnfrs_text) test_vnfr[0]["revision"] = 2 - test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + test_nsr = yaml.safe_load(db_nsrs_text) test_nsr[0]["constituent-vnfr-ref"][ 0 ] = "99d90b0c-faff-4b9f-bccd-017f33985984" @@ -348,9 +458,9 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): ) with self.subTest(i=2, t="Ns update request validated with no exception"): - test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader) + test_vnfr = yaml.safe_load(db_vnfrs_text) test_vnfr[0]["revision"] = 2 - test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + test_nsr = yaml.safe_load(db_nsrs_text) self.db.create_list("vnfrs", test_vnfr) self.db.create_list("nsrs", test_nsr) nsrs = self.db.get_list("nsrs")[1] @@ -386,9 +496,9 @@ class TestNsLcmOpTopicWithMock(unittest.TestCase): ) with self.subTest(i=4, t="wrong vnfdid is given as an update parameter"): - test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader) + test_vnfr = yaml.safe_load(db_vnfrs_text) test_vnfr[0]["revision"] = 2 - test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + test_nsr = yaml.safe_load(db_nsrs_text) self.db.create_list("vnfrs", test_vnfr) self.db.create_list("nsrs", test_nsr) nsrs = self.db.get_list("nsrs")[2] @@ -409,10 +519,12 @@ 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"): - test_vnfr = yaml.load(db_vnfrs_text, Loader=yaml.Loader) + with self.subTest( + i=5, t="Ns update REMOVE_VNF request validated with no exception" + ): + test_vnfr = yaml.safe_load(db_vnfrs_text) test_vnfr[0]["revision"] = 2 - test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + test_nsr = yaml.safe_load(db_nsrs_text) self.db.create_list("vnfrs", test_vnfr) self.db.create_list("nsrs", test_nsr) nsrs = self.db.get_list("nsrs")[1] @@ -424,6 +536,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() @@ -435,11 +548,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("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("vim_accounts", yaml.safe_load(db_vim_accounts_text)) + self.db.create_list("nsds", yaml.safe_load(db_nsds_text)) + self.db.create_list("vnfds", yaml.safe_load(db_vnfds_text)) self.db.create = Mock(return_value="created_id") self.nsd = self.db.get_list("nsds")[0] self.nsd_id = self.nsd["_id"] @@ -586,57 +697,80 @@ 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_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)) + self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) 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)) + self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) 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)) + self.db.create_list("nsrs", yaml.safe_load(db_nsrs_text)) self.nsr = self.db.get_list("nsrs")[0] self.nsr_id = self.nsr["_id"] self.db_set_one = self.db.set_one