osm_lcm/tests/test_lcm_utils.py

   1 # Copyright 2022 Canonical Ltd.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   4 # not use this file except in compliance with the License. You may obtain
   5 # a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 # License for the specific language governing permissions and limitations
  13 # under the License.
  14 #
  15 # For those usages not covered by the Apache License, Version 2.0 please
  16 # contact: alfonso.tiernosepulveda@telefonica.com
  17 ##
  18 import logging
  19 from unittest.mock import Mock, patch, MagicMock
  20 from unittest import TestCase
  21
  22 from osm_common.msgkafka import MsgKafka
  23 from osm_common import fslocal
  24 from osm_lcm.data_utils.database.database import Database
  25 from osm_lcm.data_utils.filesystem.filesystem import Filesystem
  26 from osm_lcm.lcm_utils import LcmBase, LcmException
  27 from osm_lcm.tests import test_db_descriptors as descriptors
  28 import yaml
  29 from zipfile import BadZipfile
  30
  31
  32 class TestLcmBase(TestCase):
  33
  34     test_nsr_id = "f48163a6-c807-47bc-9682-f72caef5af85"
  35     test_nsd_id = "8c2f8b95-bb1b-47ee-8001-36dc090678da"
  36     nsd_package_path = "/" + test_nsd_id
  37     nsd_package_name = "test_nsd"
  38     charm_metadata_file = "/path/charm/metadata.yaml"
  39
  40     def setUp(self):
  41         # DB
  42         Database.instance = None
  43         self.db = Database({"database": {"driver": "memory"}}).instance.db
  44         self.db.create_list("nsds", yaml.safe_load(descriptors.db_nsds_text))
  45         self.db.create_list("nsds_revisions", yaml.safe_load(descriptors.db_nsds_text))
  46         self.db.create_list("nsrs", yaml.safe_load(descriptors.db_nsrs_text))
  47         # Filesystem
  48         self.fs = Filesystem({"storage": {"driver": "local", "path": "/"}}).instance.fs
  49         # Create LCMBase class
  50         self.msg = Mock(MsgKafka())
  51         self.logger = Mock(logging)
  52         self.my_ns = LcmBase(self.msg, self.logger)
  53         self.my_ns.fs = self.fs
  54         self.my_ns.db = self.db
  55
  56     def test_get_charm_name_successfully(self):
  57         instance = self.my_ns
  58         mock_open = MagicMock(open)
  59         mock_yaml = MagicMock(yaml)
  60         mock_yaml.safe_load.return_value = {"name": "test_charm"}
  61         expected_result = "test_charm"
  62
  63         with patch("osm_lcm.lcm_utils.open", mock_open), patch(
  64             "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml.safe_load
  65         ):
  66             result = instance.get_charm_name(TestLcmBase.charm_metadata_file)
  67             self.assertEqual(result, expected_result, "wrong charm name")
  68             self.assertEqual(mock_yaml.safe_load.call_count, 1)
  69             self.assertEqual(mock_open.call_count, 1)
  70
  71     def test_get_charm_name_can_not_open_metadata_file(self):
  72         instance = self.my_ns
  73         mock_open = MagicMock(open)
  74         mock_open.side_effect = IOError
  75         mock_yaml = MagicMock(create_autospec=True)
  76
  77         with patch("osm_lcm.lcm_utils.open", mock_open), patch(
  78             "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml.safe_load
  79         ):
  80             with self.assertRaises(IOError):
  81                 instance.get_charm_name(TestLcmBase.charm_metadata_file)
  82                 mock_yaml.safe_load.assert_not_called()
  83                 self.assertEqual(mock_open.call_count, 1)
  84
  85     def test_get_charm_name_wrong_metadata_file_format(self):
  86         instance = self.my_ns
  87         mock_open = MagicMock(open)
  88         mock_yaml = MagicMock(create_autospec=True)
  89         mock_yaml.safe_load.return_value = {}
  90
  91         with patch("osm_lcm.lcm_utils.open", mock_open), patch(
  92             "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml.safe_load
  93         ):
  94             with self.assertRaises(KeyError):
  95                 instance.get_charm_name(TestLcmBase.charm_metadata_file)
  96                 self.assertEqual(mock_open.call_count, 1)
  97                 self.assertEqual(mock_yaml.safe_load.call_count, 1)
  98
  99     def test_get_charm_path_successfully(self):
 100         instance = self.my_ns
 101         fs = fslocal.FsLocal()
 102         fs.path = "/app/storage"
 103         instance.fs = fs
 104         charm_folder_name = "simple_charm"
 105         expected_result = (
 106             "/app/storage/" + TestLcmBase.test_nsd_id + "/test_nsd/charms/simple_charm"
 107         )
 108         result = instance._get_charm_path(
 109             TestLcmBase.nsd_package_path,
 110             TestLcmBase.nsd_package_name,
 111             charm_folder_name,
 112         )
 113         self.assertEqual(result, expected_result, "wrong_charm_path")
 114
 115     def test_get_charm_metadata_file_charm_is_not_zipped(self):
 116         instance = self.my_ns
 117         fs = fslocal.FsLocal()
 118         fs.path = "/app/storage"
 119         instance.fs = fs
 120         mock_zipfile = MagicMock(create_autospec=True)
 121         charm_folder_name = "simple_charm"
 122         charm_path = (
 123             "/app/storage/" + TestLcmBase.test_nsd_id + "/test_nsd/charms/simple_charm"
 124         )
 125         expected_result = (
 126             "/app/storage/"
 127             + TestLcmBase.test_nsd_id
 128             + "/test_nsd/charms/simple_charm/metadata.yaml"
 129         )
 130
 131         with patch("osm_lcm.lcm_utils.ZipFile", mock_zipfile):
 132             result = instance._get_charm_metadata_file(
 133                 charm_folder_name,
 134                 TestLcmBase.nsd_package_path,
 135                 TestLcmBase.nsd_package_name,
 136                 charm_path=charm_path,
 137             )
 138             self.assertEqual(result, expected_result, "wrong charm metadata path")
 139             mock_zipfile.assert_not_called()
 140
 141     def test_get_charm_metadata_file_charm_is_zipped(self):
 142         instance = self.my_ns
 143         fs = fslocal.FsLocal()
 144         fs.path = "/app/storage"
 145         instance.fs = fs
 146         mock_zipfile = MagicMock(create_autospec=True)
 147         mock_zipfile.side_effect = None
 148         charm_folder_name = "ubuntu_18.04_simple_charm2.charm"
 149         charm_path = (
 150             "/app/storage/" + TestLcmBase.test_nsd_id + "/test_nsd/charms/simple_charm"
 151         )
 152         expected_result = (
 153             "/app/storage/"
 154             + TestLcmBase.test_nsd_id
 155             + "/test_nsd/charms/ubuntu_18.04_simple_charm2/metadata.yaml"
 156         )
 157
 158         with patch("osm_lcm.lcm_utils.ZipFile", mock_zipfile):
 159             result = instance._get_charm_metadata_file(
 160                 charm_folder_name,
 161                 TestLcmBase.nsd_package_path,
 162                 TestLcmBase.nsd_package_name,
 163                 charm_path=charm_path,
 164             )
 165             self.assertEqual(result, expected_result, "wrong charm metadata path")
 166             self.assertEqual(mock_zipfile.call_count, 1)
 167
 168     def test_find_charm_name_successfully(self):
 169         db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
 170         instance = self.my_ns
 171         mock_charm_path = MagicMock()
 172         mock_metadata_file = MagicMock()
 173         mock_metadata_file.return_value = (
 174             "/" + TestLcmBase.test_nsd_id + "/new_test_nsd/charms/simple/metadata.yaml"
 175         )
 176         mock_charm_name = MagicMock()
 177         mock_charm_name.return_value = "test_charm"
 178         expected_result = "test_charm"
 179
 180         with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
 181             "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
 182         ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name):
 183
 184             result = instance.find_charm_name(db_nsr, "simple")
 185             self.assertEqual(result, expected_result, "Wrong charm name")
 186             mock_charm_path.assert_called_once()
 187             mock_metadata_file.assert_called_once()
 188             mock_charm_name.assert_called_once_with(
 189                 "/"
 190                 + TestLcmBase.test_nsd_id
 191                 + "/new_test_nsd/charms/simple/metadata.yaml"
 192             )
 193
 194     def test_find_charm_name_charm_bad_zipfile(self):
 195         db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
 196         instance = self.my_ns
 197         mock_charm_path = MagicMock()
 198         mock_metadata_file = MagicMock()
 199         mock_metadata_file.side_effect = BadZipfile
 200         mock_charm_name = MagicMock()
 201
 202         with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
 203             "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
 204         ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name):
 205
 206             with self.assertRaises(LcmException):
 207
 208                 instance.find_charm_name(db_nsr, "simple")
 209                 self.assertEqual(mock_charm_path.call_count, 1)
 210                 self.assertEqual(mock_metadata_file.call_count, 1)
 211                 mock_charm_name.assert_not_called()
 212
 213     def test_find_charm_name_missing_input_charm_folder_name(self):
 214         db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
 215         instance = self.my_ns
 216         mock_metadata_file = MagicMock()
 217         mock_charm_name = MagicMock()
 218         mock_charm_path = MagicMock()
 219
 220         with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
 221             "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
 222         ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name):
 223
 224             with self.assertRaises(LcmException):
 225                 instance.find_charm_name(db_nsr, "")
 226                 mock_charm_path.assert_not_called()
 227                 mock_metadata_file.assert_not_called()
 228                 mock_charm_name.assert_not_called()
 229
 230     def test_find_charm_name_can_not_open_metadata_file(self):
 231         db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
 232         instance = self.my_ns
 233
 234         mock_charm_path = MagicMock()
 235         mock_metadata_file = MagicMock()
 236         mock_charm_name = MagicMock()
 237         mock_charm_name.side_effect = yaml.YAMLError
 238
 239         with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
 240             "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
 241         ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name):
 242
 243             with self.assertRaises(LcmException):
 244                 instance.find_charm_name(db_nsr, "simple")
 245                 self.assertEqual(mock_charm_path.call_count, 1)
 246                 self.assertEqual(mock_metadata_file.call_count, 1)
 247                 self.assertEqual(mock_charm_name.call_count, 1)