"Fossies" - the Fresh Open Source Software Archive

Member "senlin-8.0.0/senlin/tests/unit/engine/test_environment.py" (16 Oct 2019, 13402 Bytes) of package /linux/misc/openstack/senlin-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_environment.py": 4.0.0_vs_6.0.0.

    1 # Licensed under the Apache License, Version 2.0 (the "License"); you may
    2 # not use this file except in compliance with the License. You may obtain
    3 # a copy of the License at
    4 #
    5 #         http://www.apache.org/licenses/LICENSE-2.0
    6 #
    7 # Unless required by applicable law or agreed to in writing, software
    8 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    9 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   10 # License for the specific language governing permissions and limitations
   11 # under the License.
   12 
   13 import glob
   14 import mock
   15 import six
   16 
   17 from senlin.common import exception
   18 from senlin.engine import environment
   19 from senlin.tests.unit.common import base
   20 
   21 fake_env_str = """
   22 parameters:
   23   pa: va
   24   pb: vb
   25 custom_profiles:
   26   prof_1: plugin_1
   27 custom_policies:
   28   policy_2: plugin_2
   29 """
   30 
   31 
   32 class TestEnvironment(base.SenlinTestCase):
   33 
   34     def test_create_global(self):
   35         e = environment.Environment(is_global=True)
   36 
   37         self.assertEqual({}, e.params)
   38         self.assertEqual('profiles', e.profile_registry.registry_name)
   39         self.assertEqual('policies', e.policy_registry.registry_name)
   40         self.assertEqual('drivers', e.driver_registry.registry_name)
   41         self.assertEqual('endpoints', e.endpoint_registry.registry_name)
   42         self.assertTrue(e.profile_registry.is_global)
   43         self.assertTrue(e.policy_registry.is_global)
   44         self.assertTrue(e.driver_registry.is_global)
   45         self.assertTrue(e.endpoint_registry.is_global)
   46 
   47     def test_create_default(self):
   48         ge = environment.global_env()
   49         e = environment.Environment()
   50 
   51         reg_prof = e.profile_registry
   52         reg_plcy = e.policy_registry
   53         reg_driv = e.driver_registry
   54         reg_endp = e.endpoint_registry
   55 
   56         self.assertEqual({}, e.params)
   57         self.assertEqual('profiles', reg_prof.registry_name)
   58         self.assertEqual('policies', reg_plcy.registry_name)
   59         self.assertEqual('drivers', reg_driv.registry_name)
   60         self.assertEqual('endpoints', reg_endp.registry_name)
   61         self.assertFalse(reg_prof.is_global)
   62         self.assertFalse(reg_plcy.is_global)
   63         self.assertFalse(reg_driv.is_global)
   64         self.assertFalse(reg_endp.is_global)
   65         self.assertEqual('profiles', ge.profile_registry.registry_name)
   66         self.assertEqual('policies', ge.policy_registry.registry_name)
   67         self.assertEqual('drivers', ge.driver_registry.registry_name)
   68         self.assertEqual('endpoints', ge.endpoint_registry.registry_name)
   69         self.assertEqual(ge.profile_registry, reg_prof.global_registry)
   70         self.assertEqual(ge.policy_registry, reg_plcy.global_registry)
   71         self.assertEqual(ge.driver_registry, reg_driv.global_registry)
   72         self.assertEqual(ge.endpoint_registry, reg_endp.global_registry)
   73 
   74     def test_create_with_env(self):
   75         env = {
   76             'parameters': {
   77                 'p1': 'v1',
   78                 'p2': True,
   79             },
   80             'custom_profiles': {
   81                 'PROFILE_FOO': 'some.class',
   82                 'PROFILE_BAR': 'other.class',
   83             },
   84             'custom_policies': {
   85                 'POLICY_Alpha': 'package.alpha',
   86                 'POLICY_Beta': 'package.beta',
   87             },
   88         }
   89 
   90         e = environment.Environment(env=env, is_global=True)
   91 
   92         self.assertEqual('v1', e.params['p1'])
   93         self.assertTrue(e.params['p2'])
   94         self.assertEqual('some.class', e.get_profile('PROFILE_FOO'))
   95         self.assertEqual('other.class', e.get_profile('PROFILE_BAR'))
   96         self.assertEqual('package.alpha', e.get_policy('POLICY_Alpha'))
   97         self.assertEqual('package.beta', e.get_policy('POLICY_Beta'))
   98 
   99     def test_parse(self):
  100         env = environment.Environment()
  101         result = env.parse(fake_env_str)
  102 
  103         self.assertEqual('va', result['parameters']['pa'])
  104         self.assertEqual('vb', result['parameters']['pb'])
  105         self.assertEqual('plugin_1', result['custom_profiles']['prof_1'])
  106         self.assertEqual('plugin_2', result['custom_policies']['policy_2'])
  107 
  108         # unknown sections
  109         env_str = "variables:\n  p1: v1"
  110         err = self.assertRaises(ValueError, env.parse, env_str)
  111         self.assertEqual('environment has unknown section "variables"',
  112                          six.text_type(err))
  113 
  114         # omitted sections
  115         env_str = "parameters:\n  p1: v1"
  116         result = env.parse(env_str)
  117         self.assertEqual('v1', result['parameters']['p1'])
  118         self.assertEqual({}, result['custom_profiles'])
  119         self.assertEqual({}, result['custom_policies'])
  120 
  121     def test_parse_empty(self):
  122         env = environment.Environment()
  123         result = env.parse(None)
  124         self.assertEqual({}, result)
  125 
  126     def test_load(self):
  127         env = environment.Environment()
  128         env.load({})
  129         self.assertEqual({}, env.params)
  130         self.assertEqual({}, env.profile_registry._registry)
  131         self.assertEqual({}, env.policy_registry._registry)
  132         self.assertEqual({}, env.driver_registry._registry)
  133 
  134         env_dict = {
  135             'parameters': {
  136                 'P': 'V'
  137             },
  138             'custom_profiles': {
  139                 'C1': 'class1',
  140             },
  141             'custom_policies': {
  142                 'C2': 'class2',
  143             },
  144         }
  145         env.load(env_dict)
  146         self.assertEqual('V', env.params['P'])
  147         self.assertEqual('class1', env.get_profile('C1'))
  148         self.assertEqual('class2', env.get_policy('C2'))
  149 
  150     def test_check_plugin_name(self):
  151         env = environment.Environment()
  152 
  153         for pt in ['Profile', 'Policy', 'Driver', 'Endpoint']:
  154             res = env._check_plugin_name(pt, 'abc')
  155             self.assertIsNone(res)
  156 
  157             ex = self.assertRaises(exception.InvalidPlugin,
  158                                    env._check_plugin_name, pt, '')
  159             self.assertEqual('%s type name not specified' % pt,
  160                              six.text_type(ex))
  161 
  162             ex = self.assertRaises(exception.InvalidPlugin,
  163                                    env._check_plugin_name, pt, None)
  164             self.assertEqual('%s type name not specified' % pt,
  165                              six.text_type(ex))
  166 
  167             for v in [123, {}, ['a'], ('b', 'c'), True]:
  168                 ex = self.assertRaises(exception.InvalidPlugin,
  169                                        env._check_plugin_name, pt, v)
  170             self.assertEqual('%s type name is not a string' % pt,
  171                              six.text_type(ex))
  172 
  173     def test_register_and_get_profile(self):
  174         plugin = mock.Mock()
  175         env = environment.Environment()
  176 
  177         ex = self.assertRaises(exception.ResourceNotFound,
  178                                env.get_profile, 'foo')
  179         self.assertEqual("The profile_type 'foo' could not be found.",
  180                          six.text_type(ex))
  181 
  182         env.register_profile('foo', plugin)
  183         self.assertEqual(plugin, env.get_profile('foo'))
  184 
  185     def test_get_profile_types(self):
  186         env = environment.Environment()
  187         plugin1 = mock.Mock(VERSIONS={'1.0': 'v'})
  188         env.register_profile('foo-1.0', plugin1)
  189         plugin2 = mock.Mock(VERSIONS={'1.2': 'v1'})
  190         env.register_profile('bar-1.2', plugin2)
  191 
  192         actual = env.get_profile_types()
  193         self.assertIn(
  194             {'name': 'foo', 'version': '1.0', 'support_status': {'1.0': 'v'}},
  195             actual)
  196         self.assertIn(
  197             {'name': 'bar', 'version': '1.2', 'support_status': {'1.2': 'v1'}},
  198             actual)
  199 
  200     def test_register_and_get_policy(self):
  201         plugin = mock.Mock()
  202         env = environment.Environment()
  203 
  204         ex = self.assertRaises(exception.ResourceNotFound,
  205                                env.get_policy, 'foo')
  206         self.assertEqual("The policy_type 'foo' could not be found.",
  207                          six.text_type(ex))
  208 
  209         env.register_policy('foo', plugin)
  210         self.assertEqual(plugin, env.get_policy('foo'))
  211 
  212     def test_get_policy_types(self):
  213         env = environment.Environment()
  214         plugin1 = mock.Mock(VERSIONS={'0.1': 'v'})
  215         env.register_policy('foo-0.1', plugin1)
  216         plugin2 = mock.Mock(VERSIONS={'0.1': 'v1'})
  217         env.register_policy('bar-0.1', plugin2)
  218 
  219         actual = env.get_policy_types()
  220         self.assertIn(
  221             {'name': 'foo', 'version': '0.1', 'support_status': {'0.1': 'v'}},
  222             actual)
  223         self.assertIn(
  224             {'name': 'bar', 'version': '0.1', 'support_status': {'0.1': 'v1'}},
  225             actual)
  226 
  227     def test_register_and_get_driver_types(self):
  228         plugin = mock.Mock()
  229         env = environment.Environment()
  230 
  231         ex = self.assertRaises(exception.InvalidPlugin,
  232                                env.get_driver, 'foo')
  233         self.assertEqual('Driver plugin foo is not found.',
  234                          six.text_type(ex))
  235 
  236         env.register_driver('foo', plugin)
  237         self.assertEqual(plugin, env.get_driver('foo'))
  238 
  239     def test_get_driver_types(self):
  240         env = environment.Environment()
  241         plugin1 = mock.Mock(VERSIONS={})
  242         env.register_driver('foo', plugin1)
  243         plugin2 = mock.Mock(VERSIONS={})
  244         env.register_driver('bar', plugin2)
  245 
  246         actual = env.get_driver_types()
  247         self.assertIn(
  248             {'name': 'foo', 'version': '', 'support_status': {'': ''}},
  249             actual)
  250         self.assertIn(
  251             {'name': 'bar', 'version': '', 'support_status': {'': ''}},
  252             actual)
  253 
  254     def test_register_and_get_endpoints(self):
  255         plugin = mock.Mock()
  256         env = environment.Environment()
  257 
  258         ex = self.assertRaises(exception.InvalidPlugin,
  259                                env.get_endpoint, 'foo')
  260         self.assertEqual('Endpoint plugin foo is not found.',
  261                          six.text_type(ex))
  262 
  263         env.register_endpoint('foo', plugin)
  264         self.assertEqual(plugin, env.get_endpoint('foo'))
  265 
  266     def test_read_global_environment(self):
  267         mock_dir = self.patchobject(glob, 'glob')
  268         mock_dir.return_value = ['/etc/senlin/environments/e.yaml']
  269         env_dir = '/etc/senlin/environments'
  270         env_contents = 'parameters:\n  p1: v1'
  271         env = environment.Environment(is_global=True)
  272         with mock.patch('senlin.engine.environment.open',
  273                         mock.mock_open(read_data=env_contents),
  274                         create=True) as mock_open:
  275             env.read_global_environment()
  276 
  277         mock_dir.assert_called_with(env_dir + '/*')
  278         mock_open.assert_called_with('%s/e.yaml' % env_dir)
  279 
  280     def test_empty_environment_dir(self):
  281         mock_dir = self.patchobject(glob, 'glob', return_value=[])
  282         env_dir = '/etc/senlin/environments'
  283         env = environment.Environment()
  284         env.read_global_environment()
  285 
  286         mock_dir.assert_called_once_with(env_dir + '/*')
  287 
  288     def test_read_global_environment_oserror(self):
  289         mock_dir = self.patchobject(glob, 'glob')
  290         mock_dir.side_effect = OSError
  291 
  292         env = environment.Environment(is_global=True)
  293         env_dir = '/etc/senlin/environments'
  294         env.read_global_environment()
  295 
  296         mock_dir.assert_called_once_with(env_dir + '/*')
  297 
  298     def test_read_global_environment_ioerror(self):
  299         mock_dir = self.patchobject(glob, 'glob')
  300         mock_dir.return_value = ['/etc/senlin/environments/e.yaml']
  301         env_dir = '/etc/senlin/environments'
  302         env = environment.Environment(is_global=True)
  303         env_contents = ''
  304 
  305         with mock.patch('senlin.engine.environment.open',
  306                         mock.mock_open(read_data=env_contents),
  307                         create=True) as mock_open:
  308             mock_open.side_effect = IOError
  309             env.read_global_environment()
  310 
  311         mock_dir.assert_called_once_with(env_dir + '/*')
  312         mock_open.assert_called_once_with('%s/e.yaml' % env_dir)
  313 
  314     def test_read_global_environment_parse_error(self):
  315         mock_dir = self.patchobject(glob, 'glob')
  316         mock_dir.return_value = ['/etc/senlin/environments/e.yaml']
  317         env_dir = '/etc/senlin/environments'
  318         env_contents = 'aii$%@@$#7'
  319         env = environment.Environment(is_global=True)
  320 
  321         with mock.patch('senlin.engine.environment.open',
  322                         mock.mock_open(read_data=env_contents),
  323                         create=True) as mock_open:
  324             env.read_global_environment()
  325 
  326         mock_dir.assert_called_once_with(env_dir + '/*')
  327         mock_open.assert_called_once_with('%s/e.yaml' % env_dir)
  328 
  329     @mock.patch.object(environment, '_get_mapping')
  330     def test_global_initialize(self, mock_mapping):
  331         mock_mapping.return_value = [['aaa', mock.Mock()]]
  332 
  333         environment._environment = None
  334         environment.initialize()
  335 
  336         expected = [mock.call('senlin.profiles'),
  337                     mock.call('senlin.policies'),
  338                     mock.call('senlin.drivers'),
  339                     mock.call('senlin.endpoints')]
  340 
  341         self.assertIsNotNone(environment._environment)
  342         self.assertEqual(expected, mock_mapping.call_args_list)
  343         self.assertIsNotNone(environment.global_env().get_profile('aaa'))
  344         self.assertIsNotNone(environment.global_env().get_policy('aaa'))
  345         self.assertIsNotNone(environment.global_env().get_driver('aaa'))
  346         self.assertIsNotNone(environment.global_env().get_endpoint('aaa'))
  347         environment._environment = None