X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_lcm%2Ftests%2Ftest_ns.py;h=d2062d348a5c182360cf85cdcb79dd86ff93ff82;hb=e11384e1797ea0a5f8cd084d6f336948170bc640;hp=7e7270049405667cfe543a812c2fc383d8e1048d;hpb=120695e9fe7ee315b83466d664a0f143c5f90044;p=osm%2FLCM.git diff --git a/osm_lcm/tests/test_ns.py b/osm_lcm/tests/test_ns.py index 7e72700..d2062d3 100644 --- a/osm_lcm/tests/test_ns.py +++ b/osm_lcm/tests/test_ns.py @@ -117,7 +117,7 @@ class TestBaseNS(asynctest.TestCase): ("active", "Ready!"), ): # call callback after some time - asyncio.sleep(5, loop=self.loop) + asyncio.sleep(5) callback(model_name, application_name, status, message, *callback_args) @staticmethod @@ -245,7 +245,7 @@ class TestBaseNS(asynctest.TestCase): def mock_vca_k8s(self): if not getenv("OSMLCMTEST_VCA_K8s_NOMOCK"): ns.K8sJujuConnector = asynctest.MagicMock(ns.K8sJujuConnector) - ns.K8sHelmConnector = asynctest.MagicMock(ns.K8sHelmConnector) + # ns.K8sHelmConnector = asynctest.MagicMock(ns.K8sHelmConnector) ns.K8sHelm3Connector = asynctest.MagicMock(ns.K8sHelm3Connector) if not getenv("OSMLCMTEST_VCA_NOMOCK"): @@ -253,7 +253,7 @@ class TestBaseNS(asynctest.TestCase): ns.LCMHelmConn = asynctest.MagicMock(ns.LCMHelmConn) def create_nslcm_class(self): - self.my_ns = ns.NsLcm(self.msg, self.lcm_tasks, lcm_config, self.loop) + self.my_ns = ns.NsLcm(self.msg, self.lcm_tasks, lcm_config) self.my_ns.fs = self.fs self.my_ns.db = self.db self.my_ns._wait_dependent_n2vc = asynctest.CoroutineMock() @@ -312,9 +312,7 @@ class TestBaseNS(asynctest.TestCase): def mock_ro(self): if not getenv("OSMLCMTEST_RO_NOMOCK"): - self.my_ns.RO = asynctest.Mock( - NgRoClient(self.loop, **lcm_config.RO.to_dict()) - ) + self.my_ns.RO = asynctest.Mock(NgRoClient(**lcm_config.RO.to_dict())) # TODO first time should be empty list, following should return a dict # self.my_ns.RO.get_list = asynctest.CoroutineMock(self.my_ns.RO.get_list, return_value=[]) self.my_ns.RO.deploy = asynctest.CoroutineMock( @@ -326,87 +324,6 @@ class TestBaseNS(asynctest.TestCase): # "description": "done"}}) self.my_ns.RO.delete = asynctest.CoroutineMock(self.my_ns.RO.delete) - # @asynctest.fail_on(active_handles=True) # all async tasks must be completed - # async def test_instantiate(self): - # nsr_id = descriptors.test_ids["TEST-A"]["ns"] - # nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"] - # # print("Test instantiate started") - - # # delete deployed information of database - # if not getenv("OSMLCMTEST_DB_NOMOCK"): - # if self.db.get_list("nsrs")[0]["_admin"].get("deployed"): - # del self.db.get_list("nsrs")[0]["_admin"]["deployed"] - # for db_vnfr in self.db.get_list("vnfrs"): - # db_vnfr.pop("ip_address", None) - # for db_vdur in db_vnfr["vdur"]: - # db_vdur.pop("ip_address", None) - # db_vdur.pop("mac_address", None) - # if getenv("OSMLCMTEST_RO_VIMID"): - # self.db.get_list("vim_accounts")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID") - # if getenv("OSMLCMTEST_RO_VIMID"): - # self.db.get_list("nsrs")[0]["_admin"]["deployed"]["RO"] = getenv("OSMLCMTEST_RO_VIMID") - - # await self.my_ns.instantiate(nsr_id, nslcmop_id) - - # self.msg.aiowrite.assert_called_once_with("ns", "instantiated", - # {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id, - # "operationState": "COMPLETED"}, - # loop=self.loop) - # self.lcm_tasks.lock_HA.assert_called_once_with('ns', 'nslcmops', nslcmop_id) - # if not getenv("OSMLCMTEST_LOGGING_NOMOCK"): - # self.assertTrue(self.my_ns.logger.debug.called, "Debug method not called") - # self.my_ns.logger.error.assert_not_called() - # self.my_ns.logger.exception().assert_not_called() - - # if not getenv("OSMLCMTEST_DB_NOMOCK"): - # self.assertTrue(self.db.set_one.called, "db.set_one not called") - # db_nsr = self.db.get_one("nsrs", {"_id": nsr_id}) - # db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id}) - # self.assertEqual(db_nsr["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated") - # for vnfr in db_vnfrs_list: - # self.assertEqual(vnfr["_admin"].get("nsState"), "INSTANTIATED", "Not instantiated") - - # if not getenv("OSMLCMTEST_VCA_NOMOCK"): - # # check intial-primitives called - # self.assertTrue(self.my_ns.n2vc.exec_primitive.called, - # "Exec primitive not called for initial config primitive") - # for _call in self.my_ns.n2vc.exec_primitive.call_args_list: - # self.assertIn(_call[1]["primitive_name"], ("config", "touch"), - # "called exec primitive with a primitive different than config or touch") - - # # TODO add more checks of called methods - # # TODO add a terminate - - # async def test_instantiate_ee_list(self): - # # Using modern IM where configuration is in the new format of execution_environment_list - # ee_descriptor_id = "charm_simple" - # non_used_initial_primitive = { - # "name": "not_to_be_called", - # "seq": 3, - # "execution-environment-ref": "not_used_ee" - # } - # ee_list = [ - # { - # "id": ee_descriptor_id, - # "juju": {"charm": "simple"}, - - # }, - # ] - - # self.db.set_one( - # "vnfds", - # q_filter={"_id": "7637bcf8-cf14-42dc-ad70-c66fcf1e6e77"}, - # update_dict={"vnf-configuration.0.execution-environment-list": ee_list, - # "vnf-configuration.0.initial-config-primitive.0.execution-environment-ref": ee_descriptor_id, - # "vnf-configuration.0.initial-config-primitive.1.execution-environment-ref": ee_descriptor_id, - # "vnf-configuration.0.initial-config-primitive.2": non_used_initial_primitive, - # "vnf-configuration.0.config-primitive.0.execution-environment-ref": ee_descriptor_id, - # "vnf-configuration.0.config-primitive.0.execution-environment-primitive": "touch_charm", - # }, - # unset={"vnf-configuration.juju": None}) - # 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)