X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;ds=inline;f=osm_lcm%2Ftests%2Ftest_ns.py;h=7e7270049405667cfe543a812c2fc383d8e1048d;hb=ceb0386270a13ab6b68c3549c01813eb41c178a5;hp=f44dbf691609427b0ea9a30a8e11dab2003caca4;hpb=c09b842d41437f1c2bb4a931182afbe7088d696f;p=osm%2FLCM.git diff --git a/osm_lcm/tests/test_ns.py b/osm_lcm/tests/test_ns.py index f44dbf6..7e72700 100644 --- a/osm_lcm/tests/test_ns.py +++ b/osm_lcm/tests/test_ns.py @@ -31,7 +31,7 @@ from osm_lcm.lcm_utils import TaskRegistry from osm_lcm.ng_ro import NgRoClient from osm_lcm.data_utils.database.database import Database from osm_lcm.data_utils.filesystem.filesystem import Filesystem -from osm_lcm.data_utils.vca import Relation, EERelation +from osm_lcm.data_utils.vca import Relation, EERelation, DeployedVCA from osm_lcm.data_utils.vnfd import find_software_version from osm_lcm.lcm_utils import check_juju_bundle_existence, get_charm_artifact_path from osm_lcm.lcm_utils import LcmException @@ -99,7 +99,7 @@ def callable(a): return a -class TestMyNS(asynctest.TestCase): +class TestBaseNS(asynctest.TestCase): async def _n2vc_DeployCharms( self, model_name, @@ -205,18 +205,15 @@ class TestMyNS(asynctest.TestCase): self.db = Database({"database": {"driver": "memory"}}).instance.db self.db.create_list("vnfds", yaml.safe_load(descriptors.db_vnfds_text)) self.db.create_list( - "vnfds_revisions", - yaml.safe_load(descriptors.db_vnfds_revisions_text), + "vnfds_revisions", yaml.safe_load(descriptors.db_vnfds_revisions_text) ) self.db.create_list("nsds", yaml.safe_load(descriptors.db_nsds_text)) self.db.create_list("nsrs", yaml.safe_load(descriptors.db_nsrs_text)) self.db.create_list( - "vim_accounts", - yaml.safe_load(descriptors.db_vim_accounts_text), + "vim_accounts", yaml.safe_load(descriptors.db_vim_accounts_text) ) self.db.create_list( - "k8sclusters", - yaml.safe_load(descriptors.db_k8sclusters_text), + "k8sclusters", yaml.safe_load(descriptors.db_k8sclusters_text) ) self.db.create_list( "nslcmops", yaml.safe_load(descriptors.db_nslcmops_text) @@ -410,6 +407,8 @@ class TestMyNS(asynctest.TestCase): # await self.test_instantiate() # # this will check that the initial-congig-primitive 'not_to_be_called' is not called + +class TestMyNS(TestBaseNS): @asynctest.fail_on(active_handles=True) async def test_start_stop_rebuild_pass(self): nsr_id = descriptors.test_ids["TEST-OP-VNF"]["ns"] @@ -515,13 +514,16 @@ class TestMyNS(asynctest.TestCase): ): if ( v.get("execution-environment-list") - and "juju" in v["execution-environment-list"][k] + and "juju" in v["execution-environment-list"][0] ): expected_value = self.db.get_list("nsrs")[i][ "vcaStatus" ] await self.my_ns._on_update_n2vc_db( - "nsrs", {"_id": nsr_id}, "_admin.deployed.VCA.0", {} + "nsrs", + {"_id": nsr_id}, + "_admin.deployed.VCA.{}".format(k), + {}, ) return_value = self.db.get_list("nsrs")[i]["vcaStatus"] self.assertEqual(return_value, expected_value) @@ -1194,11 +1196,7 @@ class TestMyNS(asynctest.TestCase): q_filter={"_id": vnfd_id + ":1"}, update_dict={"_admin.revision": 1, "kdu": []}, ) - self.db.set_one( - "vnfrs", - q_filter={"_id": vnfr_id}, - update_dict={"revision": 1}, - ) + self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1}) mock_charm_hash.return_value = False mock_charm_artifact.side_effect = [ @@ -1276,9 +1274,7 @@ class TestMyNS(asynctest.TestCase): self.db.set_one( "nsrs", q_filter={"_id": nsr_id}, - update_dict={ - "_admin.deployed.VCA.0.kdu_name": "native-kdu", - }, + update_dict={"_admin.deployed.VCA.0.kdu_name": "native-kdu"}, ) self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1}) @@ -1320,20 +1316,12 @@ class TestMyNS(asynctest.TestCase): self.db.set_one( "vnfds", q_filter={"_id": vnfd_id}, - update_dict={ - "_admin.revision": 3, - "software-version": "1.0", - "kdu": [], - }, + update_dict={"_admin.revision": 3, "software-version": "1.0", "kdu": []}, ) self.db.set_one( "vnfds_revisions", q_filter={"_id": vnfd_id + ":1"}, - update_dict={ - "_admin.revision": 1, - "software-version": "1.0", - "kdu": [], - }, + update_dict={"_admin.revision": 1, "software-version": "1.0", "kdu": []}, ) self.db.set_one("vnfrs", q_filter={"_id": vnfr_id}, update_dict={"revision": 1}) @@ -1520,10 +1508,7 @@ class TestMyNS(asynctest.TestCase): def test_ns_update_check_juju_charm_artifacts_base_folder_wth_pkgdir(self): """Check charm artifacts""" - base_folder = { - "folder": vnfd_id, - "pkg-dir": "hackfest_3charmed_vnfd", - } + base_folder = {"folder": vnfd_id, "pkg-dir": "hackfest_3charmed_vnfd"} charm_name = "simple" charm_type = "lxc_proxy_charm" revision = 3 @@ -1533,9 +1518,7 @@ class TestMyNS(asynctest.TestCase): def test_ns_update_check_juju_charm_artifacts_base_folder_wthout_pkgdir(self): """Check charm artifacts, SOL004 packages""" - base_folder = { - "folder": vnfd_id, - } + base_folder = {"folder": vnfd_id} charm_name = "basic" charm_type, revision = "", "" expected_result = f"{vnfd_id}/Scripts/helm-charts/basic" @@ -1543,7 +1526,7 @@ class TestMyNS(asynctest.TestCase): self.assertEqual(result, expected_result, "Wrong charm artifact path") -class TestInstantiateN2VC(TestMyNS): +class TestInstantiateN2VC(TestBaseNS): async def setUp(self): await super().setUp() self.db_nsr = yaml.safe_load(descriptors.db_nsrs_text)[0] @@ -1566,6 +1549,7 @@ class TestInstantiateN2VC(TestMyNS): vdu_id=None, kdu_name=None, vdu_index=None, + kdu_index=None, config_descriptor=config_descriptor, deploy_params={}, base_folder=base_folder, @@ -1644,5 +1628,63 @@ class TestInstantiateN2VC(TestMyNS): self.assertTrue(await self.call_ns_add_relation()) +class TestGetVNFRelations(TestBaseNS): + async def setUp(self): + await super().setUp() + self.db_nsd = yaml.safe_load(descriptors.db_nsds_text)[0] + + def test_ns_charm_vca_returns_empty_relations(self): + ns_charm_vca = {"member-vnf-index": None, "target_element": "ns"} + nsr_id = self.db_nsd["id"] + deployed_vca = DeployedVCA(nsr_id, ns_charm_vca) + + expected_relations = [] + self.assertEqual( + expected_relations, + self.my_ns._get_vnf_relations( + nsr_id=nsr_id, nsd=self.db_nsd, vca=deployed_vca, cached_vnfds={} + ), + ) + + def test_vnf_returns_relation(self): + vnf_vca = { + "member-vnf-index": "1", + "target_element": "vnf/0", + "ee_descriptor_id": "simple-ee", + "vdu_id": "mgmtVM", + } + nsr_id = self.db_nsd["id"] + deployed_vca = DeployedVCA(nsr_id, vnf_vca) + + provider_dict = { + "nsr-id": nsr_id, + "vnf-profile-id": "1", + "vdu-profile-id": "mgmtVM", + "kdu-resource-profile-id": None, + "execution-environment-ref": "simple-ee", + "endpoint": "interface", + } + + requirer_dict = { + "nsr-id": nsr_id, + "vnf-profile-id": "1", + "vdu-profile-id": "dataVM", + "kdu-resource-profile-id": None, + "execution-environment-ref": "simple-ee", + "endpoint": "interface", + } + + provider = EERelation(provider_dict) + requirer = EERelation(requirer_dict) + relation = Relation("relation", provider, requirer) + + relations_found = self.my_ns._get_vnf_relations( + nsr_id=nsr_id, nsd=self.db_nsd, vca=deployed_vca, cached_vnfds={} + ) + + self.assertEqual(1, len(relations_found)) + self.assertEqual(relation, relations_found[0]) + + if __name__ == "__main__": asynctest.main()