Add UTs for VNF activities
Change-Id: I2e055bcecb358f1531052815d6a8ba89f8270b07
Signed-off-by: Patricia Reinoso <patricia.reinoso@canonical.com>
diff --git a/osm_lcm/tests/test_vnf_activities.py b/osm_lcm/tests/test_vnf_activities.py
index 592c523..02fad62 100644
--- a/osm_lcm/tests/test_vnf_activities.py
+++ b/osm_lcm/tests/test_vnf_activities.py
@@ -19,11 +19,19 @@
from asyncio.exceptions import CancelledError
from copy import deepcopy
from osm_common.dataclasses.temporal_dataclasses import (
+ ChangeVnfInstantiationStateInput,
+ ChangeVnfStateInput,
+ GetTaskQueueInput,
GetVimCloudInput,
GetVnfDetailsInput,
VnfInstantiateInput,
+ VnfInstantiationState,
+ VnfState,
)
from osm_common.dbbase import DbException
+from osm_common.temporal_constants import (
+ LCM_TASK_QUEUE,
+)
from osm_lcm.temporal.vnf_activities import VnfDbActivity, VnfOperations
from temporalio.testing import ActivityEnvironment
from unittest.mock import Mock
@@ -70,13 +78,56 @@
self.env = ActivityEnvironment()
self.vnf_db_activity = VnfDbActivity(self.db)
- async def test_set_vnf_model(self):
+ async def test_change_vnf_state_nominal_case(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
+ )
+ self.db.set_one.assert_called_with(
+ "vnfrs", {"_id": vnfr_uuid}, {"vnfState": vnf_state.name}
+ )
+
+ async def test_change_vnf_state_db_raises_exception(self):
+ change_vnf_state_input = ChangeVnfStateInput(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
+ )
+
+ async def test_change_vnf_instantiation_state_nominal_case(self):
+ instantiation_state = VnfInstantiationState.NOT_INSTANTIATED
+ change_instantiation_input = ChangeVnfInstantiationStateInput(
+ vnfr_uuid, instantiation_state
+ )
+ await self.env.run(
+ self.vnf_db_activity.change_vnf_instantiation_state,
+ change_instantiation_input,
+ )
+ self.db.set_one.assert_called_with(
+ "vnfrs",
+ {"_id": vnfr_uuid},
+ {"instantiationState": instantiation_state.name},
+ )
+
+ async def test_change_vnf_instantiation_state_db_raises_exception(self):
+ change_instantiation_input = ChangeVnfInstantiationStateInput(
+ 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_state, change_instantiation_input
+ )
+
+ async def test_set_vnf_model_nominal_case(self):
await self.env.run(self.vnf_db_activity.set_vnf_model, vnf_instantiate_input)
self.db.set_one.assert_called_with(
"vnfrs", {"_id": vnfr_uuid}, {"namespace": model_name}
)
- async def test_db_raises_exception(self):
+ async def test_set_vnf_model_db_raises_exception(self):
self.db.set_one.side_effect = DbException("not found")
with self.assertRaises(DbException):
await self.env.run(
@@ -84,6 +135,50 @@
)
+class TestGetTaskQueue(asynctest.TestCase):
+ async def setUp(self):
+ self.db = Mock()
+ self.vnf_operations_instance = VnfOperations(self.db)
+ self.env = ActivityEnvironment()
+ self.get_task_queue_input = GetTaskQueueInput(vnfr_uuid=sample_vnfr["id"])
+
+ 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_task_queue,
+ self.get_task_queue_input,
+ )
+ self.assertEqual(activity_result.task_queue, LCM_TASK_QUEUE)
+
+ async def test_db_raises_exception(self):
+ 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_input,
+ )
+
+ async def test_invalid_task_queue(self):
+ vim_record = deepcopy(sample_vim_record)
+ vim_record["vim_type"] = "some-vim-type"
+ 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_input,
+ )
+
+ 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_task_queue,
+ self.get_task_queue_input,
+ )
+ self.assertEqual(activity_result, None)
+
+
class TestVnfDetails(asynctest.TestCase):
async def setUp(self):
self.db = Mock()