"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)
# TODO add a terminate
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
self.assertEqual(return_value, expected_value)
# print("scale_result: {}".format(self.db.get_one("nslcmops", {"_id": nslcmop_id}).get("detailed-status")))
- # Test _reintent_or_skip_suboperation()
+ # Test _retry_or_skip_suboperation()
# Expected result:
# - if a suboperation's 'operationState' is marked as 'COMPLETED', SUBOPERATION_STATUS_SKIP is expected
# - if marked as anything but 'COMPLETED', the suboperation index is expected
- def test_scale_reintent_or_skip_suboperation(self):
+ def test_scale_retry_or_skip_suboperation(self):
# Load an alternative 'nslcmops' YAML for this test
nslcmop_id = descriptors.test_ids["TEST-A"]["instantiate"]
db_nslcmop = self.db.get_one('nslcmops', {"_id": nslcmop_id})
op_index = 2
# Test when 'operationState' is 'COMPLETED'
db_nslcmop['_admin']['operations'][op_index]['operationState'] = 'COMPLETED'
- return_value = self.my_ns._reintent_or_skip_suboperation(db_nslcmop, op_index)
+ return_value = self.my_ns._retry_or_skip_suboperation(db_nslcmop, op_index)
expected_value = self.my_ns.SUBOPERATION_STATUS_SKIP
self.assertEqual(return_value, expected_value)
# Test when 'operationState' is not 'COMPLETED'
db_nslcmop['_admin']['operations'][op_index]['operationState'] = None
- return_value = self.my_ns._reintent_or_skip_suboperation(db_nslcmop, op_index)
+ return_value = self.my_ns._retry_or_skip_suboperation(db_nslcmop, op_index)
expected_value = op_index
self.assertEqual(return_value, expected_value)
self.assertIn("K8s", db_nsr["_admin"]["deployed"], "K8s entry not created at '_admin.deployed'")
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": None, "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
+ k8s_instace_info = {"kdu-instance": "k8s_id", "k8scluster-uuid": "73d96432-d692-40d2-8440-e0c73aee209c",
"k8scluster-type": "helm-chart",
"kdu-name": "ldap", "kdu-model": "stable/openldap:1.2.1",
- "member-vnf-index": "multikdu"}
+ "member-vnf-index": "multikdu", "namespace": None}
self.assertEqual(db_nsr["_admin"]["deployed"]["K8s"][0], k8s_instace_info)
k8s_instace_info["kdu-name"] = "mongo"