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_ns_activities.py b/osm_lcm/tests/test_ns_activities.py
index c3176ee..833f440 100644
--- a/osm_lcm/tests/test_ns_activities.py
+++ b/osm_lcm/tests/test_ns_activities.py
@@ -16,17 +16,13 @@
import asynctest
-from osm_common.dataclasses.temporal_dataclasses import (
- GetNsRecordInput,
- GetVnfDetailsInput,
-)
+from osm_lcm.temporal.ns_activities import GetNsRecordImpl, GetVnfDetailsImpl
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_details_input = GetVnfDetailsInput(ns_uuid=ns_uuid)
+get_vnf_details_input = GetVnfDetailsImpl.Input(ns_uuid=ns_uuid)
sample_vnf_details = [
{
"id": "00000000-0000-0000-0000-000000000000",
@@ -56,13 +52,11 @@
def setUp(self):
self.db = Mock()
self.env = ActivityEnvironment()
- self.ns_operations_activity = NsOperations(self.db)
+ self.get_vnf_details_impl = GetVnfDetailsImpl(self.db)
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_details, get_vnf_details_input
- )
+ result = await self.env.run(self.get_vnf_details_impl, get_vnf_details_input)
self.assertEqual(
result.vnf_details,
@@ -75,22 +69,20 @@
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_details, get_vnf_details_input
- )
+ await self.env.run(self.get_vnf_details_impl, 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)
+ self.get_ns_record_impl = GetNsRecordImpl(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.get_ns_record_impl,
+ GetNsRecordImpl.Input(nsr_uuid=sample_nsr["_id"]),
)
self.assertEqual(activity_result.nsr, sample_nsr)
@@ -98,6 +90,6 @@
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"]),
+ self.get_ns_record_impl,
+ GetNsRecordImpl.Input(nsr_uuid=sample_nsr["_id"]),
)