"Fossies" - the Fresh Open Source Software Archive

Member "murano-8.0.0/murano/tests/unit/packages/test_load_utils.py" (16 Oct 2019, 9010 Bytes) of package /linux/misc/openstack/murano-8.0.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the last Fossies "Diffs" side-by-side code changes report for "test_load_utils.py": 3.2.0_vs_4.0.0.

    1 # Copyright (c) 2016 AT&T Corp
    2 # All Rights Reserved.
    3 #
    4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    5 #    not use this file except in compliance with the License. You may obtain
    6 #    a copy of the License at
    7 #
    8 #         http://www.apache.org/licenses/LICENSE-2.0
    9 #
   10 #    Unless required by applicable law or agreed to in writing, software
   11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   13 #    License for the specific language governing permissions and limitations
   14 #    under the License.
   15 
   16 import mock
   17 import os
   18 import random
   19 import shutil
   20 import string
   21 import tempfile
   22 import yaml
   23 import zipfile
   24 
   25 from murano.packages import exceptions
   26 from murano.packages import load_utils
   27 import murano.tests.unit.base as test_base
   28 
   29 
   30 class TestLoadUtils(test_base.MuranoTestCase):
   31 
   32     def setUp(cls):
   33         super(TestLoadUtils, cls).setUp()
   34         cls.temp_directories = []
   35         cls.temp_files = []
   36 
   37     def _create_temp_dir(self):
   38         temp_directory = tempfile.mkdtemp()
   39         self.temp_directories.append(temp_directory)
   40         return temp_directory
   41 
   42     def _create_temp_file(self):
   43         temp_file = tempfile.NamedTemporaryFile(delete=True)
   44         self.temp_files.append(temp_file)
   45         return temp_file
   46 
   47     def _create_temp_zip_file(self, zip_path, manifest_path,
   48                               arcname='manifest.yaml'):
   49         zip_ = zipfile.ZipFile(zip_path, 'w')
   50         zip_.write(manifest_path, arcname=arcname)
   51         zip_.close()
   52         self.temp_files.append(zip_)
   53         return zip_
   54 
   55     def tearDown(cls):
   56         super(TestLoadUtils, cls).tearDown()
   57         for directory in cls.temp_directories:
   58             if os.path.isdir(directory):
   59                 shutil.rmtree(directory)
   60         for file in cls.temp_files:
   61             if isinstance(file, zipfile.ZipFile):
   62                 if zipfile.is_zipfile(file.filename):
   63                     os.remove(file)
   64             else:
   65                 if os.path.isfile(file.name):
   66                     os.remove(file.name)
   67 
   68     def _test_load_from_file(self, target_dir=None, drop_dir=True):
   69         manifest_file_contents = dict(
   70             Format='MuranoPL/1.1',
   71             FullName='test_full_name',
   72             Type='Application',
   73             Description='test_description',
   74             Author='test_author',
   75             Supplier='test_supplier',
   76             Tags=[]
   77         )
   78         test_directory = self._create_temp_dir()
   79         manifest_path = os.path.join(test_directory, 'manifest.yaml')
   80         zip_path = os.path.join(test_directory, 'test_zip_load_utils.zip')
   81 
   82         with open(manifest_path, 'w') as manifest_file:
   83             yaml.dump(manifest_file_contents, manifest_file,
   84                       default_flow_style=True)
   85             self._create_temp_zip_file(zip_path, manifest_path)
   86 
   87         with load_utils.load_from_file(archive_path=zip_path,
   88                                        target_dir=target_dir,
   89                                        drop_dir=drop_dir) as plugin:
   90             self.assertEqual('MuranoPL', plugin.format_name)
   91             self.assertEqual('1.1.0', str(plugin.runtime_version))
   92             self.assertEqual(manifest_file_contents['FullName'],
   93                              plugin.full_name)
   94             self.assertEqual(manifest_file_contents['Description'],
   95                              plugin.description)
   96             self.assertEqual(manifest_file_contents['Author'],
   97                              plugin.author)
   98             self.assertEqual(manifest_file_contents['Supplier'],
   99                              plugin.supplier)
  100             self.assertEqual(manifest_file_contents['Tags'],
  101                              plugin.tags)
  102 
  103     def test_load_from_file(self):
  104         self._test_load_from_file(target_dir=None, drop_dir=True)
  105 
  106     def test_load_from_file_with_custom_target_directory(self):
  107         target_dir = self._create_temp_dir()
  108         self._test_load_from_file(target_dir=target_dir, drop_dir=True)
  109 
  110     @mock.patch('murano.packages.load_utils.get_plugin_loader')
  111     def test_load_from_file_with_invalid_handler(self, mock_plugin_loader):
  112         mock_plugin_loader().get_package_handler = mock.MagicMock(
  113             return_value=None)
  114         test_format = 'Invalid Format'
  115         manifest_file_contents = dict(
  116             Format=test_format,
  117             FullName='test_full_name',
  118             Type='Application',
  119             Description='test_description',
  120             Author='test_author',
  121             Supplier='test_supplier',
  122             Tags=[]
  123         )
  124         test_directory = self._create_temp_dir()
  125         target_dir = self._create_temp_dir()
  126         manifest_path = os.path.join(test_directory, 'manifest.yaml')
  127         zip_path = os.path.join(test_directory, 'test_zip_load_utils.zip')
  128 
  129         with open(manifest_path, 'w') as manifest_file:
  130             yaml.dump(manifest_file_contents, manifest_file,
  131                       default_flow_style=True)
  132             self._create_temp_zip_file(zip_path, manifest_path)
  133 
  134         expected_error_msg = "Unsupported format {0}".format(test_format)
  135         with self.assertRaisesRegex(exceptions.PackageLoadError,
  136                                     expected_error_msg):
  137             with load_utils.load_from_file(archive_path=zip_path,
  138                                            target_dir=target_dir,
  139                                            drop_dir=True):
  140                 pass
  141         mock_plugin_loader().get_package_handler.assert_called_once_with(
  142             test_format)
  143 
  144     def test_load_from_file_with_invalid_archive_path(self):
  145         expected_error_msg = "Unable to find package file"
  146         with self.assertRaisesRegex(exceptions.PackageLoadError,
  147                                     expected_error_msg):
  148             with load_utils.load_from_file('invalid file path'):
  149                 pass
  150 
  151     @mock.patch('murano.packages.load_utils.os')
  152     def test_load_from_file_with_nonempty_target_directory(self, mock_os):
  153         mock_os.listdir = mock.MagicMock(return_value=True)
  154         temp_file = self._create_temp_file()
  155         expected_error_msg = "Target directory is not empty"
  156         with self.assertRaisesRegex(exceptions.PackageLoadError,
  157                                     expected_error_msg):
  158             this_dir = os.path.dirname(os.path.realpath(__file__))
  159             with load_utils.load_from_file(temp_file.name,
  160                                            target_dir=this_dir):
  161                 pass
  162 
  163     def test_load_from_file_without_zip_file(self):
  164         temp_file = self._create_temp_file()
  165         expected_error_msg = "Uploaded file {0} is not a zip archive".\
  166                              format(temp_file.name)
  167         with self.assertRaisesRegex(exceptions.PackageLoadError,
  168                                     expected_error_msg):
  169             with load_utils.load_from_file(temp_file.name):
  170                 pass
  171 
  172     @mock.patch('murano.packages.load_utils.zipfile')
  173     def test_load_from_file_handle_value_error(self, mock_zipfile):
  174         test_error_msg = 'Random error message.'
  175         expected_error_msg = "Couldn't load package from file: {0}".\
  176                              format(test_error_msg)
  177         mock_zipfile.is_zipfile = mock.MagicMock(
  178             side_effect=ValueError(test_error_msg))
  179         temp_file = self._create_temp_file()
  180         with self.assertRaisesRegex(exceptions.PackageLoadError,
  181                                     expected_error_msg):
  182             with load_utils.load_from_file(temp_file.name):
  183                 pass
  184         mock_zipfile.is_zipfile.assert_called_once_with(
  185             temp_file.name)
  186 
  187     def test_load_from_dir_without_source_directory(self):
  188         expected_error_msg = 'Invalid package directory'
  189         with self.assertRaisesRegex(exceptions.PackageLoadError,
  190                                     expected_error_msg):
  191             load_utils.load_from_dir('random_test_directory')
  192 
  193     def test_load_from_dir_with_invalid_source_directory(self):
  194         source_directory = self._create_temp_dir()
  195         expected_error_msg = 'Unable to find package manifest'
  196         with self.assertRaisesRegex(exceptions.PackageLoadError,
  197                                     expected_error_msg):
  198             load_utils.load_from_dir(source_directory)
  199 
  200     @mock.patch('murano.packages.load_utils.os.path.isfile')
  201     def test_load_from_dir_open_file_negative(self, mock_isfile):
  202         mock_isfile.return_value = True
  203         source_directory = self._create_temp_dir()
  204         random_filename = ''.join(random.choice(string.ascii_lowercase)
  205                                   for i in range(20))
  206         expected_error_msg = 'Unable to load due to'
  207         with self.assertRaisesRegex(exceptions.PackageLoadError,
  208                                     expected_error_msg):
  209             load_utils.load_from_dir(source_directory,
  210                                      filename=random_filename)