X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FNBI.git;a=blobdiff_plain;f=osm_nbi%2Ftests%2Ftest_instance_topics.py;h=2ad596d87b73f53c98b917b28376db356ed72715;hp=bff7775111b45ddb0c830d2e827ac38fa4997cdf;hb=c9c0339dee11c893b35e9eb05098554dbda751a3;hpb=7cbd03c5c21202fff473ae943e75dd49a18f2516 diff --git a/osm_nbi/tests/test_instance_topics.py b/osm_nbi/tests/test_instance_topics.py index bff7775..2ad596d 100644 --- a/osm_nbi/tests/test_instance_topics.py +++ b/osm_nbi/tests/test_instance_topics.py @@ -15,23 +15,30 @@ # contact: esousa@whitestack.com or alfonso.tiernosepulveda@telefonica.com ## +from contextlib import contextmanager import unittest +from time import time 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 from osm_common.fsbase import FsBase from osm_common.msgbase import MsgBase +from osm_common import dbbase 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,13 +49,14 @@ 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)) self.db.create_list("nsrs", yaml.load(db_nsrs_text, Loader=yaml.Loader)) self.db.create = Mock(return_value="created_id") - self.db.set_one = Mock(return_value={"updated": 1}) self.nsd = self.db.get_list("nsds")[0] self.nsd_id = self.nsd["_id"] self.nsr = self.db.get_list("nsrs")[0] @@ -59,64 +67,138 @@ 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"} + self.db.set_one = Mock(return_value={"updated": 1}) + 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 +208,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 +218,213 @@ 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), + ) + + def test_update_remove_vnf(self): + vnfr_id = self.db.get_list("vnfrs")[0]["_id"] + session = {} + self.db.set_one( + "nsrs", + {"_id": self.nsr_id}, + {"_admin.nsState": "INSTANTIATED"}, + ) + indata = { + "lcmOperationType": "update", + "updateType": "REMOVE_VNF", + "nsInstanceId": self.nsr_id, + "removeVnfInstanceId": vnfr_id + } + session = { + "force": False, + "admin": False, + "public": False, + "project_id": [self.nsr_project], + "method": "write", + } + rollback = [] + headers = {} + + 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"] == "update", + "Database record must contain 'lcmOperationType=update'", + ) + self.assertTrue( + created_nslcmop["operationParams"]["updateType"] == "REMOVE_VNF", + "Database record must contain 'updateType=REMOVE_VNF'", + ) -class TestNsrTopic(unittest.TestCase): +class TestNsLcmOpTopicWithMock(unittest.TestCase): + def setUp(self): + self.db = Mock(dbbase.DbBase()) + self.fs = Mock(FsBase()) + self.fs.get_params.return_value = {"./fake/folder"} + self.fs.file_open = mock_open() + self.msg = Mock(MsgBase()) + # create class + 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) + 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") + + 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) + 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") + + @contextmanager + def assertNotRaises(self, exception_type): + try: + yield None + except exception_type: + raise self.failureException("{} raised".format(exception_type.__name__)) + + def test_check_ns_update_operation(self): + self.db = DbMemory() + self.nslcmop_topic = NsLcmOpTopic(self.db, self.fs, self.msg, None) + 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[0]["revision"] = 2 + test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + test_nsr[0]["constituent-vnfr-ref"][ + 0 + ] = "99d90b0c-faff-4b9f-bccd-017f33985984" + self.db.create_list("vnfrs", test_vnfr) + self.db.create_list("nsrs", test_nsr) + nsrs = self.db.get_list("nsrs")[0] + indata = { + "updateType": "CHANGE_VNFPKG", + "changeVnfPackageData": { + "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", + "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77", + }, + "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", + } + with self.assertRaises(EngineException) as expected_exception: + self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) + self.assertEqual( + str(expected_exception.exception), + "Error in validating ns-update request: vnf 88d90b0c-faff-4b9f-bccd-017f33985984" + " does not belong to NS f48163a6-c807-47bc-9682-f72caef5af85", + ) + + 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[0]["revision"] = 2 + test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + self.db.create_list("vnfrs", test_vnfr) + self.db.create_list("nsrs", test_nsr) + nsrs = self.db.get_list("nsrs")[1] + indata = { + "updateType": "CHANGE_VNFPKG", + "changeVnfPackageData": { + "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", + "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77", + }, + "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", + } + with self.assertNotRaises(EngineException): + self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) + + with self.subTest( + i=3, t="Ns update request rejected because of too small timeout" + ): + indata = { + "updateType": "CHANGE_VNFPKG", + "changeVnfPackageData": { + "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", + "vnfdId": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77", + }, + "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", + "timeout_ns_update": 50, + } + with self.assertRaises(EngineException) as expected_exception: + self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) + self.assertEqual( + str(expected_exception.exception), + "Error in validating ns-update request: 50 second is not enough " + "to upgrade the VNF instance: 88d90b0c-faff-4b9f-bccd-017f33985984", + ) + + 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[0]["revision"] = 2 + test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + self.db.create_list("vnfrs", test_vnfr) + self.db.create_list("nsrs", test_nsr) + nsrs = self.db.get_list("nsrs")[2] + indata = { + "updateType": "CHANGE_VNFPKG", + "changeVnfPackageData": { + "vnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", + "vnfdId": "9637bcf8-cf14-42dc-ad70-c66fcf1e6e77", + }, + "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", + } + with self.assertRaises(EngineException) as expected_exception: + self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) + self.assertEqual( + str(expected_exception.exception), + "Error in validating ns-update request: vnfd-id 9637bcf8-cf14-42dc-ad70-c66fcf1e6e77 does not " + "match with the vnfd-id: 7637bcf8-cf14-42dc-ad70-c66fcf1e6e77 of " + "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) + test_vnfr[0]["revision"] = 2 + test_nsr = yaml.load(db_nsrs_text, Loader=yaml.Loader) + self.db.create_list("vnfrs", test_vnfr) + self.db.create_list("nsrs", test_nsr) + nsrs = self.db.get_list("nsrs")[1] + indata = { + "updateType": "REMOVE_VNF", + "removeVnfInstanceId": "88d90b0c-faff-4b9f-bccd-017f33985984", + "nsInstanceId": "f48163a6-c807-47bc-9682-f72caef5af85", + } + with self.assertNotRaises(EngineException): + self.nslcmop_topic._check_ns_operation(session, nsrs, "update", indata) + +class TestNsrTopic(unittest.TestCase): def setUp(self): self.db = DbMemory() self.fs = Mock(FsBase()) @@ -154,7 +435,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,20 +449,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": "hackfest_vnf1", "additionalParams": {"touch_filename": "file"}}, - {"member-vnf-index": "hackfest_vnf2", "additionalParams": {"touch_filename": "file"}} - ] + { + "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 = [] @@ -190,44 +487,101 @@ 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.assertEqual( + len(created_vnfrs), 2, "created a mismatch number of vnfr at database" + ) - 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( + created_vnfrs[0]["vdur"][0]["interfaces"][0]["position"], + 1, + "vdur first interface position does not match", + ) - 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") + self.assertEqual( + created_vnfrs[0]["vdur"][0]["interfaces"][1]["position"], + 2, + "vdur second interface position does not match", + ) + + 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: @@ -235,6 +589,52 @@ class TestNsrTopic(unittest.TestCase): 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"} + 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 + 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.") + + 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'} + 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" + 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" + 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.") + + # When vcaStatus-refresh is false but modified time is less than threshold + 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.") + 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] @@ -243,10 +643,28 @@ class TestNsrTopic(unittest.TestCase): 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'): + 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) @@ -254,19 +672,30 @@ class TestNsrTopic(unittest.TestCase): 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[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.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}) + 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() @@ -278,12 +707,16 @@ class TestNsrTopic(unittest.TestCase): 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.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'): + 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() @@ -293,7 +726,9 @@ class TestNsrTopic(unittest.TestCase): 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[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") @@ -302,29 +737,44 @@ class TestNsrTopic(unittest.TestCase): 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}) + 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: + 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.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'): + 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: + 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.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