--- /dev/null
+# 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)
projects / osm / LCM.git / blobdiff