OSMENG-1155: Implementation of Constants and Dataclasses
Add implementation for workflows and activities
Change-Id: I58226765c41d18821724ac5763a3fe390c371ca6
Signed-off-by: Dario Faccin <dario.faccin@canonical.com>
Signed-off-by: Mark Beierl <mark.beierl@canonical.com>
diff --git a/osm_lcm/tests/test_vnf_activities.py b/osm_lcm/tests/test_vnf_activities.py
index 0c346ee..1f34563 100644
--- a/osm_lcm/tests/test_vnf_activities.py
+++ b/osm_lcm/tests/test_vnf_activities.py
@@ -19,23 +19,20 @@
from unittest import TestCase
from unittest.mock import Mock, patch
-from osm_common.dataclasses.temporal_dataclasses import (
- ChangeVnfInstantiationStateInput,
- ChangeVnfStateInput,
- GetTaskQueueInput,
- GetVimCloudInput,
- GetVnfDescriptorInput,
- GetVnfRecordInput,
- SetVnfModelInput,
- VnfInstantiationState,
- VnfState,
- VduComputeConstraints,
-)
from osm_common.dbbase import DbException
-from osm_common.temporal_constants import (
- LCM_TASK_QUEUE,
+from osm_common.temporal_task_queues.task_queues_mappings import LCM_TASK_QUEUE
+from osm_lcm.temporal.vnf_activities import (
+ ChangeVnfInstantiationStateImpl,
+ ChangeVnfStateImpl,
+ GetTaskQueueImpl,
+ GetVimCloudImpl,
+ GetVnfDescriptorImpl,
+ GetVnfRecordImpl,
+ SetVnfModelImpl,
+ VnfSpecifications,
)
-from osm_lcm.temporal.vnf_activities import VnfDbActivity, VnfOperations
+from osm_common.temporal.dataclasses_common import VduComputeConstraints
+from osm_common.temporal.states import VnfState, VnfInstantiationState
from temporalio.testing import ActivityEnvironment
@@ -43,7 +40,7 @@
vnfd_uuid = "97784f19-d254-4252-946c-cf92d85443ca"
vim_uuid = "a64f7c6c-bc27-4ec8-b664-5500a3324eca"
model_name = "my-model-name"
-set_vnf_model_input = SetVnfModelInput(vnfr_uuid=vnfr_uuid, model_name=model_name)
+set_vnf_model_input = SetVnfModelImpl.Input(vnfr_uuid=vnfr_uuid, model_name=model_name)
cloud = "microk8s"
nsr_uuid = "583726d4-957d-47f5-8df5-199456f7afd0"
sample_vim_record = {
@@ -133,35 +130,33 @@
def setUp(self):
self.db = Mock()
self.env = ActivityEnvironment()
- self.vnf_db_activity = VnfDbActivity(self.db)
+ self.change_vnf_state = ChangeVnfStateImpl(self.db)
+ self.change_vnf_instantiation_state = ChangeVnfInstantiationStateImpl(self.db)
+ self.set_vnf_model = SetVnfModelImpl(self.db)
async def test_change_vnf_state__successful__db_updated_as_expected(self):
vnf_state = VnfState.STOPPED
- change_vnf_state_input = ChangeVnfStateInput(vnfr_uuid, vnf_state)
- await self.env.run(
- self.vnf_db_activity.change_vnf_state, change_vnf_state_input
- )
+ change_vnf_state_input = ChangeVnfStateImpl.Input(vnfr_uuid, vnf_state)
+ await self.env.run(self.change_vnf_state, change_vnf_state_input)
self.db.set_one.assert_called_with(
"vnfrs", {"_id": vnfr_uuid}, {"vnfState": vnf_state.name}
)
async def test_change_vnf_state__failed__raises_db_exception(self):
- change_vnf_state_input = ChangeVnfStateInput(vnfr_uuid, VnfState.STARTED)
+ change_vnf_state_input = ChangeVnfStateImpl.Input(vnfr_uuid, VnfState.STARTED)
self.db.set_one.side_effect = DbException("not found")
with self.assertRaises(DbException):
- await self.env.run(
- self.vnf_db_activity.change_vnf_state, change_vnf_state_input
- )
+ await self.env.run(self.change_vnf_state, change_vnf_state_input)
async def test_change_vnf_instantiation_state__successful__db_updated_as_expected(
self,
):
instantiation_state = VnfInstantiationState.NOT_INSTANTIATED
- change_instantiation_input = ChangeVnfInstantiationStateInput(
+ change_instantiation_input = ChangeVnfInstantiationStateImpl.Input(
vnfr_uuid, instantiation_state
)
await self.env.run(
- self.vnf_db_activity.change_vnf_instantiation_state,
+ self.change_vnf_instantiation_state,
change_instantiation_input,
)
self.db.set_one.assert_called_with(
@@ -171,18 +166,18 @@
)
async def test_change_vnf_instantiation_state__failed__raises_db_exception(self):
- change_instantiation_input = ChangeVnfInstantiationStateInput(
+ change_instantiation_input = ChangeVnfInstantiationStateImpl.Input(
vnfr_uuid, VnfInstantiationState.INSTANTIATED
)
self.db.set_one.side_effect = DbException("not found")
with self.assertRaises(DbException):
await self.env.run(
- self.vnf_db_activity.change_vnf_instantiation_state,
+ self.change_vnf_instantiation_state,
change_instantiation_input,
)
async def test_set_vnf_model__successful__db_updated_as_expected(self):
- await self.env.run(self.vnf_db_activity.set_vnf_model, set_vnf_model_input)
+ await self.env.run(self.set_vnf_model, set_vnf_model_input)
self.db.set_one.assert_called_with(
"vnfrs", {"_id": vnfr_uuid}, {"namespace": model_name}
)
@@ -190,20 +185,20 @@
async def test_set_vnf_model__failed__raises_db_exception(self):
self.db.set_one.side_effect = DbException("not found")
with self.assertRaises(DbException):
- await self.env.run(self.vnf_db_activity.set_vnf_model, set_vnf_model_input)
+ await self.env.run(self.set_vnf_model, set_vnf_model_input)
class TestGetTaskQueue(asynctest.TestCase):
async def setUp(self):
self.db = Mock()
- self.vnf_operations_instance = VnfOperations(self.db)
+ self.get_task_queue = GetTaskQueueImpl(self.db)
self.env = ActivityEnvironment()
- self.get_task_queue_input = GetTaskQueueInput(vnfr_uuid=sample_vnfr["id"])
+ self.get_task_queue_input = GetTaskQueueImpl.Input(vnfr_uuid=sample_vnfr["id"])
async def test_activity__succeeded__get_expected_task_queue(self):
self.db.get_one.side_effect = [sample_vnfr, sample_vim_record]
activity_result = await self.env.run(
- self.vnf_operations_instance.get_task_queue,
+ self.get_task_queue,
self.get_task_queue_input,
)
self.assertEqual(activity_result.task_queue, LCM_TASK_QUEUE)
@@ -212,7 +207,7 @@
self.db.get_one.side_effect = DbException("not found")
with self.assertRaises(DbException):
await self.env.run(
- self.vnf_operations_instance.get_task_queue,
+ self.get_task_queue,
self.get_task_queue_input,
)
@@ -222,7 +217,7 @@
self.db.get_one.side_effect = [sample_vnfr, vim_record]
with self.assertRaises(KeyError):
await self.env.run(
- self.vnf_operations_instance.get_task_queue,
+ self.get_task_queue,
self.get_task_queue_input,
)
@@ -230,14 +225,14 @@
class TestGetVnfDescriptor(asynctest.TestCase):
async def setUp(self):
self.db = Mock()
- self.vnf_operations_instance = VnfOperations(self.db)
+ self.get_vnf_descriptor = GetVnfDescriptorImpl(self.db)
self.env = ActivityEnvironment()
async def test_activity__succeeded__get_expected_vnfd(self):
self.db.get_one.return_value = sample_vnfd
activity_result = await self.env.run(
- self.vnf_operations_instance.get_vnf_descriptor,
- GetVnfDescriptorInput(vnfd_uuid=vnfd_uuid),
+ self.get_vnf_descriptor,
+ GetVnfDescriptorImpl.Input(vnfd_uuid=vnfd_uuid),
)
self.assertEqual(activity_result.vnfd, sample_vnfd)
@@ -245,22 +240,22 @@
self.db.get_one.side_effect = DbException("Can not connect to Database.")
with self.assertRaises(DbException):
await self.env.run(
- self.vnf_operations_instance.get_vnf_descriptor,
- GetVnfDescriptorInput(vnfd_uuid=vnfd_uuid),
+ self.get_vnf_descriptor,
+ GetVnfDescriptorImpl.Input(vnfd_uuid=vnfd_uuid),
)
class TestGetVnfRecord(asynctest.TestCase):
async def setUp(self):
self.db = Mock()
- self.vnf_operations_instance = VnfOperations(self.db)
+ self.get_vnf_record = GetVnfRecordImpl(self.db)
self.env = ActivityEnvironment()
async def test_activity__succeeded__get_expected_vnfr(self):
self.db.get_one.return_value = sample_vnfr
activity_result = await self.env.run(
- self.vnf_operations_instance.get_vnf_record,
- GetVnfRecordInput(vnfr_uuid=vnfr_uuid),
+ self.get_vnf_record,
+ GetVnfRecordImpl.Input(vnfr_uuid=vnfr_uuid),
)
self.assertEqual(activity_result.vnfr, sample_vnfr)
@@ -268,22 +263,22 @@
self.db.get_one.side_effect = DbException("Can not connect to Database.")
with self.assertRaises(DbException):
await self.env.run(
- self.vnf_operations_instance.get_vnf_record,
- GetVnfRecordInput(vnfr_uuid=vnfr_uuid),
+ self.get_vnf_record,
+ GetVnfRecordImpl.Input(vnfr_uuid=vnfr_uuid),
)
class TestGetVimCloud(asynctest.TestCase):
async def setUp(self):
self.db = Mock()
- self.vnf_operations_instance = VnfOperations(self.db)
+ self.get_vim_cloud = GetVimCloudImpl(self.db)
self.env = ActivityEnvironment()
async def test_activity__succeeded__get_vim_cloud(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.get_vim_cloud,
+ GetVimCloudImpl.Input(vnfr_uuid=sample_vnfr["id"]),
)
self.assertEqual(activity_result.cloud, cloud)
@@ -292,8 +287,8 @@
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.get_vim_cloud,
+ GetVimCloudImpl.Input(vnfr_uuid=sample_vnfr["id"]),
)
self.assertEqual(activity_result.cloud, "")
@@ -301,8 +296,8 @@
self.db.get_one.side_effect = DbException("Can not connect to Database.")
with self.assertRaises(DbException):
await self.env.run(
- self.vnf_operations_instance.get_vim_cloud,
- GetVimCloudInput(vnfr_uuid=sample_vnfr["id"]),
+ self.get_vim_cloud,
+ GetVimCloudImpl.Input(vnfr_uuid=sample_vnfr["id"]),
)
async def test_activity__wrong_vim_record__raise_key_error(self):
@@ -311,30 +306,30 @@
self.db.get_one.side_effect = [sample_vnfr, vim_record]
with self.assertRaises(KeyError):
await self.env.run(
- self.vnf_operations_instance.get_vim_cloud,
- GetVimCloudInput(vnfr_uuid=sample_vnfr["id"]),
+ self.get_vim_cloud,
+ GetVimCloudImpl.Input(vnfr_uuid=sample_vnfr["id"]),
)
class TestGetVduInstantiateInfoMethods(TestCase):
def test_get_flavor_details__successful__get_flavor(self):
compute_desc_id = "compute-id-1"
- result = VnfOperations._get_flavor_details(compute_desc_id, sample_vnfd)
+ result = VnfSpecifications._get_flavor_details(compute_desc_id, sample_vnfd)
self.assertEqual(result, flavor_1)
def test_get_flavor_details__empty_compute_desc__flavor_is_none(self):
compute_desc_id = ""
- result = VnfOperations._get_flavor_details(compute_desc_id, sample_vnfd)
+ result = VnfSpecifications._get_flavor_details(compute_desc_id, sample_vnfd)
self.assertEqual(result, None)
def test_get_flavor_details__compute_desc_not_found__flavor_is_none(self):
compute_desc_id = "compute-id-5"
- result = VnfOperations._get_flavor_details(compute_desc_id, sample_vnfd)
+ result = VnfSpecifications._get_flavor_details(compute_desc_id, sample_vnfd)
self.assertEqual(result, None)
def test_get_flavor_details__empty_vnfd__flavor_is_none(self):
compute_desc_id = "compute-id-5"
- result = VnfOperations._get_flavor_details(compute_desc_id, {})
+ result = VnfSpecifications._get_flavor_details(compute_desc_id, {})
self.assertEqual(result, None)
def test_get_flavor_details__wrong_vnfd_format__flavor_is_none(self):
@@ -348,47 +343,47 @@
}
],
}
- result = VnfOperations._get_flavor_details(compute_desc_id, sample_vnfd)
+ result = VnfSpecifications._get_flavor_details(compute_desc_id, sample_vnfd)
self.assertEqual(result, None)
- @patch("osm_lcm.temporal.vnf_activities.VnfOperations._get_flavor_details")
+ @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_flavor_details")
def test_get_compute_constraints__succeeded__get_constraints(
self, mock_get_flavor_details
):
mock_get_flavor_details.return_value = flavor_1
- result = VnfOperations.get_compute_constraints(vdu, sample_vnfd)
+ result = VnfSpecifications.get_compute_constraints(vdu, sample_vnfd)
self.assertEqual(VduComputeConstraints(cores=2, mem=4), result)
- @patch("osm_lcm.temporal.vnf_activities.VnfOperations._get_flavor_details")
+ @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_flavor_details")
def test_get_compute_constraints__empty_flavor_details__no_constraints(
self, mock_get_flavor_details
):
mock_get_flavor_details.return_value = {}
- result = VnfOperations.get_compute_constraints(vdu, sample_vnfd)
+ result = VnfSpecifications.get_compute_constraints(vdu, sample_vnfd)
self.assertEqual(VduComputeConstraints(cores=0, mem=0), result)
- @patch("osm_lcm.temporal.vnf_activities.VnfOperations._get_flavor_details")
+ @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_flavor_details")
def test_get_compute_constraints__flavor_details_is_none__no_constraints(
self, mock_get_flavor_details
):
mock_get_flavor_details.return_value = None
- result = VnfOperations.get_compute_constraints(vdu, sample_vnfd)
+ result = VnfSpecifications.get_compute_constraints(vdu, sample_vnfd)
self.assertEqual(VduComputeConstraints(cores=0, mem=0), result)
- @patch("osm_lcm.temporal.vnf_activities.VnfOperations._get_flavor_details")
+ @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_flavor_details")
def test_get_compute_constraints__flavor_has_only_cpu__get_cpu_constraint(
self, mock_get_flavor_details
):
mock_get_flavor_details.return_value = flavor_2
- result = VnfOperations.get_compute_constraints(vdu, sample_vnfd)
+ result = VnfSpecifications.get_compute_constraints(vdu, sample_vnfd)
self.assertEqual(VduComputeConstraints(cores=2, mem=0), result)
- @patch("osm_lcm.temporal.vnf_activities.VnfOperations._get_flavor_details")
+ @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_flavor_details")
def test_get_compute_constraints__flavor_has_only_memory__get_memory_constraint(
self, mock_get_flavor_details
):
mock_get_flavor_details.return_value = flavor_3
- result = VnfOperations.get_compute_constraints(vdu, sample_vnfd)
+ result = VnfSpecifications.get_compute_constraints(vdu, sample_vnfd)
self.assertEqual(VduComputeConstraints(cores=0, mem=4), result)
def test_list_to_dict__config_has_several_items__get_expected_dict(self):
@@ -396,45 +391,45 @@
{"key": "config::domain_name1", "value": "osm.org"},
{"key": "domain_name2", "value": "osm.com"},
]
- result = VnfOperations._list_to_dict(config)
+ result = VnfSpecifications._list_to_dict(config)
self.assertEqual(
result, {"config::domain_name1": "osm.org", "domain_name2": "osm.com"}
)
def test_list_to_dict__empty_input__get_empty_dict(self):
config = []
- result = VnfOperations._list_to_dict(config)
+ result = VnfSpecifications._list_to_dict(config)
self.assertEqual(result, {})
def test_get_only_config_items__with_identifier__get_expected_config(self):
config = {"config::redirect-map": "https://osm.instantiation.params"}
- result = VnfOperations._get_only_config_items(config)
+ result = VnfSpecifications._get_only_config_items(config)
self.assertEqual(result, {"redirect-map": "https://osm.instantiation.params"})
def test_get_only_config_items__without_identifier__no_config(self):
config = {"key": "domain_name1", "value": "osm.org"}
- result = VnfOperations._get_only_config_items(config)
+ result = VnfSpecifications._get_only_config_items(config)
self.assertEqual(result, {})
def test_get_only_config_items__empty_input__no_config(self):
config = {}
- result = VnfOperations._get_only_config_items(config)
+ result = VnfSpecifications._get_only_config_items(config)
self.assertEqual(result, {})
def test_get_vdu_instantiation_params__empty_vnf_config__get_params(self):
- result = VnfOperations.get_vdu_instantiation_params(vdu_id, vnf_config)
+ result = VnfSpecifications.get_vdu_instantiation_params(vdu_id, vnf_config)
self.assertEqual(
result, {"config::redirect-map": "https://osm.instantiation.params"}
)
def test_get_vdu_instantiation_params__empty_vnf_config__no_params(self):
- result = VnfOperations.get_vdu_instantiation_params(vdu_id, {})
+ result = VnfSpecifications.get_vdu_instantiation_params(vdu_id, {})
self.assertEqual(result, {})
def test_get_vdu_instantiation_params__vdu_id_mismatch__no_params(self):
config = deepcopy(vnf_config)
config["vdu"][0]["id"] = "other-vdu-id"
- result = VnfOperations.get_vdu_instantiation_params(vdu_id, config)
+ result = VnfSpecifications.get_vdu_instantiation_params(vdu_id, config)
self.assertEqual(result, {})
def test_get_vdu_instantiation_params__empty_configurable_properties__no_params(
@@ -442,11 +437,11 @@
):
config = deepcopy(vnf_config)
config["vdu"][0]["configurable-properties"] = {}
- result = VnfOperations.get_vdu_instantiation_params(vdu_id, config)
+ result = VnfSpecifications.get_vdu_instantiation_params(vdu_id, config)
self.assertEqual(result, {})
- @patch("osm_lcm.temporal.vnf_activities.VnfOperations._get_only_config_items")
- @patch("osm_lcm.temporal.vnf_activities.VnfOperations._list_to_dict")
+ @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_only_config_items")
+ @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._list_to_dict")
def test_get_application_config__instantiate_config_overrides_vdu_config__get_expected_application_config(
self,
mock_list_to_dict,
@@ -460,13 +455,13 @@
},
{"domain_name1": "osm.public"},
]
- result = VnfOperations.get_application_config(vdu, vdu_instantiate_config)
+ result = VnfSpecifications.get_application_config(vdu, vdu_instantiate_config)
self.assertEqual(
result, {"domain_name1": "osm.public", "domain_name2": "osm.com"}
)
- @patch("osm_lcm.temporal.vnf_activities.VnfOperations._get_only_config_items")
- @patch("osm_lcm.temporal.vnf_activities.VnfOperations._list_to_dict")
+ @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_only_config_items")
+ @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._list_to_dict")
def test_get_application_config__empty_instantiate_config__get_descriptor_config(
self,
mock_list_to_dict,
@@ -480,11 +475,11 @@
},
{},
]
- result = VnfOperations.get_application_config(vdu, vdu_instantiate_config)
+ result = VnfSpecifications.get_application_config(vdu, vdu_instantiate_config)
self.assertEqual(result, app_config)
- @patch("osm_lcm.temporal.vnf_activities.VnfOperations._get_only_config_items")
- @patch("osm_lcm.temporal.vnf_activities.VnfOperations._list_to_dict")
+ @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_only_config_items")
+ @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._list_to_dict")
def test_get_application_config__no_config__empty_application_config(
self,
mock_list_to_dict,
@@ -494,7 +489,9 @@
vdu_copy = deepcopy(vdu)
vdu_copy["configurable-properties"] = []
mock_get_only_config_items.side_effect = [{}, {}]
- result = VnfOperations.get_application_config(vdu_copy, vdu_instantiate_config)
+ result = VnfSpecifications.get_application_config(
+ vdu_copy, vdu_instantiate_config
+ )
self.assertEqual(result, {})