OSMENG-1048 Implement day1 configuration for VDU
Day1 config implementation, arranging existing unit tests and imports
adding task and execution timeout policy for workflows in tests.
Change-Id: Ie5a2626eec01176723d8130576facbf4934d5285
Signed-off-by: Gulsum Atici <gulsum.atici@canonical.com>
diff --git a/osm_lcm/tests/test_ns_activities.py b/osm_lcm/tests/test_ns_activities.py
index 623f072..c3176ee 100644
--- a/osm_lcm/tests/test_ns_activities.py
+++ b/osm_lcm/tests/test_ns_activities.py
@@ -14,45 +14,90 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-
-from unittest.mock import Mock
-
import asynctest
-from osm_common.dataclasses.temporal_dataclasses import GetVnfRecordIdsInput
+
+from osm_common.dataclasses.temporal_dataclasses import (
+ GetNsRecordInput,
+ GetVnfDetailsInput,
+)
from osm_common.dbbase import DbException
from temporalio.testing import ActivityEnvironment
-
from osm_lcm.temporal.ns_activities import NsOperations
+from unittest.mock import Mock
ns_uuid = "00000000-0000-0000-0000-000000000000"
-get_vnf_record_ids_input = GetVnfRecordIdsInput(ns_uuid=ns_uuid)
+get_vnf_details_input = GetVnfDetailsInput(ns_uuid=ns_uuid)
+sample_vnf_details = [
+ {
+ "id": "00000000-0000-0000-0000-000000000000",
+ "member-vnf-index-ref": "vnf1",
+ },
+ {
+ "id": "00000000-0000-0000-0000-000000000000",
+ "member-vnf-index-ref": "vnf2",
+ },
+]
+sample_nsr = {
+ "_id": ns_uuid,
+ "name": "sol006_juju24",
+ "name-ref": "sol006_juju24",
+ "short-name": "sol006_juju24",
+ "admin-status": "ENABLED",
+ "nsState": "NOT_INSTANTIATED",
+ "currentOperation": "IDLE",
+}
-class TestGetModelInfo(asynctest.TestCase):
+class TestException(Exception):
+ pass
+
+
+class TestGetVnfDetails(asynctest.TestCase):
def setUp(self):
self.db = Mock()
self.env = ActivityEnvironment()
self.ns_operations_activity = NsOperations(self.db)
- async def test_get_vnfr_ids(self):
- self.db.get_list.return_value = [
- {"id": "00000000-0000-0000-0000-000000000000"},
- {"id": "00000000-0000-0000-0000-000000000000"},
- ]
+ async def test_activity__succeded__get_expected_result(self):
+ self.db.get_list.return_value = sample_vnf_details
result = await self.env.run(
- self.ns_operations_activity.get_vnf_record_ids, get_vnf_record_ids_input
+ self.ns_operations_activity.get_vnf_details, get_vnf_details_input
)
- assert result.vnfr_ids == [
- "00000000-0000-0000-0000-000000000000",
- "00000000-0000-0000-0000-000000000000",
- ]
+ self.assertEqual(
+ result.vnf_details,
+ [
+ ("00000000-0000-0000-0000-000000000000", "vnf1"),
+ ("00000000-0000-0000-0000-000000000000", "vnf2"),
+ ],
+ )
- self.db.get_list.assert_called_with("vnfrs", {"nsr-id-ref": ns_uuid})
-
- async def test_activity_raises_db_exception(self):
- self.db.get_list.side_effect = DbException("not found")
+ async def test_activity__failed__raise_db_exception(self):
+ self.db.get_list.side_effect = DbException("not found.")
with self.assertRaises(DbException):
await self.env.run(
- self.ns_operations_activity.get_vnf_record_ids, get_vnf_record_ids_input
+ self.ns_operations_activity.get_vnf_details, get_vnf_details_input
+ )
+
+
+class TestGetNsRecord(asynctest.TestCase):
+ async def setUp(self):
+ self.db = Mock()
+ self.env = ActivityEnvironment()
+ self.ns_operations_activity = NsOperations(self.db)
+
+ async def test_activity__succeeded__get_expected_result(self):
+ self.db.get_one.return_value = sample_nsr
+ activity_result = await self.env.run(
+ self.ns_operations_activity.get_ns_record,
+ GetNsRecordInput(nsr_uuid=sample_nsr["_id"]),
+ )
+ self.assertEqual(activity_result.nsr, sample_nsr)
+
+ async def test_activity__failed__raise_test_exception(self):
+ self.db.get_one.side_effect = TestException("Can not connect to Database.")
+ with self.assertRaises(TestException):
+ await self.env.run(
+ self.ns_operations_activity.get_ns_record,
+ GetNsRecordInput(nsr_uuid=sample_nsr["_id"]),
)