osm_lcm/tests/test_lcm_utils.py

# Copyright 2022 Canonical Ltd.
#
# 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.
#
# For those usages not covered by the Apache License, Version 2.0 please
# contact: alfonso.tiernosepulveda@telefonica.com
##
import logging
import os
from unittest.mock import Mock, patch, MagicMock
from unittest import TestCase

from osm_common.msgkafka import MsgKafka
from osm_common import fslocal
from osm_lcm.data_utils.database.database import Database
from osm_lcm.data_utils.filesystem.filesystem import Filesystem
from osm_lcm.lcm_utils import LcmBase, LcmException
from osm_lcm.tests import test_db_descriptors as descriptors
import yaml
from zipfile import BadZipfile


class TestLcmBase(TestCase):

    test_nsr_id = "b63aa1ba-996e-43a7-921a-1aca5ccbc63f"
    nsd_package_path = "/" + test_nsr_id
    nsd_package_name = "test_nsd"
    charm_metadata_file = "/path/charm/metadata.yaml"

    def setUp(self):
        # DB
        Database.instance = None
        self.db = Database({"database": {"driver": "memory"}}).instance.db
        self.db.create_list("nsds", yaml.safe_load(descriptors.db_nsds_text))
        self.db.create_list("nsrs", yaml.safe_load(descriptors.db_nsrs_text))
        # Filesystem
        self.fs = Filesystem({"storage": {"driver": "local", "path": "/"}}).instance.fs
        # Create LCMBase class
        self.msg = Mock(MsgKafka())
        self.logger = Mock(logging)
        self.my_ns = LcmBase(self.msg, self.logger)
        self.my_ns.fs = self.fs
        self.my_ns.db = self.db

    def test_get_charm_name_successfully(self):
        instance = self.my_ns
        mock_open = MagicMock(open)
        mock_yaml = MagicMock(yaml)
        mock_yaml.safe_load.return_value = {"name": "test_charm"}
        expected_result = "test_charm"

        with patch("osm_lcm.lcm_utils.open", mock_open), patch(
            "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml.safe_load
        ):
            result = instance.get_charm_name(TestLcmBase.charm_metadata_file)
            self.assertEqual(result, expected_result, "wrong charm name")
            self.assertEqual(mock_yaml.safe_load.call_count, 1)
            self.assertEqual(mock_open.call_count, 1)

    def test_get_charm_name_can_not_open_metadata_file(self):
        instance = self.my_ns
        mock_open = MagicMock(open)
        mock_open.side_effect = IOError
        mock_yaml = MagicMock(create_autospec=True)

        with patch("osm_lcm.lcm_utils.open", mock_open), patch(
            "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml.safe_load
        ):
            with self.assertRaises(IOError):
                instance.get_charm_name(TestLcmBase.charm_metadata_file)
                mock_yaml.safe_load.assert_not_called()
                self.assertEqual(mock_open.call_count, 1)

    def test_get_charm_name_wrong_metadata_file_format(self):
        instance = self.my_ns
        mock_open = MagicMock(open)
        mock_yaml = MagicMock(create_autospec=True)
        mock_yaml.safe_load.return_value = {}

        with patch("osm_lcm.lcm_utils.open", mock_open), patch(
            "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml.safe_load
        ):
            with self.assertRaises(KeyError):
                instance.get_charm_name(TestLcmBase.charm_metadata_file)
                self.assertEqual(mock_open.call_count, 1)
                self.assertEqual(mock_yaml.safe_load.call_count, 1)

    def test_get_charm_path_successfully(self):
        instance = self.my_ns
        fs = fslocal.FsLocal()
        fs.path = "/app/storage"
        instance.fs = fs
        charm_folder_name = "simple_charm"
        expected_result = "/app/storage/b63aa1ba-996e-43a7-921a-1aca5ccbc63f/test_nsd/charms/simple_charm"
        result = instance._get_charm_path(
            TestLcmBase.nsd_package_path, TestLcmBase.nsd_package_name, charm_folder_name
        )
        self.assertEqual(result, expected_result, "wrong_charm_path")

    def test_get_charm_metadata_file_charm_is_not_zipped(self):
        instance = self.my_ns
        fs = fslocal.FsLocal()
        fs.path = "/app/storage"
        instance.fs = fs
        mock_zipfile = MagicMock(create_autospec=True)
        charm_folder_name = "simple_charm"
        charm_path = "/app/storage/b63aa1ba-996e-43a7-921a-1aca5ccbc63f/test_nsd/charms/simple_charm"
        expected_result = "/app/storage/b63aa1ba-996e-43a7-921a-1aca5ccbc63f/test_nsd/charms/simple_charm/metadata.yaml"

        with patch("osm_lcm.lcm_utils.ZipFile", mock_zipfile):
            result = instance._get_charm_metadata_file(
                charm_folder_name,
                TestLcmBase.nsd_package_path,
                TestLcmBase.nsd_package_name,
                charm_path=charm_path,
            )
            self.assertEqual(result, expected_result, "wrong charm metadata path")
            mock_zipfile.assert_not_called()

    def test_get_charm_metadata_file_charm_is_zipped(self):
        instance = self.my_ns
        fs = fslocal.FsLocal()
        fs.path = "/app/storage"
        instance.fs = fs
        mock_zipfile = MagicMock(create_autospec=True)
        mock_zipfile.side_effect = None
        charm_folder_name = "simple_charm2.charm"
        charm_path = "/app/storage/b63aa1ba-996e-43a7-921a-1aca5ccbc63f/test_nsd/charms/simple_charm"
        expected_result = "/app/storage/b63aa1ba-996e-43a7-921a-1aca5ccbc63f/test_nsd/charms/simple_charm2/metadata.yaml"

        with patch("osm_lcm.lcm_utils.ZipFile", mock_zipfile):
            result = instance._get_charm_metadata_file(
                charm_folder_name,
                TestLcmBase.nsd_package_path,
                TestLcmBase.nsd_package_name,
                charm_path=charm_path,
            )
            self.assertEqual(result, expected_result, "wrong charm metadata path")
            self.assertEqual(mock_zipfile.call_count, 1)

    def test_find_charm_name_successfully(self):
        db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
        instance = self.my_ns
        mock_listdir = MagicMock(os.listdir())
        mock_listdir.side_effect = None
        mock_charm_path = MagicMock()
        mock_metadata_file = MagicMock()
        mock_metadata_file.return_value = (
            "/b63aa1ba-996e-43a7-921a-1aca5ccbc63f/test_nsd/charms/simple/metadata.yaml"
        )
        mock_charm_name = MagicMock()
        mock_charm_name.return_value = "test_charm"
        expected_result = "test_charm"

        with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
                "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
            ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name), patch(
                "osm_lcm.lcm_utils.os.listdir", mock_listdir
        ):

            result = instance.find_charm_name(db_nsr, "simple")
            self.assertEqual(result, expected_result, "Wrong charm name")
            mock_charm_path.assert_called_once()
            mock_metadata_file.assert_called_once()
            mock_charm_name.assert_called_once_with(
                "/b63aa1ba-996e-43a7-921a-1aca5ccbc63f/test_nsd/charms/simple/metadata.yaml"
            )
            self.assertEqual(mock_listdir.call_count, 1)

    def test_find_charm_name_charm_bad_zipfile(self):
        db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
        instance = self.my_ns
        mock_listdir = MagicMock(os.listdir())
        mock_listdir.side_effect = None
        mock_charm_path = MagicMock()
        mock_metadata_file = MagicMock()
        mock_metadata_file.side_effect = BadZipfile
        mock_charm_name = MagicMock()

        with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
                "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
            ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name), patch(
                "osm_lcm.lcm_utils.os.listdir", mock_listdir
        ):

            with self.assertRaises(LcmException):

                instance.find_charm_name(db_nsr, "simple")
                self.assertEqual(mock_listdir.call_count, 1)
                self.assertEqual(mock_charm_path.call_count, 1)
                self.assertEqual(mock_metadata_file.call_count, 1)
                mock_charm_name.assert_not_called()

    def test_find_charm_name_charm_nsd_package_does_not_exist(self):
        db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
        instance = self.my_ns
        mock_listdir = MagicMock(os.listdir())
        mock_listdir.side_effect = FileNotFoundError
        mock_charm_path = MagicMock()
        mock_metadata_file = MagicMock()
        mock_charm_name = MagicMock()

        with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
                "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
            ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name), patch(
                "osm_lcm.lcm_utils.os.listdir", mock_listdir
        ):

            with self.assertRaises(LcmException):
                instance.find_charm_name(db_nsr, "simple")
                self.assertEqual(mock_listdir.call_count, 1)
                mock_charm_path.assert_not_called()
                mock_metadata_file.assert_not_called()
                mock_charm_name.assert_not_called()

    def test_find_charm_name_missing_input_charm_folder_name(self):
        db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
        instance = self.my_ns
        mock_listdir = MagicMock(os.listdir())
        mock_charm_path = MagicMock()
        mock_metadata_file = MagicMock()
        mock_charm_name = MagicMock()

        with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
                "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
            ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name), patch(
                "osm_lcm.lcm_utils.os.listdir", mock_listdir
        ):

            with self.assertRaises(LcmException):
                instance.find_charm_name(db_nsr, "")
                mock_charm_path.assert_not_called()
                mock_metadata_file.assert_not_called()
                mock_charm_name.assert_not_called()
                mock_listdir.assert_not_called()

    def test_find_charm_name_can_not_open_metadata_file(self):
        db_nsr = self.db.get_one("nsrs", {"_id":  TestLcmBase.test_nsr_id})
        instance = self.my_ns
        mock_listdir = MagicMock(os.listdir())
        mock_charm_path = MagicMock()
        mock_metadata_file = MagicMock()
        mock_charm_name = MagicMock()
        mock_charm_name.side_effect = yaml.YAMLError

        with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
                "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
            ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name), patch(
                "osm_lcm.lcm_utils.os.listdir", mock_listdir
        ):

            with self.assertRaises(LcmException):
                instance.find_charm_name(db_nsr, "simple")
                self.assertEqual(mock_charm_path.call_count, 1)
                self.assertEqual(mock_metadata_file.call_count, 1)
                self.assertEqual(mock_charm_name.call_count, 1)
                self.assertEqual(mock_listdir.call_count, 1)