"""
# parent class
- K8sConnector.__init__(
- self,
- db,
- log=log,
- on_update_db=on_update_db,
- )
+ K8sConnector.__init__(self, db, log=log, on_update_db=on_update_db)
self.fs = fs
self.loop = loop or asyncio.get_event_loop()
cleanup_data = []
try:
self.log.debug("Initializing K8s cluster for juju")
- kubectl.create_cluster_role(
- name=metadata_name,
- labels=labels,
- )
+ kubectl.create_cluster_role(name=metadata_name, labels=labels)
self.log.debug("Cluster role created")
cleanup_data.append(
- {
- "delete": kubectl.delete_cluster_role,
- "args": (metadata_name,),
- }
+ {"delete": kubectl.delete_cluster_role, "args": (metadata_name,)}
)
- kubectl.create_service_account(
- name=metadata_name,
- labels=labels,
- )
+ kubectl.create_service_account(name=metadata_name, labels=labels)
self.log.debug("Service account created")
cleanup_data.append(
- {
- "delete": kubectl.delete_service_account,
- "args": (metadata_name,),
- }
+ {"delete": kubectl.delete_service_account, "args": (metadata_name,)}
)
- kubectl.create_cluster_role_binding(
- name=metadata_name,
- labels=labels,
- )
+ kubectl.create_cluster_role_binding(name=metadata_name, labels=labels)
self.log.debug("Role binding created")
cleanup_data.append(
{
"args": (metadata_name,),
}
)
- token, client_cert_data = await kubectl.get_secret_data(
- metadata_name,
- )
+ token, client_cert_data = await kubectl.get_secret_data(metadata_name)
default_storage_class = kubectl.get_default_storage_class()
self.log.debug("Default storage class: {}".format(default_storage_class))
async def repo_list(self):
raise MethodNotImplemented()
- async def repo_remove(
- self,
- name: str,
- ):
+ async def repo_remove(self, name: str):
raise MethodNotImplemented()
async def synchronize_repos(self, cluster_uuid: str, name: str):
return True
async def get_scale_count(
- self,
- resource_name: str,
- kdu_instance: str,
- namespace: str = None,
- **kwargs,
+ self, resource_name: str, kdu_instance: str, namespace: str = None, **kwargs
) -> int:
"""Get an application scale count
"""Rollback"""
async def rollback(
- self,
- cluster_uuid: str,
- kdu_instance: str,
- revision: int = 0,
+ self, cluster_uuid: str, kdu_instance: str, revision: int = 0
) -> str:
"""Rollback a model
"""Deletion"""
async def uninstall(
- self,
- cluster_uuid: str,
- kdu_instance: str,
- namespace: str = None,
- **kwargs,
+ self, cluster_uuid: str, kdu_instance: str, namespace: str = None, **kwargs
) -> bool:
"""Uninstall a KDU instance
"""Introspection"""
- async def inspect_kdu(
- self,
- kdu_model: str,
- ) -> dict:
+ async def inspect_kdu(self, kdu_model: str) -> dict:
"""Inspect a KDU
Inspects a bundle and returns a dictionary of config parameters and
return kdu
- async def help_kdu(
- self,
- kdu_model: str,
- ) -> str:
+ async def help_kdu(self, kdu_model: str) -> str:
"""View the README
If available, returns the README of the bundle.
return status
async def add_relation(
- self,
- provider: RelationEndpoint,
- requirer: RelationEndpoint,
+ self, provider: RelationEndpoint, requirer: RelationEndpoint
):
"""
Add relation between two charmed endpoints
self.log.debug(f"adding new relation between {provider} and {requirer}")
cross_model_relation = (
provider.model_name != requirer.model_name
- or requirer.vca_id != requirer.vca_id
+ or provider.vca_id != requirer.vca_id
)
try:
if cross_model_relation:
requirer.model_name, offer, provider_libjuju
)
await requirer_libjuju.add_relation(
- requirer.model_name,
- requirer.endpoint,
- saas_name,
+ requirer.model_name, requirer.endpoint, saas_name
)
else:
# Standard relation
"""
return "cred-{}".format(cluster_uuid)
- def get_namespace(
- self,
- cluster_uuid: str,
- ) -> str:
+ def get_namespace(self, cluster_uuid: str) -> str:
"""Get the namespace UUID
Gets the namespace's unique name
return self.libjuju
else:
vca_connection = await get_connection(self._store, vca_id)
- return Libjuju(
- vca_connection,
- loop=self.loop,
- log=self.log,
- n2vc=self,
- )
+ return Libjuju(vca_connection, loop=self.loop, log=self.log, n2vc=self)
def _get_kubectl(self, credentials: str) -> Kubectl:
"""
# parent class constructor
N2VCConnector.__init__(
- self,
- db=db,
- fs=fs,
- log=log,
- loop=loop,
- on_update_db=on_update_db,
+ self, db=db, fs=fs, log=log, loop=loop, on_update_db=on_update_db
)
# silence websocket traffic log
# create or reuse a new juju machine
try:
if not await libjuju.model_exists(model_name):
- await libjuju.add_model(
- model_name,
- libjuju.vca_connection.lxd_cloud,
- )
+ await libjuju.add_model(model_name, libjuju.vca_connection.lxd_cloud)
machine, new = await libjuju.create_machine(
model_name=model_name,
machine_id=machine_id,
raise N2VCException(message=message)
# new machine credentials
- credentials = {
- "hostname": machine.dns_name,
- }
+ credentials = {"hostname": machine.dns_name}
self.log.info(
"Execution environment created. ee_id: {}, credentials: {}".format(
# register machine on juju
try:
if not await libjuju.model_exists(model_name):
- await libjuju.add_model(
- model_name,
- libjuju.vca_connection.lxd_cloud,
- )
+ await libjuju.add_model(model_name, libjuju.vca_connection.lxd_cloud)
machine_id = await libjuju.provision_machine(
model_name=model_name,
hostname=hostname,
_, ns_id, _, _, _ = self._get_namespace_components(namespace=namespace)
model_name = "{}-k8s".format(ns_id)
if not await libjuju.model_exists(model_name):
- await libjuju.add_model(
- model_name,
- libjuju.vca_connection.k8s_cloud,
- )
+ await libjuju.add_model(model_name, libjuju.vca_connection.k8s_cloud)
application_name = self._get_application_name(namespace)
try:
self.log.info("K8s proxy charm installed")
ee_id = N2VCJujuConnector._build_ee_id(
- model_name=model_name,
- application_name=application_name,
- machine_id="k8s",
+ model_name=model_name, application_name=application_name, machine_id="k8s"
)
self._write_ee_id_db(db_dict=db_dict, ee_id=ee_id)
return await libjuju.get_metrics(model_name, application_name)
async def add_relation(
- self,
- provider: RelationEndpoint,
- requirer: RelationEndpoint,
+ self, provider: RelationEndpoint, requirer: RelationEndpoint
):
"""
Add relation between two charmed endpoints
self.log.debug(f"adding new relation between {provider} and {requirer}")
cross_model_relation = (
provider.model_name != requirer.model_name
- or requirer.vca_id != requirer.vca_id
+ or provider.vca_id != requirer.vca_id
)
try:
if cross_model_relation:
requirer.model_name, offer, provider_libjuju
)
await requirer_libjuju.add_relation(
- requirer.model_name,
- requirer.endpoint,
- saas_name,
+ requirer.model_name, requirer.endpoint, saas_name
)
else:
# Standard relation
if not scaling_in:
# destroy the model
await libjuju.destroy_model(
- model_name=model_name,
- total_timeout=total_timeout,
+ model_name=model_name, total_timeout=total_timeout
)
elif vca_type == "native_charm" and scaling_in:
# destroy the unit in the application
config=params_dict,
)
actions = await libjuju.get_actions(
- application_name=application_name,
- model_name=model_name,
+ application_name=application_name, model_name=model_name
)
self.log.debug(
"Application {} has these actions: {}".format(
return self.libjuju
else:
vca_connection = await get_connection(self._store, vca_id)
- return Libjuju(
- vca_connection,
- loop=self.loop,
- log=self.log,
- n2vc=self,
- )
+ return Libjuju(vca_connection, loop=self.loop, log=self.log, n2vc=self)
def _write_ee_id_db(self, db_dict: dict, ee_id: str):
from n2vc.k8s_juju_conn import K8sJujuConnector, RBAC_LABEL_KEY_NAME
from osm_common import fslocal
from .utils import kubeconfig, FakeModel, FakeFileWrapper, AsyncMock, FakeApplication
-from n2vc.exceptions import (
- MethodNotImplemented,
- K8sException,
-)
+from n2vc.exceptions import MethodNotImplemented, K8sException
from n2vc.vca.connection_data import ConnectionData
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
- self.cs_bundle,
- model_name=self.default_namespace,
- wait=True,
- timeout=1800,
+ self.cs_bundle, model_name=self.default_namespace, wait=True, timeout=1800
)
def test_success_http(self, mock_chdir):
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
- self.http_bundle,
- model_name=self.default_namespace,
- wait=True,
- timeout=1800,
+ self.http_bundle, model_name=self.default_namespace, wait=True, timeout=1800
)
def test_success_not_kdu_name(self, mock_chdir):
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
- self.cs_bundle,
- model_name=self.default_namespace,
- wait=True,
- timeout=1800,
+ self.cs_bundle, model_name=self.default_namespace, wait=True, timeout=1800
)
def test_missing_db_dict(self, mock_chdir):
)
self.k8s_juju_conn.libjuju.add_model.assert_called_once()
self.k8s_juju_conn.libjuju.deploy.assert_called_once_with(
- self.cs_bundle,
- model_name=self.default_namespace,
- wait=True,
- timeout=1800,
+ self.cs_bundle, model_name=self.default_namespace, wait=True, timeout=1800
)
def test_missing_bundle(self, mock_chdir):
self.k8s_juju_conn.libjuju.get_controller = AsyncMock()
self.k8s_juju_conn.libjuju.consume = AsyncMock()
- def test_standard_relation(self):
- relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
- relation_endpoint_2 = RelationEndpoint("model-1.app2.1", None, "endpoint")
+ def test_standard_relation_same_model_and_controller(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint1")
+ relation_endpoint_2 = RelationEndpoint("model-1.app2.1", None, "endpoint2")
self.loop.run_until_complete(
self.k8s_juju_conn.add_relation(relation_endpoint_1, relation_endpoint_2)
)
self.k8s_juju_conn.libjuju.add_relation.assert_called_once_with(
- model_name="model-1", endpoint_1="app1:endpoint", endpoint_2="app2:endpoint"
+ model_name="model-1",
+ endpoint_1="app1:endpoint1",
+ endpoint_2="app2:endpoint2",
)
self.k8s_juju_conn.libjuju.offer.assert_not_called()
self.k8s_juju_conn.libjuju.consume.assert_not_called()
"model-2", "app2:endpoint", "saas"
)
+ def test_cmr_relation_different_controller(self):
+ self.k8s_juju_conn._get_libjuju = AsyncMock(
+ return_value=self.k8s_juju_conn.libjuju
+ )
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", "vca-id-1", "endpoint")
+ relation_endpoint_2 = RelationEndpoint("model-1.app2.1", "vca-id-2", "endpoint")
+ offer = Offer("admin/model-1.app1")
+ self.k8s_juju_conn.libjuju.offer.return_value = offer
+ self.k8s_juju_conn.libjuju.consume.return_value = "saas"
+ self.loop.run_until_complete(
+ self.k8s_juju_conn.add_relation(relation_endpoint_1, relation_endpoint_2)
+ )
+ self.k8s_juju_conn.libjuju.offer.assert_called_once_with(relation_endpoint_1)
+ self.k8s_juju_conn.libjuju.consume.assert_called_once()
+ self.k8s_juju_conn.libjuju.add_relation.assert_called_once_with(
+ "model-1", "app2:endpoint", "saas"
+ )
+
def test_relation_exception(self):
relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
@asynctest.mock.patch("n2vc.n2vc_juju_conn.get_connection")
@asynctest.mock.patch("n2vc.vca.connection_data.base64_to_cacert")
def setUp(
- self,
- mock_base64_to_cacert=None,
- mock_get_connection=None,
- mock_store=None,
+ self, mock_base64_to_cacert=None, mock_get_connection=None, mock_store=None
):
self.loop = asyncio.get_event_loop()
self.db = Mock()
"n2vc.n2vc_juju_conn.generate_random_alfanum_string",
**{"return_value": "random"}
)
- def test_success(
- self,
- mock_generate_random_alfanum_string,
- ):
+ def test_success(self, mock_generate_random_alfanum_string):
self.n2vc.fs.file_exists = MagicMock(create_autospec=True)
self.n2vc.fs.file_exists.return_value = True
ee_id = self.loop.run_until_complete(
self.n2vc.libjuju.get_controller = AsyncMock()
self.n2vc.libjuju.consume = AsyncMock()
- def test_standard_relation(self):
- relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
- relation_endpoint_2 = RelationEndpoint("model-1.app2.1", None, "endpoint")
+ def test_standard_relation_same_model_and_controller(self):
+ relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint1")
+ relation_endpoint_2 = RelationEndpoint("model-1.app2.1", None, "endpoint2")
self.loop.run_until_complete(
self.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2)
)
self.n2vc.libjuju.add_relation.assert_called_once_with(
- model_name="model-1", endpoint_1="app1:endpoint", endpoint_2="app2:endpoint"
+ model_name="model-1",
+ endpoint_1="app1:endpoint1",
+ endpoint_2="app2:endpoint2",
)
self.n2vc.libjuju.offer.assert_not_called()
self.n2vc.libjuju.consume.assert_not_called()
"model-2", "app2:endpoint", "saas"
)
+ def test_cmr_relation_different_controller(self):
+ self.n2vc._get_libjuju = AsyncMock(return_value=self.n2vc.libjuju)
+ relation_endpoint_1 = RelationEndpoint(
+ "model-1.app1.0", "vca-id-1", "endpoint1"
+ )
+ relation_endpoint_2 = RelationEndpoint(
+ "model-1.app2.1", "vca-id-2", "endpoint2"
+ )
+ offer = Offer("admin/model-1.app1")
+ self.n2vc.libjuju.offer.return_value = offer
+ self.n2vc.libjuju.consume.return_value = "saas"
+ self.loop.run_until_complete(
+ self.n2vc.add_relation(relation_endpoint_1, relation_endpoint_2)
+ )
+ self.n2vc.libjuju.offer.assert_called_once_with(relation_endpoint_1)
+ self.n2vc.libjuju.consume.assert_called_once()
+ self.n2vc.libjuju.add_relation.assert_called_once_with(
+ "model-1", "app2:endpoint2", "saas"
+ )
+
def test_relation_exception(self):
relation_endpoint_1 = RelationEndpoint("model-1.app1.0", None, "endpoint")
relation_endpoint_2 = RelationEndpoint("model-2.app2.1", None, "endpoint")
"ee_id": None,
"application": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
- },
+ }
]
vnf_count = ""
vdu_count = ""
def test_generate_application_name_vnf_charm(self):
charm_level = "vnf-level"
- vnfrs = {
- "member-vnf-index-ref": "vnf111-xxx-yyy-zzz",
- }
+ vnfrs = {"member-vnf-index-ref": "vnf111-xxx-yyy-zzz"}
vca_records = [
{
"target_element": "vnf/vnf1",
"ee_descriptor_id": "simple-ee-abc-000-rrrr-nnnn-4444-hhh-3333-yyyy-333-hhh-ttt-444",
"charm_name": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
- },
+ }
]
vnf_count = "1"
vdu_count = ""
self,
):
charm_level = "vdu-level"
- vnfrs = {
- "member-vnf-index-ref": "vnf111-xxx-yyy-zzz",
- }
+ vnfrs = {"member-vnf-index-ref": "vnf111-xxx-yyy-zzz"}
vca_records = [
{
"target_element": "vnf/vnf1/mgmtVM",
"ee_descriptor_id": "simple-ee-abc-000-rrrr-nnnn-4444-hhh-3333-yyyy-333-hhh-ttt-444",
"charm_name": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
- },
+ }
]
vnf_count = "2"
vdu_count = "0"
def test_generate_application_name_vdu_charm_given_vdu_id_is_none(self):
charm_level = "vdu-level"
- vnfrs = {
- "member-vnf-index-ref": "vnf111-xxx-yyy-zzz",
- }
+ vnfrs = {"member-vnf-index-ref": "vnf111-xxx-yyy-zzz"}
vca_records = [
{
"target_element": "vnf/vnf1/mgmtvVM",
"ee_descriptor_id": "simple-ee-abc-000-rrrr-nnnn-4444-hhh-3333-yyyy-333-hhh-ttt-444",
"charm_name": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
- },
+ }
]
vnf_count = "2"
vdu_count = "0"
self,
):
charm_level = "vdu-level"
- vnfrs = {
- "member-vnf-index-ref": "vnf111-xxx-yyy-zzz",
- }
+ vnfrs = {"member-vnf-index-ref": "vnf111-xxx-yyy-zzz"}
vca_records = [
{
"target_element": "vnf/vnf1/mgmtVM",
"ee_descriptor_id": "simple-ee-abc-000-rrrr-nnnn-4444-hhh-3333-yyyy-333-hhh-ttt-444",
"charm_name": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
- },
+ }
]
vnf_count = "2"
vdu_count = "0"
def test_generate_application_name_vdu_charm_vdu_id_in_vca_record_is_none(self):
charm_level = "vdu-level"
- vnfrs = {
- "member-vnf-index-ref": "vnf111-xxx-yyy-zzz",
- }
+ vnfrs = {"member-vnf-index-ref": "vnf111-xxx-yyy-zzz"}
vca_records = [
{
"target_element": "vnf/vnf1/mgmtVM",
"ee_descriptor_id": "simple-ee-abc-000-rrrr-nnnn-4444-hhh-3333-yyyy-333-hhh-ttt-444",
"charm_name": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
- },
+ }
]
vnf_count = "2"
vdu_count = "0"
"charm_name": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
},
- ],
- },
- },
+ ]
+ }
+ }
}
expected_result = [
{
"ee_descriptor_id": "simple-ee-abc-000-rrrr-nnnn-4444-hhh-3333-yyyy-333-hhh-ttt-444",
"charm_name": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
- },
+ }
]
vca_records = self.n2vc._get_vca_records(charm_level, db_nsr, db_vnfr)
self.assertEqual(vca_records, expected_result)
def test_get_vca_records_vnf_charm_member_vnf_index_mismatch(self):
charm_level = "vnf-level"
- db_vnfr = {
- "member-vnf-index-ref": "vnf222-xxx-yyy-zzz",
- }
+ db_vnfr = {"member-vnf-index-ref": "vnf222-xxx-yyy-zzz"}
db_nsr = {
"_admin": {
"deployed": {
"charm_name": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
},
- ],
- },
- },
+ ]
+ }
+ }
}
expected_result = []
vca_records = self.n2vc._get_vca_records(charm_level, db_nsr, db_vnfr)
def test_get_vca_records_ns_charm(self):
charm_level = "ns-level"
- db_vnfr = {
- "member-vnf-index-ref": "vnf222-xxx-yyy-zzz",
- }
+ db_vnfr = {"member-vnf-index-ref": "vnf222-xxx-yyy-zzz"}
db_nsr = {
"_admin": {
"deployed": {
"charm_name": "simple-ns-charm-abc-000-rrrr-nnnn-4444-hhh-3333-yyyy-333-hhh-ttt-444",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
},
- ],
- },
- },
+ ]
+ }
+ }
}
expected_result = [
{
"ee_descriptor_id": "",
"charm_name": "simple-ns-charm-abc-000-rrrr-nnnn-4444-hhh-3333-yyyy-333-hhh-ttt-444",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
- },
+ }
]
vca_records = self.n2vc._get_vca_records(charm_level, db_nsr, db_vnfr)
self.assertEqual(vca_records, expected_result)
def test_get_vca_records_ns_charm_empty_charm_name(self):
charm_level = "ns-level"
- db_vnfr = {
- "member-vnf-index-ref": "vnf222-xxx-yyy-zzz",
- }
+ db_vnfr = {"member-vnf-index-ref": "vnf222-xxx-yyy-zzz"}
db_nsr = {
"_admin": {
"deployed": {
"charm_name": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
},
- ],
- },
- },
+ ]
+ }
+ }
}
expected_result = [
{
"ee_descriptor_id": "",
"charm_name": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
- },
+ }
]
vca_records = self.n2vc._get_vca_records(charm_level, db_nsr, db_vnfr)
self.assertEqual(vca_records, expected_result)
"charm_name": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
},
- ],
- },
- },
+ ]
+ }
+ }
}
mock_vnf_count_and_record = MagicMock()
- db_vnfr = {
- "member-vnf-index-ref": "vnf111-xxx-yyy-zzz",
- }
+ db_vnfr = {"member-vnf-index-ref": "vnf111-xxx-yyy-zzz"}
vnf_count = "0"
mock_vnf_count_and_record.return_value = (vnf_count, db_vnfr)
expected_result = "simple-ee-ab-z0-vnf111-xxx-y-vnf"
"ee_descriptor_id": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
},
- ],
- },
- },
+ ]
+ }
+ }
}
mock_vnf_count_and_record = MagicMock()
- db_vnfr = {
- "member-vnf-index-ref": "vnf111-xxx-yyy-zzz",
- }
+ db_vnfr = {"member-vnf-index-ref": "vnf111-xxx-yyy-zzz"}
vnf_count = "0"
mock_vnf_count_and_record.return_value = (vnf_count, db_vnfr)
expected_result = "app-vnf-eb3161eec0-z0-random"
"charm_name": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
},
- ],
- },
- },
+ ]
+ }
+ }
}
mock_vnf_count_and_record = MagicMock()
- db_vnfr = {
- "member-vnf-index-ref": "vnf222-xxx-yyy-zzz",
- }
+ db_vnfr = {"member-vnf-index-ref": "vnf222-xxx-yyy-zzz"}
vnf_count = "0"
mock_vnf_count_and_record.return_value = (vnf_count, db_vnfr)
with patch.object(self.n2vc, "db", self.db), patch.object(
"application": "openldap-ee-z0-openldap-vdu",
"model": "82b11965-e580-47c0-9ee0-329f318a305b",
"config_sw_installed": True,
- },
+ }
]
}
}
}
mock_vnf_count_and_record = MagicMock()
- db_vnfr = {
- "member-vnf-index-ref": "openldap",
- "vdur": {},
- }
+ db_vnfr = {"member-vnf-index-ref": "openldap", "vdur": {}}
vnf_count = "0"
mock_vnf_count_and_record.return_value = (vnf_count, db_vnfr)
expected_result = "openldap-ee-z0-openldap-ldap-vdu"
"ee_descriptor_id": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
},
- ],
- },
- },
+ ]
+ }
+ }
}
mock_vnf_count_and_record = MagicMock()
db_vnfr = {
"ee_descriptor_id": "",
"charm_name": "simple-ns-charm-abc-000-rrrr-nnnn-4444-hhh-3333-yyyy-333-hhh-ttt-444",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
- },
- ],
- },
- },
+ }
+ ]
+ }
+ }
}
mock_vnf_count_and_record = MagicMock()
db_vnfr = {}
"ee_descriptor_id": "",
"charm_name": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
- },
- ],
- },
- },
+ }
+ ]
+ }
+ }
}
mock_vnf_count_and_record = MagicMock()
db_vnfr = {}
"vdu_name": "",
"ee_descriptor_id": "",
"model": "dbfbd751-3de4-4e68-bd40-ec5ae0a53898",
- },
- ],
- },
- },
+ }
+ ]
+ }
+ }
}
mock_vnf_count_and_record = MagicMock()
db_vnfr = {}
projects / osm / N2VC.git / commitdiff