"Fossies" - the Fresh Open Source Software Archive

Member "tacker-3.0.1/tacker/tests/unit/nfvo/drivers/vim/test_openstack_driver.py" (29 Jul 2021, 10853 Bytes) of package /linux/misc/openstack/tacker-3.0.1.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 latest Fossies "Diffs" side-by-side code changes report for "test_openstack_driver.py": 3.0.0_vs_3.0.1.

    1 # Copyright 2016 Brocade Communications System, Inc.
    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 os
   17 
   18 from keystoneauth1 import exceptions
   19 import mock
   20 from oslo_config import cfg
   21 
   22 from tacker import context as t_context
   23 from tacker.extensions import nfvo
   24 from tacker.nfvo.drivers.vim import openstack_driver
   25 from tacker.tests.unit import base
   26 from tacker.tests.unit.db import utils
   27 
   28 OPTS = [cfg.StrOpt('user_domain_id',
   29                    default='default',
   30                    help='User Domain Id'),
   31         cfg.StrOpt('project_domain_id',
   32                    default='default',
   33                    help='Project Domain Id'),
   34         cfg.StrOpt('password',
   35                    default='default',
   36                    help='User Password'),
   37         cfg.StrOpt('username',
   38                    default='default',
   39                    help='User Name'),
   40         cfg.StrOpt('user_domain_name',
   41                    default='default',
   42                    help='Use Domain Name'),
   43         cfg.StrOpt('project_name',
   44                    default='default',
   45                    help='Project Name'),
   46         cfg.StrOpt('project_domain_name',
   47                    default='default',
   48                    help='Project Domain Name'),
   49         cfg.StrOpt('auth_url',
   50                    default='http://localhost/identity/v3',
   51                    help='Keystone endpoint')]
   52 
   53 cfg.CONF.register_opts(OPTS, 'keystone_authtoken')
   54 CONF = cfg.CONF
   55 
   56 
   57 class FakeKeystone(mock.Mock):
   58     pass
   59 
   60 
   61 class FakeNeutronClient(mock.Mock):
   62     pass
   63 
   64 
   65 class FakeKeymgrAPI(mock.Mock):
   66     pass
   67 
   68 
   69 class mock_dict(dict):
   70     def __getattr__(self, item):
   71         return self.get(item)
   72 
   73     __setattr__ = dict.__setitem__
   74     __delattr__ = dict.__delitem__
   75 
   76 
   77 class TestOpenstack_Driver(base.TestCase):
   78     def setUp(self):
   79         super(TestOpenstack_Driver, self).setUp()
   80         self._mock_keystone()
   81         self.keystone.create_key_dir.return_value = 'test_keys'
   82         self.config_fixture.config(group='vim_keys', openstack='/tmp/')
   83         self.config_fixture.config(group='vim_keys', use_barbican=False)
   84         self.openstack_driver = openstack_driver.OpenStack_Driver()
   85         self.vim_obj = self.get_vim_obj()
   86         self.auth_obj = utils.get_vim_auth_obj()
   87         self.addCleanup(mock.patch.stopall)
   88         self._mock_keymgr()
   89 
   90     def _mock_keystone(self):
   91         self.keystone = mock.Mock(wraps=FakeKeystone())
   92         fake_keystone = mock.Mock()
   93         fake_keystone.return_value = self.keystone
   94         self._mock(
   95             'tacker.vnfm.keystone.Keystone', fake_keystone)
   96 
   97     def _mock_keymgr(self):
   98         self.keymgr = mock.Mock(wraps=FakeKeymgrAPI())
   99         fake_keymgr = mock.Mock()
  100         fake_keymgr.return_value = self.keymgr
  101         self._mock(
  102             'tacker.keymgr.barbican_key_manager.BarbicanKeyManager',
  103             fake_keymgr)
  104 
  105     def get_vim_obj(self):
  106         return {'id': '6261579e-d6f3-49ad-8bc3-a9cb974778ff', 'type':
  107                 'openstack', 'auth_url': 'http://localhost/identity',
  108                 'auth_cred': {'username': 'test_user',
  109                               'password': 'test_password',
  110                               'user_domain_name': 'default',
  111                               'cert_verify': 'True',
  112                               'auth_url': 'http://localhost/identity'},
  113                 'name': 'VIM0',
  114                 'vim_project': {'name': 'test_project',
  115                                 'project_domain_name': 'default'}}
  116 
  117     def get_vim_obj_barbican(self):
  118         return {'id': '6261579e-d6f3-49ad-8bc3-a9cb974778ff', 'type':
  119                 'openstack', 'auth_url': 'http://localhost/identity',
  120                 'auth_cred': {'username': 'test_user',
  121                               'password': 'test_password',
  122                               'user_domain_name': 'default',
  123                               'key_type': 'barbican_key',
  124                               'secret_uuid': 'fake-secret-uuid',
  125                               'cert_verify': 'True',
  126                               'auth_url': 'http://localhost/identity'},
  127                 'name': 'VIM0',
  128                 'vim_project': {'name': 'test_project',
  129                                 'project_domain_name': 'default'}}
  130 
  131     def test_register_keystone_v3(self):
  132         regions = [mock_dict({'id': 'RegionOne'})]
  133         attrs = {'regions.list.return_value': regions}
  134         keystone_version = 'v3'
  135         mock_ks_client = mock.Mock(**attrs)
  136         self.keystone.get_version.return_value = keystone_version
  137         self._test_register_vim(self.vim_obj, mock_ks_client)
  138         mock_ks_client.regions.list.assert_called_once_with()
  139         self.keystone.initialize_client.assert_called_once_with(
  140             **self.auth_obj)
  141 
  142     def _test_register_vim(self, vim_obj, mock_ks_client):
  143         self.keystone.initialize_client.return_value = mock_ks_client
  144         fernet_attrs = {'encrypt.return_value': 'encrypted_password'}
  145         mock_fernet_obj = mock.Mock(**fernet_attrs)
  146         mock_fernet_key = b'test_fernet_key'
  147         self.keystone.create_fernet_key.return_value = (mock_fernet_key,
  148                                                         mock_fernet_obj)
  149         self.openstack_driver.register_vim(vim_obj)
  150         with open('/tmp/' + vim_obj['id'], 'r') as f:
  151             # asserting that file has been written correctly.
  152             self.assertEqual('test_fernet_key', f.read())
  153         mock_fernet_obj.encrypt.assert_called_once_with(mock.ANY)
  154         os.remove('/tmp/' + vim_obj['id'])
  155 
  156     @mock.patch('tacker.nfvo.drivers.vim.openstack_driver.os.remove')
  157     @mock.patch('tacker.nfvo.drivers.vim.openstack_driver.os.path'
  158                 '.join')
  159     def test_deregister_vim(self, mock_os_path, mock_os_remove):
  160         vim_obj = self.get_vim_obj()
  161         vim_id = 'my_id'
  162         vim_obj['id'] = vim_id
  163         file_path = CONF.vim_keys.openstack + '/' + vim_id
  164         mock_os_path.return_value = file_path
  165         self.openstack_driver.deregister_vim(vim_obj)
  166         mock_os_remove.assert_called_once_with(file_path)
  167 
  168     def test_deregister_vim_barbican(self):
  169         self.keymgr.delete.return_value = None
  170         vim_obj = self.get_vim_obj_barbican()
  171         self.openstack_driver.deregister_vim(vim_obj)
  172         self.keymgr.delete.assert_called_once_with(
  173             t_context.generate_tacker_service_context(), 'fake-secret-uuid')
  174 
  175     def test_encode_vim_auth_barbican(self):
  176         self.config_fixture.config(group='vim_keys',
  177                                    use_barbican=True)
  178         fernet_attrs = {'encrypt.return_value': 'encrypted_password'}
  179         mock_fernet_obj = mock.Mock(**fernet_attrs)
  180         mock_fernet_key = 'test_fernet_key'
  181         self.keymgr.store.return_value = 'fake-secret-uuid'
  182         self.keystone.create_fernet_key.return_value = (mock_fernet_key,
  183                                                         mock_fernet_obj)
  184 
  185         vim_obj = self.get_vim_obj()
  186         self.openstack_driver.encode_vim_auth(
  187             vim_obj['id'], vim_obj['auth_cred'])
  188 
  189         self.keymgr.store.assert_called_once_with(
  190             t_context.generate_tacker_service_context(), 'test_fernet_key')
  191         mock_fernet_obj.encrypt.assert_called_once_with(mock.ANY)
  192         self.assertEqual(vim_obj['auth_cred']['key_type'],
  193                          'barbican_key')
  194         self.assertEqual(vim_obj['auth_cred']['secret_uuid'],
  195                          'fake-secret-uuid')
  196 
  197     def test_register_vim_invalid_auth(self):
  198         attrs = {'regions.list.side_effect': exceptions.Unauthorized}
  199         self._test_register_vim_auth(attrs)
  200 
  201     def test_register_vim_missing_auth(self):
  202         attrs = {'regions.list.side_effect': exceptions.BadRequest}
  203         self._test_register_vim_auth(attrs)
  204 
  205     def _test_register_vim_auth(self, attrs):
  206         keystone_version = 'v3'
  207         self.keystone.get_version.return_value = keystone_version
  208         mock_ks_client = mock.Mock(**attrs)
  209         self.keystone.initialize_client.return_value = mock_ks_client
  210         self.assertRaises(nfvo.VimUnauthorizedException,
  211                           self.openstack_driver.register_vim,
  212                           self.vim_obj)
  213         mock_ks_client.regions.list.assert_called_once_with()
  214         self.keystone.initialize_client.assert_called_once_with(
  215             **self.auth_obj)
  216 
  217     def test_get_vim_resource_id(self):
  218         resource_type = 'network'
  219         resource_name = 'net0'
  220         fake_networks = {'networks': [{'id': 'fake-uuid', 'name': 'net0'}]}
  221         fake_neutron_client = FakeNeutronClient()
  222         fake_neutron_client.list_networks.return_value = fake_networks
  223         self.openstack_driver._get_client = mock.Mock(
  224             return_value=fake_neutron_client)
  225 
  226         self.openstack_driver.get_vim_resource_id(
  227             self.vim_obj, resource_type, resource_name)
  228 
  229         self.openstack_driver._get_client.assert_called_once_with(
  230             self.vim_obj, mock.ANY)
  231         fake_neutron_client.list_networks.assert_called_once_with(
  232             **{'name': 'net0'})
  233 
  234     def test_get_vim_resource_id_name_not_unique(self):
  235         resource_type = 'network'
  236         resource_name = 'net0'
  237         fake_networks = {'networks': [{'id': 'fake-uuid-1', 'name': 'net0'},
  238                                       {'id': 'fake-uuid-2', 'name': 'net0'}]}
  239         fake_neutron_client = FakeNeutronClient()
  240         fake_neutron_client.list_networks.return_value = fake_networks
  241         self.openstack_driver._get_client = mock.Mock(
  242             return_value=fake_neutron_client)
  243 
  244         self.assertRaises(nfvo.VimGetResourceNameNotUnique,
  245                           self.openstack_driver.get_vim_resource_id,
  246                           self.vim_obj, resource_type, resource_name)
  247 
  248     def test_get_vim_resource_id_name_not_exist(self):
  249         resource_type = 'network'
  250         resource_name = 'net0'
  251         fake_networks = {'networks': []}
  252         fake_neutron_client = FakeNeutronClient()
  253         fake_neutron_client.list_networks.return_value = fake_networks
  254         self.openstack_driver._get_client = mock.Mock(
  255             return_value=fake_neutron_client)
  256 
  257         self.assertRaises(nfvo.VimGetResourceNotFoundException,
  258                           self.openstack_driver.get_vim_resource_id,
  259                           self.vim_obj, resource_type, resource_name)