X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_lcm%2Ftests%2Ftest_ns.py;h=1082648432cc70961db3cca4ceb4bc5f9261dc55;hb=afff693a2ae9757aee5f731ee12bca70cdb9a907;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..1082648 100644 --- a/osm_lcm/tests/test_ns.py +++ b/osm_lcm/tests/test_ns.py @@ -21,7 +21,6 @@ import asyncio from copy import deepcopy import yaml import copy -from n2vc.exceptions import N2VCException from os import getenv from osm_lcm import ns from osm_common.msgkafka import MsgKafka @@ -31,7 +30,6 @@ 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.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 @@ -187,17 +185,8 @@ class TestMyNS(asynctest.TestCase): return str(uuid4()) async def setUp(self): - self.mock_db() - self.mock_kafka() - self.mock_filesystem() - self.mock_task_registry() - self.mock_vca_k8s() - self.create_nslcm_class() - self.mock_logging() - self.mock_vca_n2vc() - self.mock_ro() - - def mock_db(self): + + # Mock DB if not getenv("OSMLCMTEST_DB_NOMOCK"): # Cleanup singleton Database instance Database.instance = None @@ -224,10 +213,10 @@ class TestMyNS(asynctest.TestCase): self.db.create_list("vnfrs", yaml.safe_load(descriptors.db_vnfrs_text)) self.db_vim_accounts = yaml.safe_load(descriptors.db_vim_accounts_text) - def mock_kafka(self): + # Mock kafka self.msg = asynctest.Mock(MsgKafka()) - def mock_filesystem(self): + # Mock filesystem if not getenv("OSMLCMTEST_FS_NOMOCK"): self.fs = asynctest.Mock( Filesystem({"storage": {"driver": "local", "path": "/"}}).instance.fs @@ -239,13 +228,13 @@ class TestMyNS(asynctest.TestCase): # self.fs.file_open.return_value.__enter__.return_value = asynctest.MagicMock() # called on a python "with" # self.fs.file_open.return_value.__enter__.return_value.read.return_value = "" # empty file - def mock_task_registry(self): + # Mock TaskRegistry self.lcm_tasks = asynctest.Mock(TaskRegistry()) self.lcm_tasks.lock_HA.return_value = True self.lcm_tasks.waitfor_related_HA.return_value = None self.lcm_tasks.lookfor_related.return_value = ("", []) - def mock_vca_k8s(self): + # Mock VCA - K8s if not getenv("OSMLCMTEST_VCA_K8s_NOMOCK"): ns.K8sJujuConnector = asynctest.MagicMock(ns.K8sJujuConnector) ns.K8sHelmConnector = asynctest.MagicMock(ns.K8sHelmConnector) @@ -255,17 +244,17 @@ class TestMyNS(asynctest.TestCase): ns.N2VCJujuConnector = asynctest.MagicMock(ns.N2VCJujuConnector) ns.LCMHelmConn = asynctest.MagicMock(ns.LCMHelmConn) - def create_nslcm_class(self): + # Create NsLCM class self.my_ns = ns.NsLcm(self.msg, self.lcm_tasks, lcm_config, self.loop) self.my_ns.fs = self.fs self.my_ns.db = self.db self.my_ns._wait_dependent_n2vc = asynctest.CoroutineMock() - def mock_logging(self): + # Mock logging if not getenv("OSMLCMTEST_LOGGING_NOMOCK"): self.my_ns.logger = asynctest.Mock(self.my_ns.logger) - def mock_vca_n2vc(self): + # Mock VCA - N2VC if not getenv("OSMLCMTEST_VCA_NOMOCK"): pub_key = getenv("OSMLCMTEST_NS_PUBKEY", "ssh-rsa test-pub-key t@osm.com") # self.my_ns.n2vc = asynctest.Mock(N2VC()) @@ -309,11 +298,8 @@ class TestMyNS(asynctest.TestCase): self.my_ns.n2vc.delete_namespace = asynctest.CoroutineMock( return_value=None ) - self.my_ns.n2vc.register_execution_environment = asynctest.CoroutineMock( - return_value="model-name.application-name.k8s" - ) - def mock_ro(self): + # Mock RO if not getenv("OSMLCMTEST_RO_NOMOCK"): self.my_ns.RO = asynctest.Mock( NgRoClient(self.loop, **lcm_config.RO.to_dict()) @@ -1543,106 +1529,5 @@ class TestMyNS(asynctest.TestCase): self.assertEqual(result, expected_result, "Wrong charm artifact path") -class TestInstantiateN2VC(TestMyNS): - async def setUp(self): - await super().setUp() - self.db_nsr = yaml.safe_load(descriptors.db_nsrs_text)[0] - self.db_vnfr = yaml.safe_load(descriptors.db_vnfrs_text)[0] - self.vca_index = 1 - self.my_ns._write_configuration_status = Mock() - - async def call_instantiate_N2VC(self): - logging_text = "N2VC Instantiation" - config_descriptor = {"config-access": {"ssh-access": {"default-user": "admin"}}} - base_folder = {"pkg-dir": "", "folder": "~"} - stage = ["Stage", "Message"] - - await self.my_ns.instantiate_N2VC( - logging_text=logging_text, - vca_index=self.vca_index, - nsi_id="nsi_id", - db_nsr=self.db_nsr, - db_vnfr=self.db_vnfr, - vdu_id=None, - kdu_name=None, - vdu_index=None, - config_descriptor=config_descriptor, - deploy_params={}, - base_folder=base_folder, - nslcmop_id="nslcmop_id", - stage=stage, - vca_type="native_charm", - vca_name="vca_name", - ee_config_descriptor={}, - ) - - def check_config_status(self, expected_status): - self.my_ns._write_configuration_status.assert_called_with( - nsr_id=self.db_nsr["_id"], vca_index=self.vca_index, status=expected_status - ) - - async def call_ns_add_relation(self): - ee_relation = EERelation( - { - "nsr-id": self.db_nsr["_id"], - "vdu-profile-id": None, - "kdu-resource-profile-id": None, - "vnf-profile-id": "hackfest_vnf1", - "execution-environment-ref": "f48163a6-c807-47bc-9682-f72caef5af85.alf-c-ab", - "endpoint": "127.0.0.1", - } - ) - - relation = Relation("relation-name", ee_relation, ee_relation) - cached_vnfrs = {"hackfest_vnf1": self.db_vnfr} - - return await self.my_ns._add_relation( - relation=relation, - vca_type="native_charm", - db_nsr=self.db_nsr, - cached_vnfds={}, - cached_vnfrs=cached_vnfrs, - ) - - async def test_add_relation_ok(self): - await self.call_instantiate_N2VC() - self.check_config_status(expected_status="READY") - - async def test_add_relation_returns_false_raises_exception(self): - self.my_ns._add_vca_relations = asynctest.CoroutineMock(return_value=False) - - with self.assertRaises(LcmException) as exception: - await self.call_instantiate_N2VC() - - exception_msg = "Relations could not be added to VCA." - self.assertTrue(exception_msg in str(exception.exception)) - self.check_config_status(expected_status="BROKEN") - - async def test_add_relation_raises_lcm_exception(self): - exception_msg = "Relations FAILED" - self.my_ns._add_vca_relations = asynctest.CoroutineMock( - side_effect=LcmException(exception_msg) - ) - - with self.assertRaises(LcmException) as exception: - await self.call_instantiate_N2VC() - - self.assertTrue(exception_msg in str(exception.exception)) - self.check_config_status(expected_status="BROKEN") - - async def test_n2vc_add_relation_fails_raises_exception(self): - exception_msg = "N2VC failed to add relations" - self.my_ns.n2vc.add_relation = asynctest.CoroutineMock( - side_effect=N2VCException(exception_msg) - ) - with self.assertRaises(LcmException) as exception: - await self.call_ns_add_relation() - self.assertTrue(exception_msg in str(exception.exception)) - - async def test_n2vc_add_relation_ok_returns_true(self): - self.my_ns.n2vc.add_relation = asynctest.CoroutineMock(return_value=None) - self.assertTrue(await self.call_ns_add_relation()) - - if __name__ == "__main__": asynctest.main()