Fix Workflow ID for NsInstantiate Workflow

Add UTs

The workflow ID is the nsInstanceId

Change-Id: I119ea4e9b2278362f8e737f40517abd7b62f3034
Signed-off-by: Patricia Reinoso <patricia.reinoso@canonical.com>

diff --git a/osm_nbi/temporal/nbi_temporal.py b/osm_nbi/temporal/nbi_temporal.py
index c29a66d..a9051c1 100644 (file)
--- a/osm_nbi/temporal/nbi_temporal.py
+++ b/osm_nbi/temporal/nbi_temporal.py
@@ -1,4 +1,5 @@
-# -*- coding: utf-8 -*-
+########################################################################
+# Copyright ETSI Contributors and Others.
 
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -40,20 +41,16 @@ class NbiTemporal:
         vim_uuid = content["_id"]
         # Derive the operation id (although for a create it will always be 0)
         op_id = content["op_id"].split(":")[1]
-
         workflow = NbiTemporal.workflow_mappings[action]
-
-        if workflow is not None:
-            workflow_data = VimOperationInput(vim_uuid, op_id)
-
-            asyncio.run(
-                WFTemporal(logger_name="nbi.vim_workflow").start_workflow(
-                    task_queue=LCM_TASK_QUEUE,
-                    workflow_name=workflow,
-                    workflow_data=workflow_data,
-                    id=vim_uuid,
-                )
+        workflow_data = VimOperationInput(vim_uuid, op_id)
+        asyncio.run(
+            WFTemporal(logger_name="nbi.vim_workflow").start_workflow(
+                task_queue=LCM_TASK_QUEUE,
+                workflow_name=workflow,
+                workflow_data=workflow_data,
+                id=vim_uuid,
             )
+        )
 
     lcm_workflow_mappings = {
         "instantiate": WORKFLOW_NS_INSTANTIATE,
@@ -86,6 +83,6 @@ class NbiTemporal:
                     nslcmop["lcmOperationType"]
                 ],
                 workflow_data=NsLcmOperationInput(nslcmop=nslcmop),
-                id=nslcmop["_id"],
+                id=nslcmop["nsInstanceId"],
             )
         )

diff --git a/osm_nbi/tests/test_nbi_temporal.py b/osm_nbi/tests/test_nbi_temporal.py
new file mode 100644 (file)
index 0000000..8fe6775
--- /dev/null
+++ b/osm_nbi/tests/test_nbi_temporal.py
@@ -0,0 +1,111 @@
+########################################################################
+# Copyright ETSI Contributors and Others.
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import asynctest
+from osm_common.dataclasses.temporal_dataclasses import (
+    NsLcmOperationInput,
+    VimOperationInput,
+)
+from osm_common.temporal_constants import (
+    LCM_TASK_QUEUE,
+    WORKFLOW_VIM_CREATE,
+    WORKFLOW_NS_INSTANTIATE,
+)
+from osm_nbi.temporal.nbi_temporal import NbiTemporal
+
+
+@asynctest.mock.patch("osm_common.wftemporal.WFTemporal.start_workflow")
+class TestNbiTemporal(asynctest.TestCase):
+    nslcmop = {
+        "id": "a388d577-c3cc-4c7c-9405-8db66c524e07",
+        "_id": "a388d577-c3cc-4c7c-9405-8db66c524e07",
+        "operationState": "PROCESSING",
+        "nsInstanceId": "3d43a41f-794c-4b9f-83e2-e22be11208b1",
+        "lcmOperationType": "instantiate",
+        "startTime": 1683639339.1609678,
+        "isAutomaticInvocation": False,
+        "operationParams": {
+            "nsdId": "17dfb64e-34cd-4385-80c5-bc01d25eacf2",
+            "nsName": "squid",
+            "nsDescription": "default description",
+            "vimAccountId": "29ca8632-06a6-4bf2-a222-893e20a18b55",
+            "nsr_id": "3d43a41f-794c-4b9f-83e2-e22be11208b1",
+            "lcmOperationType": "instantiate",
+            "nsInstanceId": "3d43a41f-794c-4b9f-83e2-e22be11208b1",
+        },
+        "isCancelPending": False,
+    }
+    vim_id = "f9c16d22-ea2c-45d3-b4c7-8aae8c0f7ada"
+    op_id = "5"
+    vim_operation_content = {
+        "name": "juju_paas6",
+        "vim_type": "paas",
+        "description": None,
+        "vim_url": "172.21.249.67:17070",
+        "vim_user": "admin",
+        "vim_password": "mqK0X8DPs6cmTwpW7hrsjQ==",
+        "vim_tenant_name": "null",
+        "config": {
+            "paas_provider": "juju",
+            "cloud": "microk8s",
+            "cloud_credentials": "microk8s",
+            "authorized_keys": "id_rsa.pub",
+            "ca_cert_content": "-----BEGIN CERTIFICATE-----\nMIIETQRzQ=\n-----END CERTIFICATE-----\n",
+        },
+        "_admin": {
+            "created": 1683639881.2804828,
+            "modified": 1683639881.2804828,
+            "current_operation": None,
+        },
+        "_id": vim_id,
+        "schema_version": "1.11",
+        "op_id": "{}:{}".format(vim_id, op_id),
+    }
+
+    def setUp(self) -> None:
+        self.temporal = NbiTemporal()
+
+    def test_start_vim_workflow(self, mock_start_workflow):
+        action = "created"
+        self.temporal.start_vim_workflow(action, self.vim_operation_content)
+        mock_start_workflow.assert_called_once_with(
+            task_queue=LCM_TASK_QUEUE,
+            workflow_name=WORKFLOW_VIM_CREATE,
+            workflow_data=VimOperationInput(self.vim_id, self.op_id),
+            id=self.vim_id,
+        )
+
+    def test_start_vim_workflow_invalid_action(self, mock_start_workflow):
+        action = "some_action"
+        with self.assertRaises(KeyError):
+            self.temporal.start_vim_workflow(action, self.vim_operation_content)
+        mock_start_workflow.assert_not_called()
+
+    def test_start_ns_workflow(self, mock_start_workflow):
+        self.temporal.start_ns_workflow(self.nslcmop)
+        mock_start_workflow.assert_called_once_with(
+            task_queue=LCM_TASK_QUEUE,
+            workflow_name=WORKFLOW_NS_INSTANTIATE,
+            workflow_data=NsLcmOperationInput(nslcmop=self.nslcmop),
+            id=self.nslcmop["nsInstanceId"],
+        )
+
+    def test_start_ns_workflow_invalid_operation(self, mock_start_workflow):
+        self.nslcmop["lcmOperationType"] = "some_operation"
+        with self.assertRaises(KeyError):
+            self.temporal.start_ns_workflow(self.nslcmop)
+        mock_start_workflow.assert_not_called()