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 import tempfile
  20 from unittest.mock import Mock, patch, MagicMock, mock_open
  21 from unittest import TestCase
  22
  23 from osm_common.msgkafka import MsgKafka
  24 from osm_common import fslocal
  25 from osm_lcm.data_utils.database.database import Database
  26 from osm_lcm.data_utils.filesystem.filesystem import Filesystem
  27 from osm_lcm.lcm_utils import LcmBase, LcmException
  28 from osm_lcm.tests import test_db_descriptors as descriptors
  29 import yaml
  30 from zipfile import BadZipfile
  31
  32
  33 tmpdir = tempfile.mkdtemp()[1]
  34 tmpfile = tempfile.mkstemp()[1]
  35
  36
  37 class TestLcmBase(TestCase):
  38
  39     test_nsr_id = "f48163a6-c807-47bc-9682-f72caef5af85"
  40     test_nsd_id = "8c2f8b95-bb1b-47ee-8001-36dc090678da"
  41     nsd_package_path = "/" + test_nsd_id
  42     nsd_package_name = "test_nsd"
  43     charm_metadata_file = "/path/charm/metadata.yaml"
  44
  45     def setUp(self):
  46         # DB
  47         Database.instance = None
  48         self.db = Database({"database": {"driver": "memory"}}).instance.db
  49         self.db.create_list("nsds", yaml.safe_load(descriptors.db_nsds_text))
  50         self.db.create_list("nsds_revisions", yaml.safe_load(descriptors.db_nsds_text))
  51         self.db.create_list("nsrs", yaml.safe_load(descriptors.db_nsrs_text))
  52         # Filesystem
  53         self.fs = Filesystem({"storage": {"driver": "local", "path": "/"}}).instance.fs
  54         # Create LCMBase class
  55         self.msg = Mock(MsgKafka())
  56         self.logger = Mock(logging)
  57         self.my_ns = LcmBase(self.msg, self.logger)
  58         self.my_ns.fs = self.fs
  59         self.my_ns.db = self.db
  60         self.hexdigest = "031edd7d41651593c5fe5c006f"
  61
  62     def test_get_charm_name_successfully(self):
  63         instance = self.my_ns
  64         mock_open = MagicMock(open)
  65         mock_yaml = MagicMock(yaml)
  66         mock_yaml.safe_load.return_value = {"name": "test_charm"}
  67         expected_result = "test_charm"
  68
  69         with patch("osm_lcm.lcm_utils.open", mock_open), patch(
  70             "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml.safe_load
  71         ):
  72             result = instance.get_charm_name(TestLcmBase.charm_metadata_file)
  73             self.assertEqual(result, expected_result, "wrong charm name")
  74             self.assertEqual(mock_yaml.safe_load.call_count, 1)
  75             self.assertEqual(mock_open.call_count, 1)
  76
  77     def test_get_charm_name_can_not_open_metadata_file(self):
  78         instance = self.my_ns
  79         mock_open = MagicMock(open)
  80         mock_open.side_effect = IOError
  81         mock_yaml = MagicMock(create_autospec=True)
  82
  83         with patch("osm_lcm.lcm_utils.open", mock_open), patch(
  84             "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml.safe_load
  85         ):
  86             with self.assertRaises(IOError):
  87                 instance.get_charm_name(TestLcmBase.charm_metadata_file)
  88                 mock_yaml.safe_load.assert_not_called()
  89                 self.assertEqual(mock_open.call_count, 1)
  90
  91     def test_get_charm_name_wrong_metadata_file_format(self):
  92         instance = self.my_ns
  93         mock_open = MagicMock(open)
  94         mock_yaml = MagicMock(create_autospec=True)
  95         mock_yaml.safe_load.return_value = {}
  96
  97         with patch("osm_lcm.lcm_utils.open", mock_open), patch(
  98             "osm_lcm.lcm_utils.yaml.safe_load", mock_yaml.safe_load
  99         ):
 100             with self.assertRaises(KeyError):
 101                 instance.get_charm_name(TestLcmBase.charm_metadata_file)
 102                 self.assertEqual(mock_open.call_count, 1)
 103                 self.assertEqual(mock_yaml.safe_load.call_count, 1)
 104
 105     def test_get_charm_path_successfully(self):
 106         instance = self.my_ns
 107         fs = fslocal.FsLocal()
 108         fs.path = "/app/storage"
 109         instance.fs = fs
 110         charm_folder_name = "simple_charm"
 111         expected_result = (
 112             "/app/storage/" + TestLcmBase.test_nsd_id + "/test_nsd/charms/simple_charm"
 113         )
 114         result = instance._get_charm_path(
 115             TestLcmBase.nsd_package_path,
 116             TestLcmBase.nsd_package_name,
 117             charm_folder_name,
 118         )
 119         self.assertEqual(result, expected_result, "wrong_charm_path")
 120
 121     def test_get_charm_metadata_file_charm_is_not_zipped(self):
 122         instance = self.my_ns
 123         fs = fslocal.FsLocal()
 124         fs.path = "/app/storage"
 125         instance.fs = fs
 126         mock_zipfile = MagicMock(create_autospec=True)
 127         charm_folder_name = "simple_charm"
 128         charm_path = (
 129             "/app/storage/" + TestLcmBase.test_nsd_id + "/test_nsd/charms/simple_charm"
 130         )
 131         expected_result = (
 132             "/app/storage/"
 133             + TestLcmBase.test_nsd_id
 134             + "/test_nsd/charms/simple_charm/metadata.yaml"
 135         )
 136
 137         with patch("osm_lcm.lcm_utils.ZipFile", mock_zipfile):
 138             result = instance._get_charm_metadata_file(
 139                 charm_folder_name,
 140                 TestLcmBase.nsd_package_path,
 141                 TestLcmBase.nsd_package_name,
 142                 charm_path=charm_path,
 143             )
 144             self.assertEqual(result, expected_result, "wrong charm metadata path")
 145             mock_zipfile.assert_not_called()
 146
 147     def test_get_charm_metadata_file_charm_is_zipped(self):
 148         instance = self.my_ns
 149         fs = fslocal.FsLocal()
 150         fs.path = "/app/storage"
 151         instance.fs = fs
 152         mock_zipfile = MagicMock(create_autospec=True)
 153         mock_zipfile.side_effect = None
 154         charm_folder_name = "ubuntu_18.04_simple_charm2.charm"
 155         charm_path = (
 156             "/app/storage/" + TestLcmBase.test_nsd_id + "/test_nsd/charms/simple_charm"
 157         )
 158         expected_result = (
 159             "/app/storage/"
 160             + TestLcmBase.test_nsd_id
 161             + "/test_nsd/charms/ubuntu_18.04_simple_charm2/metadata.yaml"
 162         )
 163
 164         with patch("osm_lcm.lcm_utils.ZipFile", mock_zipfile):
 165             result = instance._get_charm_metadata_file(
 166                 charm_folder_name,
 167                 TestLcmBase.nsd_package_path,
 168                 TestLcmBase.nsd_package_name,
 169                 charm_path=charm_path,
 170             )
 171             self.assertEqual(result, expected_result, "wrong charm metadata path")
 172             self.assertEqual(mock_zipfile.call_count, 1)
 173
 174     def test_find_charm_name_successfully(self):
 175         db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
 176         instance = self.my_ns
 177         mock_charm_path = MagicMock()
 178         mock_metadata_file = MagicMock()
 179         mock_metadata_file.return_value = (
 180             "/" + TestLcmBase.test_nsd_id + "/new_test_nsd/charms/simple/metadata.yaml"
 181         )
 182         mock_charm_name = MagicMock()
 183         mock_charm_name.return_value = "test_charm"
 184         expected_result = "test_charm"
 185
 186         with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
 187             "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
 188         ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name):
 189
 190             result = instance.find_charm_name(db_nsr, "simple")
 191             self.assertEqual(result, expected_result, "Wrong charm name")
 192             mock_charm_path.assert_called_once()
 193             mock_metadata_file.assert_called_once()
 194             mock_charm_name.assert_called_once_with(
 195                 "/"
 196                 + TestLcmBase.test_nsd_id
 197                 + "/new_test_nsd/charms/simple/metadata.yaml"
 198             )
 199
 200     def test_find_charm_name_charm_bad_zipfile(self):
 201         db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
 202         instance = self.my_ns
 203         mock_charm_path = MagicMock()
 204         mock_metadata_file = MagicMock()
 205         mock_metadata_file.side_effect = BadZipfile
 206         mock_charm_name = MagicMock()
 207
 208         with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
 209             "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
 210         ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name):
 211
 212             with self.assertRaises(LcmException):
 213
 214                 instance.find_charm_name(db_nsr, "simple")
 215                 self.assertEqual(mock_charm_path.call_count, 1)
 216                 self.assertEqual(mock_metadata_file.call_count, 1)
 217                 mock_charm_name.assert_not_called()
 218
 219     def test_find_charm_name_missing_input_charm_folder_name(self):
 220         db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
 221         instance = self.my_ns
 222         mock_metadata_file = MagicMock()
 223         mock_charm_name = MagicMock()
 224         mock_charm_path = MagicMock()
 225
 226         with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
 227             "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
 228         ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name):
 229
 230             with self.assertRaises(LcmException):
 231                 instance.find_charm_name(db_nsr, "")
 232                 mock_charm_path.assert_not_called()
 233                 mock_metadata_file.assert_not_called()
 234                 mock_charm_name.assert_not_called()
 235
 236     def test_find_charm_name_can_not_open_metadata_file(self):
 237         db_nsr = self.db.get_one("nsrs", {"_id": TestLcmBase.test_nsr_id})
 238         instance = self.my_ns
 239
 240         mock_charm_path = MagicMock()
 241         mock_metadata_file = MagicMock()
 242         mock_charm_name = MagicMock()
 243         mock_charm_name.side_effect = yaml.YAMLError
 244
 245         with patch("osm_lcm.lcm_utils.LcmBase._get_charm_path", mock_charm_path), patch(
 246             "osm_lcm.lcm_utils.LcmBase._get_charm_metadata_file", mock_metadata_file
 247         ), patch("osm_lcm.lcm_utils.LcmBase.get_charm_name", mock_charm_name):
 248
 249             with self.assertRaises(LcmException):
 250                 instance.find_charm_name(db_nsr, "simple")
 251                 self.assertEqual(mock_charm_path.call_count, 1)
 252                 self.assertEqual(mock_metadata_file.call_count, 1)
 253                 self.assertEqual(mock_charm_name.call_count, 1)
 254
 255     @patch("builtins.open", new_callable=mock_open(read_data="charm content"))
 256     @patch("osm_lcm.lcm_utils.hashlib")
 257     def test_calculate_charm_hash_sucessfully(self, mock_hashlib, mocking_open):
 258         """Calculate charm hash successfully."""
 259         charm = tmpfile
 260         hexdigest = self.hexdigest
 261         mock_file_hash = MagicMock()
 262         mock_hashlib.md5.return_value = mock_file_hash
 263         mock_file_hash.hexdigest.return_value = hexdigest
 264         result = LcmBase.calculate_charm_hash(charm)
 265         self.assertEqual(result, hexdigest)
 266         self.assertEqual(mocking_open.call_count, 1)
 267         self.assertEqual(mock_file_hash.update.call_count, 1)
 268         self.assertEqual(mock_file_hash.hexdigest.call_count, 1)
 269         self.assertEqual(mock_hashlib.md5.call_count, 1)
 270
 271     @patch("builtins.open", new_callable=mock_open(read_data="charm content"))
 272     @patch("osm_lcm.lcm_utils.hashlib")
 273     def test_calculate_charm_hash_open_raises(self, mock_hashlib, mocking_open):
 274         """builtins.open raises exception."""
 275         charm = tmpfile
 276         hexdigest = self.hexdigest
 277         mock_file_hash = MagicMock()
 278         mock_hashlib.md5.return_value = mock_file_hash
 279         mock_file_hash.hexdigest.return_value = hexdigest
 280         mocking_open.side_effect = IOError
 281         with self.assertRaises(IOError):
 282             LcmBase.calculate_charm_hash(charm)
 283         self.assertEqual(mocking_open.call_count, 1)
 284         mock_file_hash.update.assert_not_called()
 285         mock_file_hash.hexdigest.assert_not_called()
 286         self.assertEqual(mock_hashlib.md5.call_count, 1)
 287
 288     @patch("builtins.open", new_callable=mock_open(read_data="charm content"))
 289     @patch("osm_lcm.lcm_utils.hashlib")
 290     def test_calculate_charm_filehash_update_raises(self, mock_hashlib, mocking_open):
 291         """Filehash update raises exception."""
 292         charm = tmpfile
 293         hexdigest = self.hexdigest
 294         mock_file_hash = MagicMock()
 295         mock_file_hash.update.side_effect = Exception
 296         mock_hashlib.md5.return_value = mock_file_hash
 297         mock_file_hash.hexdigest.return_value = hexdigest
 298         with self.assertRaises(Exception):
 299             LcmBase.calculate_charm_hash(charm)
 300         self.assertEqual(mocking_open.call_count, 1)
 301         self.assertEqual(mock_file_hash.update.call_count, 1)
 302         mock_file_hash.hexdigest.assert_not_called()
 303         self.assertEqual(mock_hashlib.md5.call_count, 1)
 304
 305     @patch("builtins.open", new_callable=mock_open(read_data="charm content"))
 306     @patch("osm_lcm.lcm_utils.hashlib")
 307     def test_calculate_charm_filehash_hexdigest_raises(
 308         self, mock_hashlib, mocking_open
 309     ):
 310         """Filehash hexdigest raises exception."""
 311         charm = tmpfile
 312         mock_file_hash = MagicMock()
 313         mock_hashlib.md5.return_value = mock_file_hash
 314         mock_file_hash.hexdigest.side_effect = Exception
 315         with self.assertRaises(Exception):
 316             LcmBase.calculate_charm_hash(charm)
 317         self.assertEqual(mocking_open.call_count, 1)
 318         self.assertEqual(mock_file_hash.update.call_count, 1)
 319         mock_file_hash.hexdigest.assert_called_once()
 320         self.assertEqual(mock_hashlib.md5.call_count, 1)
 321         mock_file_hash.update.assert_called_once()
 322
 323     @patch("builtins.open", new_callable=mock_open(read_data="charm content"))
 324     @patch("osm_lcm.lcm_utils.hashlib")
 325     def test_calculate_charm_filehash_hashlib_md5_raises(
 326         self, mock_hashlib, mocking_open
 327     ):
 328         """Filehash hashlib md5 raises exception."""
 329         charm = tmpfile
 330         mock_hashlib.md5.side_effect = Exception
 331         with self.assertRaises(Exception):
 332             LcmBase.calculate_charm_hash(charm)
 333         self.assertEqual(mock_hashlib.md5.call_count, 1)
 334         mocking_open.assert_not_called()
 335
 336     @patch("builtins.open", new_callable=mock_open(read_data="charm content"))
 337     @patch("osm_lcm.lcm_utils.hashlib")
 338     def test_calculate_charm_hash_file_does_not_exist(self, mock_hashlib, mocking_open):
 339         """Calculate charm hash, charm file does not exist."""
 340         file = None
 341         mock_file_hash = MagicMock()
 342         mock_hashlib.md5.return_value = mock_file_hash
 343         mocking_open.side_effect = FileNotFoundError
 344         with self.assertRaises(FileNotFoundError):
 345             LcmBase.calculate_charm_hash(file)
 346         self.assertEqual(mocking_open.call_count, 1)
 347         mock_file_hash.update.assert_not_called()
 348         mock_file_hash.hexdigest.assert_not_called()
 349         self.assertEqual(mock_hashlib.md5.call_count, 1)
 350
 351     @patch("osm_lcm.lcm_utils.LcmBase.calculate_charm_hash")
 352     def test_compare_charm_hash_charm_changed(self, mock_calculate_charm_hash):
 353         """Compare charm hash, charm files are different."""
 354         output = True
 355         charm1, charm2 = tmpfile, tmpfile
 356         mock_calculate_charm_hash.side_effect = [
 357             self.hexdigest,
 358             "0dd7d4173747593c5fe5c006f",
 359         ]
 360         result = LcmBase.compare_charm_hash(charm1, charm2)
 361         self.assertEqual(output, result)
 362         self.assertEqual(mock_calculate_charm_hash.call_count, 2)
 363
 364     @patch("osm_lcm.lcm_utils.LcmBase.calculate_charm_hash")
 365     def test_compare_charm_hash_charm_is_same(self, mock_calculate_charm_hash):
 366         """Compare charm hash, charm files are same."""
 367         output = False
 368         charm1 = charm2 = tmpfile
 369         mock_calculate_charm_hash.side_effect = [
 370             self.hexdigest,
 371             self.hexdigest,
 372         ]
 373         result = LcmBase.compare_charm_hash(charm1, charm2)
 374         self.assertEqual(output, result)
 375         self.assertEqual(mock_calculate_charm_hash.call_count, 2)
 376
 377     @patch("osm_lcm.lcm_utils.LcmBase.calculate_charm_hash")
 378     def test_compare_charm_hash_one_charm_is_not_valid(self, mock_calculate_charm_hash):
 379         """Compare charm hash, one charm file is not valid."""
 380         charm1, charm2 = tmpdir, None
 381         mock_calculate_charm_hash.side_effect = [
 382             self.hexdigest,
 383             FileNotFoundError,
 384         ]
 385
 386         with self.assertRaises(FileNotFoundError):
 387             LcmBase.compare_charm_hash(charm1, charm2)
 388         self.assertEqual(mock_calculate_charm_hash.call_count, 2)
 389
 390     @patch("osm_lcm.lcm_utils.LcmBase.calculate_charm_hash")
 391     def test_compare_charm_hash_both_charms_are_not_valid(
 392         self, mock_calculate_charm_hash
 393     ):
 394         """Compare charm hash, both charm files are not valid."""
 395         charm1, charm2 = None, None
 396         mock_calculate_charm_hash.side_effect = [IOError, IOError]
 397         with self.assertRaises(IOError):
 398             LcmBase.compare_charm_hash(charm1, charm2)
 399         self.assertEqual(mock_calculate_charm_hash.call_count, 1)
 400
 401     @patch("osm_lcm.lcm_utils.checksumdir")
 402     def test_compare_charmdir_charm_changed(self, mock_checksum):
 403         """Compare charm directory hash, charms are changed."""
 404         expected_output = True
 405         charm_dir1, charm_dir2 = tmpdir, tmpdir
 406         mock_checksum.dirhash.side_effect = [
 407             self.hexdigest,
 408             "031eddtrtr651593c5fe5c006f",
 409         ]
 410         result = LcmBase.compare_charmdir_hash(charm_dir1, charm_dir2)
 411         self.assertEqual(expected_output, result)
 412         self.assertEqual(mock_checksum.dirhash.call_count, 2)
 413
 414     @patch("osm_lcm.lcm_utils.checksumdir")
 415     def test_compare_charmdir_charm_is_same(self, mock_checksum):
 416         """Compare charm directory hash, charms are same."""
 417         expected_output = False
 418         charm_dir1 = charm_dir2 = tmpdir
 419         mock_checksum.dirhash.side_effect = [
 420             self.hexdigest,
 421             self.hexdigest,
 422         ]
 423         result = LcmBase.compare_charmdir_hash(charm_dir1, charm_dir2)
 424         self.assertEqual(expected_output, result)
 425         self.assertEqual(mock_checksum.dirhash.call_count, 2)
 426
 427     @patch("osm_lcm.lcm_utils.checksumdir")
 428     def test_compare_charmdir_one_charmdir_is_not_valid(self, mock_checksum):
 429         """Compare charm directory hash, one charm directory is not valid."""
 430         charm_dir1, charm_dir2 = tmpdir, None
 431         mock_checksum.dirhash.side_effect = [
 432             self.hexdigest,
 433             FileNotFoundError,
 434         ]
 435         with self.assertRaises(FileNotFoundError):
 436             LcmBase.compare_charmdir_hash(charm_dir1, charm_dir2)
 437         self.assertEqual(mock_checksum.dirhash.call_count, 2)
 438
 439     @patch("osm_lcm.lcm_utils.checksumdir")
 440     def test_compare_charmdir_both_charmdirs_are_not_valid(self, mock_checksum):
 441         """Compare charm directory hash, both charm directories are not valid."""
 442         charm_dir1, charm_dir2 = None, None
 443         mock_checksum.dirhash.side_effect = [FileNotFoundError, FileNotFoundError]
 444         with self.assertRaises(FileNotFoundError):
 445             LcmBase.compare_charmdir_hash(charm_dir1, charm_dir2)
 446         self.assertEqual(mock_checksum.dirhash.call_count, 1)