from osm_lcm.lcm_utils import TaskRegistry
from osm_lcm.ROclient import ROClient
from uuid import uuid4
-# from asynctest.mock import patch
from osm_lcm.tests import test_db_descriptors as descriptors
"secret": getenv("OSMLCM_VCA_SECRET", "vca"),
"public_key": getenv("OSMLCM_VCA_PUBKEY", None),
'ca_cert': getenv("OSMLCM_VCA_CACERT", None),
- 'apiproxy': getenv("OSMLCM_VCA_APIPROXY", "192.168.1.1")
+ 'apiproxy': getenv("OSMLCM_VCA_APIPROXY", "192.168.1.1"),
},
"ro_config": {
"uri": "http://{}:{}/openmano".format(getenv("OSMLCM_RO_HOST", "ro"),
if not getenv("OSMLCMTEST_VCA_K8s_NOMOCK"):
ns.K8sJujuConnector = asynctest.MagicMock(ns.K8sJujuConnector)
ns.K8sHelmConnector = asynctest.MagicMock(ns.K8sHelmConnector)
+ ns.K8sHelm3Connector = asynctest.MagicMock(ns.K8sHelm3Connector)
if not getenv("OSMLCMTEST_VCA_NOMOCK"):
ns.N2VCJujuConnector = asynctest.MagicMock(ns.N2VCJujuConnector)
- ns.N2VCJujuConnectorLCM = asynctest.MagicMock(ns.N2VCJujuConnectorLCM)
ns.LCMHelmConn = asynctest.MagicMock(ns.LCMHelmConn)
# Create NsLCM class
self.my_ns.n2vc.install_configuration_sw = asynctest.CoroutineMock(return_value=pub_key)
self.my_ns.n2vc.get_ee_ssh_public__key = asynctest.CoroutineMock(return_value=pub_key)
self.my_ns.n2vc.exec_primitive = asynctest.CoroutineMock(side_effect=self._return_uuid)
+ self.my_ns.n2vc.exec_primitive = asynctest.CoroutineMock(side_effect=self._return_uuid)
self.my_ns.n2vc.GetPrimitiveStatus = asynctest.CoroutineMock(return_value="completed")
self.my_ns.n2vc.GetPrimitiveOutput = asynctest.CoroutineMock(return_value={"result": "ok",
"pubkey": pub_key})
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,
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
db_vnfds = {db_vnfd["_id"]: db_vnfd}
task_register = {}
logging_text = "KDU"
- self.my_ns.k8sclusterhelm.install = asynctest.CoroutineMock(return_value="k8s_id")
- self.my_ns.k8sclusterhelm.synchronize_repos = asynctest.CoroutineMock(return_value=("", ""))
+ self.my_ns.k8sclusterhelm3.install = asynctest.CoroutineMock(return_value="k8s_id")
+ self.my_ns.k8sclusterhelm3.synchronize_repos = asynctest.CoroutineMock(return_value=("", ""))
+ self.my_ns.k8sclusterhelm3.get_services = asynctest.CoroutineMock(return_value=([]))
await self.my_ns.deploy_kdus(logging_text, nsr_id, nslcmop_id, db_vnfrs, db_vnfds, task_register)
await asyncio.wait(list(task_register.keys()), timeout=100)
db_nsr = self.db.get_list("nsrs")[1]
self.assertIsInstance(db_nsr["_admin"]["deployed"]["K8s"], list, "K8s entry is not of type list")
self.assertEqual(len(db_nsr["_admin"]["deployed"]["K8s"]), 2, "K8s entry is not of type list")
k8s_instace_info = {"kdu-instance": "k8s_id", "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
- "k8scluster-type": "helm-chart",
+ "k8scluster-type": "helm-chart-v3",
"kdu-name": "ldap", "kdu-model": "stable/openldap:1.2.1",
"member-vnf-index": "multikdu", "namespace": None}
self.assertEqual(db_nsr.get("currentOperationID"), None, "currentOperationID different than None")
self.assertEqual(db_nsr.get("errorDescription "), None, "errorDescription different than None")
self.assertEqual(db_nsr.get("errorDetail"), None, "errorDetail different than None")
+ db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
+ for vnfr in db_vnfrs_list:
+ self.assertEqual(vnfr["_admin"].get("nsState"), "NOT_INSTANTIATED", "Not instantiated")
@asynctest.fail_on(active_handles=True) # all async tasks must be completed
async def test_terminate_primitive(self):