osm_lcm/tests/test_vnf_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 copy import deepcopy
from unittest import TestCase
from unittest.mock import Mock, patch

from osm_common.dbbase import DbException
from osm_common.temporal_task_queues.task_queues_mappings import LCM_TASK_QUEUE
from osm_lcm.temporal.vnf_activities import (
    ChangeVnfInstantiationStateImpl,
    ChangeVnfStateImpl,
    GetTaskQueueImpl,
    GetVimCloudImpl,
    GetVnfDescriptorImpl,
    GetVnfRecordImpl,
    SetVnfModelImpl,
    VnfSpecifications,
)
from osm_common.temporal.dataclasses_common import VduComputeConstraints
from osm_common.temporal.states import VnfState, VnfInstantiationState
from temporalio.testing import ActivityEnvironment


vnfr_uuid = "9f472177-95c0-4335-b357-5cdc17a79965"
vnfd_uuid = "97784f19-d254-4252-946c-cf92d85443ca"
vim_uuid = "a64f7c6c-bc27-4ec8-b664-5500a3324eca"
model_name = "my-model-name"
set_vnf_model_input = SetVnfModelImpl.Input(vnfr_uuid=vnfr_uuid, model_name=model_name)
cloud = "microk8s"
nsr_uuid = "583726d4-957d-47f5-8df5-199456f7afd0"
sample_vim_record = {
    "_id": vim_uuid,
    "name": "juju",
    "vim_type": "paas",
    "vim_url": "192.168.1.100:17070",
    "vim_user": "admin",
    "vim_password": "c16gylWEepEREN6vWw==",
    "config": {
        "paas_provider": "juju",
        "cloud": cloud,
        "cloud_credentials": "microk8s",
        "authorized_keys": "$HOME/.local/share/juju/ssh/juju_id_rsa.pub",
        "ca_cert_content": "-----BEGIN-----",
    },
}
vim_account_id = "9b0bedc3-ea8e-42fd-acc9-dd03f4dee73c"
vdu_id = "hackfest_basic-VM"
vnf_index = "vnf-profile-id"
sample_vnfr = {
    "_id": vnfr_uuid,
    "id": vnfr_uuid,
    "nsr-id-ref": "dcf4c922-5a73-41bf-a6ca-870c22d6336c",
    "vnfd-ref": "jar_vnfd_scalable",
    "vnfd-id": "f1b38eac-190c-485d-9a74-c6e169c929d8",
    "vim-account-id": vim_account_id,
    "namespace": model_name,
    "member-vnf-index-ref": vnf_index,
}
vnf_config = {
    "member-vnf-index": vnf_index,
    "vdu": [
        {
            "id": vdu_id,
            "configurable-properties": {
                "config::redirect-map": "https://osm.instantiation.params"
            },
        }
    ],
}
app_config = {"domain_name1": "osm.org", "domain_name2": "osm.com"}
config = [
    {"key": "config::domain_name1", "value": "osm.org"},
    {"key": "config::domain_name2", "value": "osm.com"},
    {"key": "track", "value": "latest"},
    {"key": "channel", "value": "stable"},
]
vdu = {
    "id": vdu_id,
    "name": "hackfest_basic-VM",
    "sw-image-desc": "ubuntu18.04",
    "virtual-compute-desc": "hackfest_basic-VM-compute",
    "virtual-storage-desc": ["hackfest_basic-VM-storage"],
    "configurable-properties": config,
}
vnfd_id = "97784f19-d254-4252-946c-cf92d85443ca"
flavor_1 = {
    "id": "compute-id-1",
    "virtual-memory": {"size": "4"},
    "virtual-cpu": {"cpu-architecture": "x86", "num-virtual-cpu": 2},
}
flavor_2 = {
    "id": "compute-id-2",
    "virtual-cpu": {"cpu-architecture": "x86", "num-virtual-cpu": 2},
}
flavor_3 = {
    "id": "compute-id-2",
    "virtual-memory": {"size": "4"},
}
sample_vnfd = {
    "_id": vnfd_id,
    "id": "sol006-vnf",
    "provider": "Canonical",
    "product-name": "test-vnf",
    "software-version": "1.0",
    "vdu": [vdu],
    "virtual-compute-desc": [flavor_1, flavor_2],
}


class TestException(Exception):
    pass


class TestVnfDbActivity(asynctest.TestCase):
    def setUp(self):
        self.db = Mock()
        self.env = ActivityEnvironment()
        self.change_vnf_state = ChangeVnfStateImpl(self.db)
        self.change_vnf_instantiation_state = ChangeVnfInstantiationStateImpl(self.db)
        self.set_vnf_model = SetVnfModelImpl(self.db)

    async def test_change_vnf_state__successful__db_updated_as_expected(self):
        vnf_state = VnfState.STOPPED
        change_vnf_state_input = ChangeVnfStateImpl.Input(vnfr_uuid, vnf_state)
        await self.env.run(self.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__failed__raises_db_exception(self):
        change_vnf_state_input = ChangeVnfStateImpl.Input(vnfr_uuid, VnfState.STARTED)
        self.db.set_one.side_effect = DbException("not found")
        with self.assertRaises(DbException):
            await self.env.run(self.change_vnf_state, change_vnf_state_input)

    async def test_change_vnf_instantiation_state__successful__db_updated_as_expected(
        self,
    ):
        instantiation_state = VnfInstantiationState.NOT_INSTANTIATED
        change_instantiation_input = ChangeVnfInstantiationStateImpl.Input(
            vnfr_uuid, instantiation_state
        )
        await self.env.run(
            self.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__failed__raises_db_exception(self):
        change_instantiation_input = ChangeVnfInstantiationStateImpl.Input(
            vnfr_uuid, VnfInstantiationState.INSTANTIATED
        )
        self.db.set_one.side_effect = DbException("not found")
        with self.assertRaises(DbException):
            await self.env.run(
                self.change_vnf_instantiation_state,
                change_instantiation_input,
            )

    async def test_set_vnf_model__successful__db_updated_as_expected(self):
        await self.env.run(self.set_vnf_model, set_vnf_model_input)
        self.db.set_one.assert_called_with(
            "vnfrs", {"_id": vnfr_uuid}, {"namespace": model_name}
        )

    async def test_set_vnf_model__failed__raises_db_exception(self):
        self.db.set_one.side_effect = DbException("not found")
        with self.assertRaises(DbException):
            await self.env.run(self.set_vnf_model, set_vnf_model_input)


class TestGetTaskQueue(asynctest.TestCase):
    async def setUp(self):
        self.db = Mock()
        self.get_task_queue = GetTaskQueueImpl(self.db)
        self.env = ActivityEnvironment()
        self.get_task_queue_input = GetTaskQueueImpl.Input(vnfr_uuid=sample_vnfr["id"])

    async def test_activity__succeeded__get_expected_task_queue(self):
        self.db.get_one.side_effect = [sample_vnfr, sample_vim_record]
        activity_result = await self.env.run(
            self.get_task_queue,
            self.get_task_queue_input,
        )
        self.assertEqual(activity_result.task_queue, LCM_TASK_QUEUE)

    async def test_activity__failed__raises_db_exception(self):
        self.db.get_one.side_effect = DbException("not found")
        with self.assertRaises(DbException):
            await self.env.run(
                self.get_task_queue,
                self.get_task_queue_input,
            )

    async def test_activity__invalid_task_queue__raises_key_error(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.get_task_queue,
                self.get_task_queue_input,
            )


class TestGetVnfDescriptor(asynctest.TestCase):
    async def setUp(self):
        self.db = Mock()
        self.get_vnf_descriptor = GetVnfDescriptorImpl(self.db)
        self.env = ActivityEnvironment()

    async def test_activity__succeeded__get_expected_vnfd(self):
        self.db.get_one.return_value = sample_vnfd
        activity_result = await self.env.run(
            self.get_vnf_descriptor,
            GetVnfDescriptorImpl.Input(vnfd_uuid=vnfd_uuid),
        )
        self.assertEqual(activity_result.vnfd, sample_vnfd)

    async def test_activity__failed__raises_db_exception(self):
        self.db.get_one.side_effect = DbException("Can not connect to Database.")
        with self.assertRaises(DbException):
            await self.env.run(
                self.get_vnf_descriptor,
                GetVnfDescriptorImpl.Input(vnfd_uuid=vnfd_uuid),
            )


class TestGetVnfRecord(asynctest.TestCase):
    async def setUp(self):
        self.db = Mock()
        self.get_vnf_record = GetVnfRecordImpl(self.db)
        self.env = ActivityEnvironment()

    async def test_activity__succeeded__get_expected_vnfr(self):
        self.db.get_one.return_value = sample_vnfr
        activity_result = await self.env.run(
            self.get_vnf_record,
            GetVnfRecordImpl.Input(vnfr_uuid=vnfr_uuid),
        )
        self.assertEqual(activity_result.vnfr, sample_vnfr)

    async def test_activity__failed__raise_db_exception(self):
        self.db.get_one.side_effect = DbException("Can not connect to Database.")
        with self.assertRaises(DbException):
            await self.env.run(
                self.get_vnf_record,
                GetVnfRecordImpl.Input(vnfr_uuid=vnfr_uuid),
            )


class TestGetVimCloud(asynctest.TestCase):
    async def setUp(self):
        self.db = Mock()
        self.get_vim_cloud = GetVimCloudImpl(self.db)
        self.env = ActivityEnvironment()

    async def test_activity__succeeded__get_vim_cloud(self):
        self.db.get_one.side_effect = [sample_vnfr, sample_vim_record]
        activity_result = await self.env.run(
            self.get_vim_cloud,
            GetVimCloudImpl.Input(vnfr_uuid=sample_vnfr["id"]),
        )
        self.assertEqual(activity_result.cloud, cloud)

    async def test_activity__vim_record_without_cloud__no_cloud_info(self):
        vim_record = deepcopy(sample_vim_record)
        vim_record["config"].pop("cloud")
        self.db.get_one.side_effect = [sample_vnfr, vim_record]
        activity_result = await self.env.run(
            self.get_vim_cloud,
            GetVimCloudImpl.Input(vnfr_uuid=sample_vnfr["id"]),
        )
        self.assertEqual(activity_result.cloud, "")

    async def test_activity__failed__raise_db_exception(self):
        self.db.get_one.side_effect = DbException("Can not connect to Database.")
        with self.assertRaises(DbException):
            await self.env.run(
                self.get_vim_cloud,
                GetVimCloudImpl.Input(vnfr_uuid=sample_vnfr["id"]),
            )

    async def test_activity__wrong_vim_record__raise_key_error(self):
        vim_record = deepcopy(sample_vim_record)
        vim_record.pop("config")
        self.db.get_one.side_effect = [sample_vnfr, vim_record]
        with self.assertRaises(KeyError):
            await self.env.run(
                self.get_vim_cloud,
                GetVimCloudImpl.Input(vnfr_uuid=sample_vnfr["id"]),
            )


class TestGetVduInstantiateInfoMethods(TestCase):
    def test_get_flavor_details__successful__get_flavor(self):
        compute_desc_id = "compute-id-1"
        result = VnfSpecifications._get_flavor_details(compute_desc_id, sample_vnfd)
        self.assertEqual(result, flavor_1)

    def test_get_flavor_details__empty_compute_desc__flavor_is_none(self):
        compute_desc_id = ""
        result = VnfSpecifications._get_flavor_details(compute_desc_id, sample_vnfd)
        self.assertEqual(result, None)

    def test_get_flavor_details__compute_desc_not_found__flavor_is_none(self):
        compute_desc_id = "compute-id-5"
        result = VnfSpecifications._get_flavor_details(compute_desc_id, sample_vnfd)
        self.assertEqual(result, None)

    def test_get_flavor_details__empty_vnfd__flavor_is_none(self):
        compute_desc_id = "compute-id-5"
        result = VnfSpecifications._get_flavor_details(compute_desc_id, {})
        self.assertEqual(result, None)

    def test_get_flavor_details__wrong_vnfd_format__flavor_is_none(self):
        compute_desc_id = "compute-id-2"
        sample_vnfd = {
            "_id": vnfd_id,
            "vdu": [vdu],
            "virtual-compute-desc": [
                {
                    "virtual-memory": {"size": "4"},
                }
            ],
        }
        result = VnfSpecifications._get_flavor_details(compute_desc_id, sample_vnfd)
        self.assertEqual(result, None)

    @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_flavor_details")
    def test_get_compute_constraints__succeeded__get_constraints(
        self, mock_get_flavor_details
    ):
        mock_get_flavor_details.return_value = flavor_1
        result = VnfSpecifications.get_compute_constraints(vdu, sample_vnfd)
        self.assertEqual(VduComputeConstraints(cores=2, mem=4), result)

    @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_flavor_details")
    def test_get_compute_constraints__empty_flavor_details__no_constraints(
        self, mock_get_flavor_details
    ):
        mock_get_flavor_details.return_value = {}
        result = VnfSpecifications.get_compute_constraints(vdu, sample_vnfd)
        self.assertEqual(VduComputeConstraints(cores=0, mem=0), result)

    @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_flavor_details")
    def test_get_compute_constraints__flavor_details_is_none__no_constraints(
        self, mock_get_flavor_details
    ):
        mock_get_flavor_details.return_value = None
        result = VnfSpecifications.get_compute_constraints(vdu, sample_vnfd)
        self.assertEqual(VduComputeConstraints(cores=0, mem=0), result)

    @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_flavor_details")
    def test_get_compute_constraints__flavor_has_only_cpu__get_cpu_constraint(
        self, mock_get_flavor_details
    ):
        mock_get_flavor_details.return_value = flavor_2
        result = VnfSpecifications.get_compute_constraints(vdu, sample_vnfd)
        self.assertEqual(VduComputeConstraints(cores=2, mem=0), result)

    @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_flavor_details")
    def test_get_compute_constraints__flavor_has_only_memory__get_memory_constraint(
        self, mock_get_flavor_details
    ):
        mock_get_flavor_details.return_value = flavor_3
        result = VnfSpecifications.get_compute_constraints(vdu, sample_vnfd)
        self.assertEqual(VduComputeConstraints(cores=0, mem=4), result)

    def test_list_to_dict__config_has_several_items__get_expected_dict(self):
        config = [
            {"key": "config::domain_name1", "value": "osm.org"},
            {"key": "domain_name2", "value": "osm.com"},
        ]
        result = VnfSpecifications._list_to_dict(config)
        self.assertEqual(
            result, {"config::domain_name1": "osm.org", "domain_name2": "osm.com"}
        )

    def test_list_to_dict__empty_input__get_empty_dict(self):
        config = []
        result = VnfSpecifications._list_to_dict(config)
        self.assertEqual(result, {})

    def test_get_only_config_items__with_identifier__get_expected_config(self):
        config = {"config::redirect-map": "https://osm.instantiation.params"}
        result = VnfSpecifications._get_only_config_items(config)
        self.assertEqual(result, {"redirect-map": "https://osm.instantiation.params"})

    def test_get_only_config_items__without_identifier__no_config(self):
        config = {"key": "domain_name1", "value": "osm.org"}
        result = VnfSpecifications._get_only_config_items(config)
        self.assertEqual(result, {})

    def test_get_only_config_items__empty_input__no_config(self):
        config = {}
        result = VnfSpecifications._get_only_config_items(config)
        self.assertEqual(result, {})

    def test_get_vdu_instantiation_params__empty_vnf_config__get_params(self):
        result = VnfSpecifications.get_vdu_instantiation_params(vdu_id, vnf_config)
        self.assertEqual(
            result, {"config::redirect-map": "https://osm.instantiation.params"}
        )

    def test_get_vdu_instantiation_params__empty_vnf_config__no_params(self):
        result = VnfSpecifications.get_vdu_instantiation_params(vdu_id, {})
        self.assertEqual(result, {})

    def test_get_vdu_instantiation_params__vdu_id_mismatch__no_params(self):
        config = deepcopy(vnf_config)
        config["vdu"][0]["id"] = "other-vdu-id"
        result = VnfSpecifications.get_vdu_instantiation_params(vdu_id, config)
        self.assertEqual(result, {})

    def test_get_vdu_instantiation_params__empty_configurable_properties__no_params(
        self,
    ):
        config = deepcopy(vnf_config)
        config["vdu"][0]["configurable-properties"] = {}
        result = VnfSpecifications.get_vdu_instantiation_params(vdu_id, config)
        self.assertEqual(result, {})

    @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_only_config_items")
    @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._list_to_dict")
    def test_get_application_config__instantiate_config_overrides_vdu_config__get_expected_application_config(
        self,
        mock_list_to_dict,
        mock_get_only_config_items,
    ):
        vdu_instantiate_config = {"config::domain_name1": "osm.public"}
        mock_get_only_config_items.side_effect = [
            {
                "domain_name1": "osm.org",
                "domain_name2": "osm.com",
            },
            {"domain_name1": "osm.public"},
        ]
        result = VnfSpecifications.get_application_config(vdu, vdu_instantiate_config)
        self.assertEqual(
            result, {"domain_name1": "osm.public", "domain_name2": "osm.com"}
        )

    @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_only_config_items")
    @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._list_to_dict")
    def test_get_application_config__empty_instantiate_config__get_descriptor_config(
        self,
        mock_list_to_dict,
        mock_get_only_config_items,
    ):
        vdu_instantiate_config = {}
        mock_get_only_config_items.side_effect = [
            {
                "domain_name1": "osm.org",
                "domain_name2": "osm.com",
            },
            {},
        ]
        result = VnfSpecifications.get_application_config(vdu, vdu_instantiate_config)
        self.assertEqual(result, app_config)

    @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._get_only_config_items")
    @patch("osm_lcm.temporal.vnf_activities.VnfSpecifications._list_to_dict")
    def test_get_application_config__no_config__empty_application_config(
        self,
        mock_list_to_dict,
        mock_get_only_config_items,
    ):
        vdu_instantiate_config = {}
        vdu_copy = deepcopy(vdu)
        vdu_copy["configurable-properties"] = []
        mock_get_only_config_items.side_effect = [{}, {}]
        result = VnfSpecifications.get_application_config(
            vdu_copy, vdu_instantiate_config
        )
        self.assertEqual(result, {})


if __name__ == "__main__":
    asynctest.main()