import asynctest
import asyncio
+from unittest import TestCase
import unittest.mock as mock
from juju.application import Application
app_name = "my_app_name"
channel = "latest"
entity_url = "ch:my-charm"
+ cloud_k8s = "microk8s"
+ cloud_other = "other"
charm_info = CharmInfo(app_name, channel, entity_url)
- constraints = VduComputeConstraints(1, 2)
- vdu_instantiate_input = VduInstantiateInput(
- vim_content["_id"], namespace, charm_info, constraints, cloud=""
+ vdu_instantiate_input_with_constraints_k8s = VduInstantiateInput(
+ vim_content["_id"],
+ namespace,
+ charm_info,
+ VduComputeConstraints(mem=1, cores=1),
+ cloud_k8s,
)
- async def test_deploy_charm_nominal_case(self):
- await self.env.run(self.juju_paas.deploy_charm, self.vdu_instantiate_input)
+ async def test_deploy_charm_with_constraints_k8s_cloud(self):
+ await self.env.run(
+ self.juju_paas.deploy_charm, self.vdu_instantiate_input_with_constraints_k8s
+ )
+ self.model.deploy.assert_called_once_with(
+ entity_url=self.entity_url,
+ application_name=self.app_name,
+ channel=self.channel,
+ constraints={"mem": 1024},
+ )
+
+ async def test_deploy_charm_with_constraints_other_cloud(self):
+ await self.env.run(
+ self.juju_paas.deploy_charm,
+ VduInstantiateInput(
+ vim_content["_id"],
+ namespace,
+ self.charm_info,
+ VduComputeConstraints(mem=1, cores=1),
+ self.cloud_other,
+ ),
+ )
self.model.deploy.assert_called_once_with(
entity_url=self.entity_url,
application_name=self.app_name,
channel=self.channel,
+ constraints={"mem": 1024, "cores": 1},
+ )
+
+ async def test_deploy_charm_without_constraints_k8s_cloud(self):
+ await self.env.run(
+ self.juju_paas.deploy_charm,
+ VduInstantiateInput(
+ vim_content["_id"],
+ namespace,
+ self.charm_info,
+ VduComputeConstraints(mem=0, cores=0),
+ self.cloud_k8s,
+ ),
+ )
+ self.model.deploy.assert_called_once_with(
+ entity_url=self.entity_url,
+ application_name=self.app_name,
+ channel=self.channel,
+ constraints=None,
+ )
+
+ async def test_deploy_charm_without_constraints_other_cloud(self):
+ await self.env.run(
+ self.juju_paas.deploy_charm,
+ VduInstantiateInput(
+ vim_content["_id"],
+ namespace,
+ self.charm_info,
+ VduComputeConstraints(mem=0, cores=0),
+ self.cloud_other,
+ ),
+ )
+ self.model.deploy.assert_called_once_with(
+ entity_url=self.entity_url,
+ application_name=self.app_name,
+ channel=self.channel,
+ constraints=None,
)
async def test_deploy_charm_app_already_exists(self):
self.add_application(self.app_name)
with self.assertRaises(Exception) as err:
- await self.env.run(self.juju_paas.deploy_charm, self.vdu_instantiate_input)
+ await self.env.run(
+ self.juju_paas.deploy_charm,
+ self.vdu_instantiate_input_with_constraints_k8s,
+ )
self.model.deploy.assert_not_called()
self.assertEqual(
str(err.exception.args[0]),
async def test_deploy_charm_raises_exception(self):
self.controller.get_model.side_effect = JujuError()
with self.assertRaises(JujuError):
- await self.env.run(self.juju_paas.deploy_charm, self.vdu_instantiate_input)
+ await self.env.run(
+ self.juju_paas.deploy_charm,
+ self.vdu_instantiate_input_with_constraints_k8s,
+ )
self.model.deploy.assert_not_called()
+class TestGetApplicationConstraints(TestCase):
+ constraints = VduComputeConstraints(mem=1, cores=1)
+ no_constraints = VduComputeConstraints(mem=0, cores=0)
+
+ def test_get_application_constraints_k8s_cloud(self):
+ result = JujuPaasConnector._get_application_constraints(
+ self.constraints, "kubernetes"
+ )
+ self.assertEqual(result, {"mem": 1024})
+
+ def test_get_application_constraints_aws_cloud(self):
+ result = JujuPaasConnector._get_application_constraints(self.constraints, "aws")
+ self.assertEqual(result, {"cores": 1, "mem": 1024})
+
+ def test_get_application_constraints_no_constraints_aws(self):
+ result = JujuPaasConnector._get_application_constraints(
+ self.no_constraints, "aws"
+ )
+ self.assertEqual(result, {})
+
+ def test_get_application_constraints_no_constraints_microk8s(self):
+ result = JujuPaasConnector._get_application_constraints(
+ self.no_constraints, "microk8s"
+ )
+ self.assertEqual(result, {})
+
+ def test_get_application_constraints_empty_cloud(self):
+ result = JujuPaasConnector._get_application_constraints(self.constraints, "")
+ self.assertEqual(result, {"cores": 1, "mem": 1024})
+
+
class TestTestVimConnectivity(TestJujuPaasActivitiesBase):
test_vim_connectivity_input = TestVimConnectivityInput(vim_content["_id"])
from asyncio.exceptions import CancelledError
from copy import deepcopy
from osm_common.dataclasses.temporal_dataclasses import (
+ GetVimCloudInput,
GetVnfDetailsInput,
VnfInstantiateInput,
)
vnfr_uuid = "d08d2da5-2120-476c-8538-deaeb4e88b3e"
model_name = "a-model-name"
vnf_instantiate_input = VnfInstantiateInput(vnfr_uuid=vnfr_uuid, model_name=model_name)
+cloud = "microk8s"
+sample_vnfr = {
+ "_id": "9f472177-95c0-4335-b357-5cdc17a79965",
+ "id": "9f472177-95c0-4335-b357-5cdc17a79965",
+ "nsr-id-ref": "dcf4c922-5a73-41bf-a6ca-870c22d6336c",
+ "vnfd-ref": "jar_vnfd_scalable",
+ "vnfd-id": "f1b38eac-190c-485d-9a74-c6e169c929d8",
+ "vim-account-id": "9b0bedc3-ea8e-42fd-acc9-dd03f4dee73c",
+}
+sample_vnfd = {
+ "_id": "97784f19-d254-4252-946c-cf92d85443ca",
+ "id": "sol006-vnf",
+ "provider": "Canonical",
+ "product-name": "test-vnf",
+ "software-version": "1.0",
+}
+sample_vim_record = {
+ "_id": "a64f7c6c-bc27-4ec8-b664-5500a3324eca",
+ "name": "juju",
+ "vim_type": "paas",
+ "vim_url": "192.168.1.100:17070",
+ "vim_user": "admin",
+ "vim_password": "c16gylWEepEREN6vWw==",
+ "config": {
+ "paas_provider": "juju",
+ "cloud": cloud,
+ "cloud_credentials": "microk8s",
+ "authorized_keys": "$HOME/.local/share/juju/ssh/juju_id_rsa.pub",
+ "ca_cert_content": "-----BEGIN-----",
+ },
+}
class TestVnfDbActivity(asynctest.TestCase):
)
-sample_vnfr = {
- "_id": "9f472177-95c0-4335-b357-5cdc17a79965",
- "id": "9f472177-95c0-4335-b357-5cdc17a79965",
- "nsr-id-ref": "dcf4c922-5a73-41bf-a6ca-870c22d6336c",
- "vnfd-ref": "jar_vnfd_scalable",
- "vnfd-id": "f1b38eac-190c-485d-9a74-c6e169c929d8",
- "vim-account-id": "9b0bedc3-ea8e-42fd-acc9-dd03f4dee73c",
-}
-
-sample_vnfd = {
- "_id": "97784f19-d254-4252-946c-cf92d85443ca",
- "id": "sol006-vnf",
- "provider": "Canonical",
- "product-name": "test-vnf",
- "software-version": "1.0",
-}
-
-
class TestVnfDetails(asynctest.TestCase):
async def setUp(self):
self.db = Mock()
self.assertEqual(activity_result, None)
+class TestGetVimCloud(asynctest.TestCase):
+ async def setUp(self):
+ self.db = Mock()
+ self.vnf_operations_instance = VnfOperations(self.db)
+ self.env = ActivityEnvironment()
+
+ async def test_activity_succeeded(self):
+ self.db.get_one.side_effect = [sample_vnfr, sample_vim_record]
+ activity_result = await self.env.run(
+ self.vnf_operations_instance.get_vim_cloud,
+ GetVimCloudInput(vnfr_uuid=sample_vnfr["id"]),
+ )
+ self.assertEqual(activity_result.cloud, cloud)
+
+ async def test_activity_vim_record_without_cloud(self):
+ vim_record = deepcopy(sample_vim_record)
+ vim_record["config"].pop("cloud")
+ self.db.get_one.side_effect = [sample_vnfr, vim_record]
+ activity_result = await self.env.run(
+ self.vnf_operations_instance.get_vim_cloud,
+ GetVimCloudInput(vnfr_uuid=sample_vnfr["id"]),
+ )
+ self.assertEqual(activity_result.cloud, "")
+
+ async def test_activity_failed_db_exception(self):
+ self.db.get_one.side_effect = DbException("Can not connect to Database.")
+ with self.assertRaises(DbException) as err:
+ activity_result = await self.env.run(
+ self.vnf_operations_instance.get_vim_cloud,
+ GetVimCloudInput(vnfr_uuid=sample_vnfr["id"]),
+ )
+ self.assertEqual(activity_result, None)
+ self.assertEqual(
+ str(err.exception), "database exception Can not connect to Database."
+ )
+
+ async def test_activity_failed_key_error(self):
+ vim_record = deepcopy(sample_vim_record)
+ vim_record.pop("config")
+ self.db.get_one.side_effect = [sample_vnfr, vim_record]
+ with self.assertRaises(KeyError) as err:
+ activity_result = await self.env.run(
+ self.vnf_operations_instance.get_vim_cloud,
+ GetVimCloudInput(vnfr_uuid=sample_vnfr["id"]),
+ )
+ self.assertEqual(activity_result, None)
+ self.assertEqual(str(err.exception.args[0]), "config")
+
+ async def test_activity_cancelled(self):
+ self.env._cancelled = True
+ self.db.get_one.side_effect = [sample_vnfr, sample_vim_record]
+ with self.assertRaises(CancelledError):
+ activity_result = await self.env.run(
+ self.vnf_operations_instance.get_vim_cloud,
+ GetVimCloudInput(vnfr_uuid=sample_vnfr["id"]),
+ )
+ self.assertEqual(activity_result, None)
+
+
if __name__ == "__main__":
asynctest.main()
# limitations under the License.
import asynctest
+from copy import deepcopy
from temporalio import activity
from temporalio import workflow
from temporalio.client import WorkflowFailureError
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
-from unittest.mock import Mock
+from unittest import TestCase
+from unittest.mock import Mock, patch, AsyncMock
from osm_common.dataclasses.temporal_dataclasses import (
ChangeVnfInstantiationStateInput,
CharmInfo,
GetTaskQueueInput,
GetTaskQueueOutput,
+ GetVimCloudInput,
+ GetVimCloudOutput,
GetVnfDetailsInput,
GetVnfDetailsOutput,
VduComputeConstraints,
ACTIVITY_CHANGE_VNF_STATE,
ACTIVITY_CHANGE_VNF_INSTANTIATION_STATE,
ACTIVITY_GET_TASK_QUEUE,
+ ACTIVITY_GET_VIM_CLOUD,
ACTIVITY_GET_VNF_DETAILS,
ACTIVITY_SEND_NOTIFICATION_FOR_VNF,
ACTIVITY_SET_VNF_MODEL,
WORKFLOW_VDU_INSTANTIATE,
WORKFLOW_VNF_PREPARE,
)
+
from osm_lcm.temporal.vnf_workflows import VnfInstantiateWorkflow, VnfPrepareWorkflow
+# The variables used in the tests
+model_name = "my-model-name"
wf_input = VnfInstantiateInput(
vnfr_uuid="86b53d92-4f5a-402e-8ac2-585ec6b64698",
- model_name="a-model-name",
+ model_name=model_name,
)
cloud = "microk8s"
juju_task_queue = "juju_task_queue"
vnfr_uuid = "9f472177-95c0-4335-b357-5cdc17a79965"
+vim_account_id = "9b0bedc3-ea8e-42fd-acc9-dd03f4dee73c"
sample_vnfr = {
"_id": vnfr_uuid,
"id": vnfr_uuid,
"nsr-id-ref": "dcf4c922-5a73-41bf-a6ca-870c22d6336c",
"vnfd-ref": "jar_vnfd_scalable",
"vnfd-id": "f1b38eac-190c-485d-9a74-c6e169c929d8",
- "vim-account-id": "9b0bedc3-ea8e-42fd-acc9-dd03f4dee73c",
+ "vim-account-id": vim_account_id,
+ "namespace": model_name,
}
vdu = {
"id": "hackfest_basic-VM",
"virtual-storage-desc": ["hackfest_basic-VM-storage"],
}
vnfd_id = "97784f19-d254-4252-946c-cf92d85443ca"
+flavor_1 = {
+ "id": "compute-id-1",
+ "virtual-memory": {"size": "4"},
+ "virtual-cpu": {"cpu-architecture": "x86", "num-virtual-cpu": 2},
+}
+flavor_2 = {
+ "id": "compute-id-2",
+ "virtual-cpu": {"cpu-architecture": "x86", "num-virtual-cpu": 2},
+}
+flavor_3 = {
+ "id": "compute-id-2",
+ "virtual-memory": {"size": "4"},
+}
sample_vnfd = {
"_id": vnfd_id,
"id": "sol006-vnf",
"product-name": "test-vnf",
"software-version": "1.0",
"vdu": [vdu],
+ "virtual-compute-desc": [flavor_1, flavor_2],
}
+sample_charm_info = CharmInfo(app_name="my-app", channel="latest", entity_url="my-url")
+sample_constraints = VduComputeConstraints(cores=1, mem=1)
+sample_vdu_instantiate_input = VduInstantiateInput(
+ vim_uuid=vim_account_id,
+ model_name=model_name,
+ charm_info=sample_charm_info,
+ constraints=sample_constraints,
+ cloud=cloud,
+)
SANDBOXED = False
DEBUG_MODE = True
raise TestException(f"{ACTIVITY_GET_VNF_DETAILS} failed.")
+@activity.defn(name=ACTIVITY_GET_VIM_CLOUD)
+async def mock_get_vim_cloud(
+ get_vim_cloud_input: GetVimCloudInput,
+) -> GetVimCloudOutput:
+ return GetVimCloudOutput(cloud=cloud)
+
+
+@activity.defn(name=ACTIVITY_GET_VIM_CLOUD)
+async def mock_get_vim_cloud_raise_exception(
+ get_vim_cloud_input: GetVimCloudInput,
+) -> GetVimCloudOutput:
+ raise TestException(f"{ACTIVITY_GET_VIM_CLOUD} failed.")
+
+
# Mock Workflowa
@workflow.defn(name=WORKFLOW_VNF_PREPARE, sandboxed=SANDBOXED)
class MockPrepareVnfWorkflow:
self.mock_send_notification_for_vnf,
mock_set_vnf_model,
mock_get_vnf_details,
+ mock_get_vim_cloud,
]
async with self.env, self.get_worker(
self.task_queue, self.workflows, activities
self.check_state_change_call_counts()
self.workflow_is_succeeded()
mock_instantiate_vdus.assert_called_once_with(
- sample_vnfr, sample_vnfd, LCM_TASK_QUEUE
+ vnfr=sample_vnfr, vnfd=sample_vnfd, task_queue=LCM_TASK_QUEUE, cloud=cloud
)
@asynctest.mock.patch(
self.mock_send_notification_for_vnf,
mock_set_vnf_model,
mock_get_vnf_details,
+ mock_get_vim_cloud,
]
async with self.env, self.get_worker(
self.task_queue, self.workflows, activities
self.mock_send_notification_for_vnf,
mock_set_vnf_model,
mock_get_vnf_details,
+ mock_get_vim_cloud,
]
async with self.env, self.get_worker(
self.task_queue, self.workflows, activities
self.mock_send_notification_for_vnf,
mock_set_vnf_model,
mock_get_vnf_details_empty_output,
+ mock_get_vim_cloud,
]
async with self.env, self.get_worker(
self.task_queue, self.workflows, activities
await self.execute_workflow()
self.check_state_change_call_counts()
self.workflow_is_succeeded()
- mock_instantiate_vdus.assert_called_once_with({}, {}, LCM_TASK_QUEUE)
+ mock_instantiate_vdus.assert_called_once_with(
+ vnfr={}, vnfd={}, task_queue=LCM_TASK_QUEUE, cloud=cloud
+ )
@asynctest.mock.patch(
"osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.instantiate_vdus"
self.mock_send_notification_for_vnf,
mock_set_vnf_model,
mock_get_vnf_details,
+ mock_get_vim_cloud,
]
async with self.env, self.get_worker(
self.task_queue, self.workflows, activities
self.mock_send_notification_for_vnf,
mock_set_vnf_model,
mock_get_vnf_details_raise_exception,
+ mock_get_vim_cloud,
]
async with self.env, self.get_worker(self.task_queue, workflows, activities):
with self.assertRaises(WorkflowFailureError) as err:
"get_vnf_details failed.",
)
+ @asynctest.mock.patch(
+ "osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.instantiate_vdus"
+ )
+ async def test_vnf_instantiate_workflow_get_vim_cloud_raises_exception(
+ self, mock_instantiate_vdus
+ ):
+ workflows = [VnfInstantiateWorkflow, MockPrepareVnfWorkflow]
+ activities = [
+ mock_get_task_queue,
+ self.mock_change_vnf_instantiation_state,
+ self.mock_change_vnf_state,
+ self.mock_send_notification_for_vnf,
+ mock_set_vnf_model,
+ mock_get_vnf_details,
+ mock_get_vim_cloud_raise_exception,
+ ]
+ async with self.env, self.get_worker(self.task_queue, workflows, activities):
+ with self.assertRaises(WorkflowFailureError) as err:
+ await self.execute_workflow()
+ self.check_state_change_call_counts()
+ self.workflow_is_failed(err)
+ mock_instantiate_vdus.assert_not_called()
+ self.assertEqual(
+ str(err.exception.cause.cause.message),
+ "get-vim-cloud failed.",
+ )
+
@asynctest.mock.patch(
"osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.get_vdu_instantiate_input"
)
self, mock_vdu_instantiate_input
):
mock_vdu_instantiate_input.return_value = (
- VduInstantiateInput(
- vim_uuid="123",
- model_name="a-model-name",
- charm_info=CharmInfo(
- app_name="my-app", channel="latest", entity_url="my-url"
- ),
- constraints=VduComputeConstraints(cores=1, mem=1),
- cloud=cloud,
- ),
- "vdu_instantiate_workflow_id",
+ sample_vdu_instantiate_input,
+ "vdu-instantiate-workflow-id",
)
activities = [
mock_get_task_queue,
self.mock_send_notification_for_vnf,
mock_set_vnf_model,
mock_get_vnf_details,
+ mock_get_vim_cloud,
]
async with self.env, self.get_worker(
self.task_queue, self.workflows, activities
await self.execute_workflow()
self.check_state_change_call_counts()
self.workflow_is_succeeded()
- call_mock_vdu_instantiate_input = mock_vdu_instantiate_input.call_args
- self.assertEqual(call_mock_vdu_instantiate_input.args[0], sample_vnfr)
- self.assertEqual(call_mock_vdu_instantiate_input.args[1], sample_vnfd)
- self.assertEqual(call_mock_vdu_instantiate_input.args[2], vdu)
+ call_mock_vdu_instantiate_input = mock_vdu_instantiate_input.call_args_list
+ self.assertEqual(call_mock_vdu_instantiate_input[0][1]["vnfr"], sample_vnfr)
+ self.assertEqual(call_mock_vdu_instantiate_input[0][1]["vnfd"], sample_vnfd)
+ self.assertEqual(call_mock_vdu_instantiate_input[0][1]["vdu"], vdu)
+
+ @asynctest.mock.patch(
+ "osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.get_vdu_instantiate_input"
+ )
+ @asynctest.mock.patch("temporalio.workflow.execute_child_workflow")
+ async def test_instantiate_vdus_single_vdu(
+ self, mock_execute_child_workflow: AsyncMock, mock_get_vdu_instantiate_input
+ ):
+ mock_get_vdu_instantiate_input.return_value = (
+ sample_vdu_instantiate_input,
+ "vdu-instantiate-workflow-id",
+ )
+ await VnfInstantiateWorkflow.instantiate_vdus(
+ sample_vnfr, sample_vnfd, LCM_TASK_QUEUE, cloud
+ )
+ self.assertEqual(mock_execute_child_workflow.call_count, 1)
+ mock_execute_child_workflow.assert_called_once_with(
+ workflow=WORKFLOW_VDU_INSTANTIATE,
+ arg=sample_vdu_instantiate_input,
+ task_queue=LCM_TASK_QUEUE,
+ id="vdu-instantiate-workflow-id",
+ )
+ mock_get_vdu_instantiate_input.assert_called_once_with(
+ vnfr=sample_vnfr, vnfd=sample_vnfd, vdu=vdu, cloud=cloud
+ )
+
+ @asynctest.mock.patch(
+ "osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.get_vdu_instantiate_input"
+ )
+ @asynctest.mock.patch("temporalio.workflow.execute_child_workflow")
+ async def test_instantiate_vdus_multiple_vdus(
+ self, mock_execute_child_workflow: AsyncMock, mock_get_vdu_instantiate_input
+ ):
+ sample_vnfd_multi_vdu = deepcopy(sample_vnfd)
+ sample_vnfd_multi_vdu["vdu"] = [vdu, vdu]
+ mock_get_vdu_instantiate_input.side_effect = [
+ (sample_vdu_instantiate_input, "vdu-instantiate-workflow-id-1"),
+ (sample_vdu_instantiate_input, "vdu-instantiate-workflow-id-2"),
+ ]
+ await VnfInstantiateWorkflow.instantiate_vdus(
+ sample_vnfr, sample_vnfd_multi_vdu, LCM_TASK_QUEUE, cloud
+ )
+ self.assertEqual(mock_execute_child_workflow.call_count, 2)
+ call_mock_execute_child_workflow = mock_execute_child_workflow.call_args_list
+ self.assertEqual(
+ call_mock_execute_child_workflow[0][1],
+ {
+ "arg": sample_vdu_instantiate_input,
+ "id": "vdu-instantiate-workflow-id-1",
+ "task_queue": "lcm-task-queue",
+ "workflow": "vdu-instantiate",
+ },
+ )
+ self.assertEqual(
+ call_mock_execute_child_workflow[1][1],
+ {
+ "arg": sample_vdu_instantiate_input,
+ "id": "vdu-instantiate-workflow-id-2",
+ "task_queue": "lcm-task-queue",
+ "workflow": "vdu-instantiate",
+ },
+ )
+ call_mock_get_vdu_instantiate_input = (
+ mock_get_vdu_instantiate_input.call_args_list
+ )
+ self.assertEqual(
+ call_mock_get_vdu_instantiate_input[0][1],
+ {
+ "vnfr": sample_vnfr,
+ "vnfd": sample_vnfd_multi_vdu,
+ "vdu": vdu,
+ "cloud": cloud,
+ },
+ )
+ self.assertEqual(
+ call_mock_get_vdu_instantiate_input[1][1],
+ {
+ "vnfr": sample_vnfr,
+ "vnfd": sample_vnfd_multi_vdu,
+ "vdu": vdu,
+ "cloud": cloud,
+ },
+ )
@asynctest.mock.patch(
"osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.instantiate_vdus"
self.mock_send_notification_for_vnf,
mock_set_vnf_model,
mock_get_vnf_details,
+ mock_get_vim_cloud,
]
async with self.env, self.get_worker(
self.task_queue, workflows, activities
self.check_state_change_call_counts()
self.workflow_is_succeeded()
mock_instantiate_vdus.assert_called_once_with(
- sample_vnfr, sample_vnfd, juju_task_queue
+ vnfr=sample_vnfr, vnfd=sample_vnfd, task_queue=juju_task_queue, cloud=cloud
+ )
+
+
+class TestGetVduInstantiateInputMethods(TestCase):
+ def test_get_flavor_details_successful(self):
+ compute_desc_id = "compute-id-1"
+ result = VnfInstantiateWorkflow.get_flavor_details(compute_desc_id, sample_vnfd)
+ self.assertEqual(result, flavor_1)
+
+ def test_get_flavor_details_empty_compute_desc(self):
+ compute_desc_id = ""
+ result = VnfInstantiateWorkflow.get_flavor_details(compute_desc_id, sample_vnfd)
+ self.assertEqual(result, {})
+
+ def test_get_flavor_details_compute_desc_not_found(self):
+ compute_desc_id = "compute-id-5"
+ result = VnfInstantiateWorkflow.get_flavor_details(compute_desc_id, sample_vnfd)
+ self.assertEqual(result, {})
+
+ def test_get_flavor_details_empty_vnfd(self):
+ compute_desc_id = "compute-id-5"
+ result = VnfInstantiateWorkflow.get_flavor_details(compute_desc_id, {})
+ self.assertEqual(result, {})
+
+ def test_get_flavor_details_wrong_vnfd_format(self):
+ compute_desc_id = "compute-id-2"
+ sample_vnfd = {
+ "_id": vnfd_id,
+ "vdu": [vdu],
+ "virtual-compute-desc": [
+ {
+ "virtual-memory": {"size": "4"},
+ }
+ ],
+ }
+ result = VnfInstantiateWorkflow.get_flavor_details(compute_desc_id, sample_vnfd)
+ self.assertEqual(result, {})
+
+ @patch("osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.get_flavor_details")
+ def test_get_compute_constraints(self, mock_get_flavor_details):
+ mock_get_flavor_details.return_value = flavor_1
+ result = VnfInstantiateWorkflow.get_compute_constraints(vdu, sample_vnfd)
+ self.assertEqual(VduComputeConstraints(cores=2, mem=4), result)
+
+ @patch("osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.get_flavor_details")
+ def test_get_compute_constraints_empty_flavor_details(
+ self, mock_get_flavor_details
+ ):
+ mock_get_flavor_details.return_value = {}
+ result = VnfInstantiateWorkflow.get_compute_constraints(vdu, sample_vnfd)
+ self.assertEqual(VduComputeConstraints(cores=0, mem=0), result)
+
+ @patch("osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.get_flavor_details")
+ def test_get_compute_constraints_flavor_details_is_none(
+ self, mock_get_flavor_details
+ ):
+ mock_get_flavor_details.return_value = None
+ result = VnfInstantiateWorkflow.get_compute_constraints(vdu, sample_vnfd)
+ self.assertEqual(VduComputeConstraints(cores=0, mem=0), result)
+
+ @patch("osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.get_flavor_details")
+ def test_get_compute_constraints_flavor_has_only_cpu(self, mock_get_flavor_details):
+ mock_get_flavor_details.return_value = flavor_2
+ result = VnfInstantiateWorkflow.get_compute_constraints(vdu, sample_vnfd)
+ self.assertEqual(VduComputeConstraints(cores=2, mem=0), result)
+
+ @patch("osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.get_flavor_details")
+ def test_get_compute_constraints_flavor_has_only_memory(
+ self, mock_get_flavor_details
+ ):
+ mock_get_flavor_details.return_value = flavor_3
+ result = VnfInstantiateWorkflow.get_compute_constraints(vdu, sample_vnfd)
+ self.assertEqual(VduComputeConstraints(cores=0, mem=4), result)
+
+ @patch("osm_lcm.temporal.juju_paas_activities.CharmInfoUtils.get_charm_info")
+ @patch(
+ "osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.get_compute_constraints"
+ )
+ def test_get_vdu_instantiate_input(
+ self, mock_get_compute_constraints, mock_get_charm_info
+ ):
+ mock_get_compute_constraints.return_value = sample_constraints
+ mock_get_charm_info.return_value = sample_charm_info
+ (
+ vdu_instantiate_input,
+ vdu_instantiate_wf_id,
+ ) = VnfInstantiateWorkflow.get_vdu_instantiate_input(
+ sample_vnfr, sample_vnfd, vdu, cloud
+ )
+ self.assertEqual(vdu_instantiate_input, sample_vdu_instantiate_input)
+ self.assertEqual(vdu_instantiate_wf_id, "my-model-name-my-app")
+ mock_get_charm_info.assert_called_once()
+ mock_get_compute_constraints.assert_called_once()
+
+ @patch("osm_lcm.temporal.juju_paas_activities.CharmInfoUtils.get_charm_info")
+ @patch(
+ "osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.get_compute_constraints"
+ )
+ def test_get_vdu_instantiate_input_vnfr_without_namespace(
+ self, mock_get_compute_constraints, mock_get_charm_info
+ ):
+ vnfr = deepcopy(sample_vnfr)
+ vnfr.pop("namespace")
+ mock_get_compute_constraints.return_value = sample_constraints
+ mock_get_charm_info.return_value = sample_charm_info
+ with self.assertRaises(TypeError) as err:
+ VnfInstantiateWorkflow.get_vdu_instantiate_input(
+ vnfr, sample_vnfd, vdu, cloud
+ )
+ self.assertEqual(
+ str(err.exception),
+ "unsupported operand type(s) for +: 'NoneType' and 'str'",
+ )
+ mock_get_charm_info.assert_called_once()
+ mock_get_compute_constraints.assert_called_once()
+
+ @patch("osm_lcm.temporal.juju_paas_activities.CharmInfoUtils.get_charm_info")
+ @patch(
+ "osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.get_compute_constraints"
+ )
+ def test_get_vdu_instantiate_input_app_name_is_empty(
+ self, mock_get_compute_constraints, mock_get_charm_info
+ ):
+ mock_get_compute_constraints.return_value = sample_constraints
+ charm_info = CharmInfo(app_name="", channel="latest", entity_url="my-url")
+ mock_get_charm_info.return_value = charm_info
+ (
+ vdu_instantiate_input,
+ vdu_instantiate_wf_id,
+ ) = VnfInstantiateWorkflow.get_vdu_instantiate_input(
+ sample_vnfr, sample_vnfd, vdu, cloud
+ )
+ self.assertEqual(
+ vdu_instantiate_input,
+ VduInstantiateInput(
+ vim_uuid=vim_account_id,
+ model_name=model_name,
+ charm_info=charm_info,
+ constraints=sample_constraints,
+ cloud=cloud,
+ ),
+ )
+ self.assertEqual(vdu_instantiate_wf_id, "my-model-name-")
+ mock_get_charm_info.assert_called_once()
+ mock_get_compute_constraints.assert_called_once()
+
+ @patch("osm_lcm.temporal.juju_paas_activities.CharmInfoUtils.get_charm_info")
+ @patch(
+ "osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.get_compute_constraints"
+ )
+ def test_get_vdu_instantiate_input_get_compute_constraints_raises(
+ self, mock_get_compute_constraints, mock_get_charm_info
+ ):
+ mock_get_compute_constraints.side_effect = TestException(
+ "get_compute_constraints failed"
+ )
+ mock_get_charm_info.return_value = sample_charm_info
+ with self.assertRaises(TestException) as err:
+ result = VnfInstantiateWorkflow.get_vdu_instantiate_input(
+ sample_vnfr, sample_vnfd, vdu, cloud
+ )
+ self.assertEqual(result, None)
+ self.assertEqual(str(err.exception), "get_compute_constraints failed")
+ mock_get_charm_info.assert_called_once()
+ mock_get_compute_constraints.assert_called_once()
+
+ @patch("osm_lcm.temporal.juju_paas_activities.CharmInfoUtils.get_charm_info")
+ @patch(
+ "osm_lcm.temporal.vnf_workflows.VnfInstantiateWorkflow.get_compute_constraints"
+ )
+ def test_get_vdu_instantiate_input_get_charm_info_raises(
+ self, mock_get_compute_constraints, mock_get_charm_info
+ ):
+ mock_get_compute_constraints.return_value = sample_constraints
+ mock_get_charm_info.side_effect = TestException(
+ "get_compute_constraints failed"
)
+ with self.assertRaises(TestException) as err:
+ result = VnfInstantiateWorkflow.get_vdu_instantiate_input(
+ sample_vnfr, sample_vnfd, vdu, cloud
+ )
+ self.assertEqual(result, None)
+ self.assertEqual(str(err.exception), "get_compute_constraints failed")
+ mock_get_charm_info.assert_called_once()
+ mock_get_compute_constraints.assert_not_called()
if __name__ == "__main__":
projects / osm / LCM.git / commitdiff