osm_lcm/tests/test_ns_activities.py

#######################################################################################
# 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 ModelInfo, NsInstantiateInput
from osm_common.dbbase import DbException
from osm_lcm.temporal.ns_activities import NsDbActivity
from temporalio.testing import ActivityEnvironment
from unittest.mock import Mock


vim_id = "some-vim-uuid"
ns_id = "0123456789-9876543210"
nsr = {
    "_id": ns_id,
    "datacenter": vim_id,
}
input = NsInstantiateInput(ns_id, "op_id")
expected_namespace = "9-9876543210-ome-vim-uuid"
expected_model_info = ModelInfo(vim_id, expected_namespace)


class TestGetModelInfo(asynctest.TestCase):
    def setUp(self):
        self.db = Mock()
        self.env = ActivityEnvironment()
        self.ns_db_activity = NsDbActivity(self.db)

    async def test_nominal_case(self):
        self.db.get_one.return_value = nsr
        model_info = await self.env.run(self.ns_db_activity.get_model_info, input)
        self.assertEqual(model_info, expected_model_info)

    async def test_db_raises_exception(self):
        self.db.get_one.side_effect = DbException("not found")
        with self.assertRaises(DbException):
            model_info = await self.env.run(self.ns_db_activity.get_model_info, input)
            self.assertIsNone(model_info)

    async def test_no_datacenter_raises_exception(self):
        nsr = {"_id": ns_id}
        self.db.get_one.return_value = nsr
        with self.assertRaises(TypeError):
            model_info = await self.env.run(self.ns_db_activity.get_model_info, input)
            self.assertIsNone(model_info)