"user": getenv("OSMLCM_VCA_USER", "admin"),
"secret": getenv("OSMLCM_VCA_SECRET", "vca"),
"public_key": getenv("OSMLCM_VCA_PUBKEY", None),
- 'ca_cert': getenv("OSMLCM_VCA_CACERT", None)
+ 'ca_cert': getenv("OSMLCM_VCA_CACERT", None),
+ 'apiproxy': getenv("OSMLCM_VCA_APIPROXY", "192.168.1.1")
},
"ro_config": {
- "endpoint_url": "http://{}:{}/openmano".format(getenv("OSMLCM_RO_HOST", "ro"),
- getenv("OSMLCM_RO_PORT", "9090")),
+ "uri": "http://{}:{}/openmano".format(getenv("OSMLCM_RO_HOST", "ro"),
+ getenv("OSMLCM_RO_PORT", "9090")),
"tenant": getenv("OSMLCM_RO_TENANT", "osm"),
"logger_name": "lcm.ROclient",
"loglevel": "DEBUG",
yield "app_name-{}".format(num_calls)
num_calls += 1
- def _n2vc_CreateExecutionEnvironment(self, namespace, reuse_ee_id, db_dict):
+ def _n2vc_CreateExecutionEnvironment(self, namespace, reuse_ee_id, db_dict, *args, **kwargs):
k_list = namespace.split(".")
ee_id = k_list[1] + "."
if len(k_list) >= 2:
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 = ns.NsLcm(self.db, self.msg, self.fs, self.lcm_tasks, lcm_config, self.loop)
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.execution-environment-list": ee_list,
+ "vnf-configuration.initial-config-primitive.0.execution-environment-ref": ee_descriptor_id,
+ "vnf-configuration.initial-config-primitive.1.execution-environment-ref": ee_descriptor_id,
+ "vnf-configuration.initial-config-primitive.2": non_used_initial_primitive,
+ "vnf-configuration.config-primitive.0.execution-environment-ref": ee_descriptor_id,
+ "vnf-configuration.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
+
def test_ns_params_2_RO(self):
- vim = self.db.get_list("vim_accounts")[0]
- vim_id = vim["_id"]
- ro_vim_id = vim["_admin"]["deployed"]["RO"]
+ vims = self.db.get_list("vim_accounts")
+ vim_id = vims[0]["_id"]
+ ro_vim_id = vims[0]["_admin"]["deployed"]["RO"]
+ vim_id2 = vims[1]["_id"]
+ ro_vim_id2 = vims[1]["_admin"]["deployed"]["RO"]
+
ns_params = {"vimAccountId": vim_id}
mgmt_interface = {"cp": "cp"}
vdu = [{"id": "vdu_id", "interface": [{"external-connection-point-ref": "cp"}]}]
"vnf-configuration": {"config-access": {"ssh-access": {"required": True, "default_user": "U"}}}},
}
nsd = {"constituent-vnfd": []}
+ db_vnfrs = {}
for k in vnfd_dict.keys():
- nsd["constituent-vnfd"].append({"vnfd-id-ref": k, "member-vnf-index": k})
+ nsd["constituent-vnfd"].append({"vnfd-id-ref": k, "member-vnf-index": "member " + k})
+ db_vnfrs["member " + k] = {"vim-account-id": vim_id2 if k == "1" else vim_id}
n2vc_key_list = ["key"]
- ro_ns_params = self.my_ns._ns_params_2_RO(ns_params, nsd, vnfd_dict, n2vc_key_list)
+ ro_ns_params = self.my_ns._ns_params_2_RO(ns_params, nsd, vnfd_dict, db_vnfrs, n2vc_key_list)
ro_params_expected = {'wim_account': None, "datacenter": ro_vim_id,
- "vnfs": {"5": {"vdus": {"vdu_id": {"mgmt_keys": n2vc_key_list}}}}}
+ "vnfs": {
+ "member 5": {"vdus": {"vdu_id": {"mgmt_keys": n2vc_key_list}}},
+ "member 1": {"datacenter": ro_vim_id2}
+ }}
self.assertEqual(ro_ns_params, ro_params_expected)
# Test scale() and related methods
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.k8sclusterhelm.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.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):