7a17cd1ae3
At the moment quite vital arguments, such as region_name and valid_interfaces, are ignored by the plugin, which results in inconsistent behaviour with Octavia when trying to interact with Neutron. For instance, while Octavia connects to Neutron through the internal endpoint, the plugin still tries to reach it through the public one, ignoring options defined in [service_auth] and [neutron] sections. This patch is basically a copy-paste from Octavia [1]. [1] https://review.opendev.org/c/openstack/octavia/+/905794 Closes-Bug: #2110488 Related-Bug: #2049551 Change-Id: I3a98825e40143dfa9017ca512a27197c48c31ee9
203 lines
7.1 KiB
Python
203 lines
7.1 KiB
Python
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
#
|
|
from unittest import mock
|
|
|
|
from keystoneauth1 import exceptions as ks_exceptions
|
|
from octavia_lib.api.drivers import exceptions as driver_exceptions
|
|
from oslo_config import cfg
|
|
from oslo_config import fixture as oslo_fixture
|
|
from oslotest import base
|
|
|
|
from ovn_octavia_provider.common import clients
|
|
from ovn_octavia_provider.common import config
|
|
|
|
|
|
class TestKeystoneSession(base.BaseTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
config.register_opts()
|
|
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
|
|
|
|
@mock.patch(
|
|
'keystoneauth1.loading.load_auth_from_conf_options')
|
|
def test_auth(self, kl_auth):
|
|
missing_options = [mock.Mock(dest='username')]
|
|
auth = mock.Mock()
|
|
|
|
# service_auth with missing option
|
|
kl_auth.side_effect = [
|
|
ks_exceptions.auth_plugins.MissingRequiredOptions(missing_options)
|
|
]
|
|
|
|
ksession = clients.KeystoneSession()
|
|
self.assertRaises(
|
|
ks_exceptions.auth_plugins.MissingRequiredOptions,
|
|
lambda: ksession.auth)
|
|
|
|
# neutron with missing option, missing option also in service_auth
|
|
kl_auth.reset_mock()
|
|
kl_auth.side_effect = [
|
|
ks_exceptions.auth_plugins.MissingRequiredOptions(missing_options),
|
|
auth,
|
|
ks_exceptions.auth_plugins.MissingRequiredOptions(missing_options),
|
|
]
|
|
|
|
ksession = clients.KeystoneSession('neutron')
|
|
self.assertRaises(
|
|
ks_exceptions.auth_plugins.MissingRequiredOptions,
|
|
lambda: ksession.auth)
|
|
|
|
# neutron with missing option, it is copied from service_auth
|
|
kl_auth.reset_mock()
|
|
kl_auth.side_effect = [
|
|
ks_exceptions.auth_plugins.MissingRequiredOptions(missing_options),
|
|
auth,
|
|
auth,
|
|
]
|
|
|
|
self.conf.config(group='service_auth',
|
|
endpoint_override='foo')
|
|
|
|
ksession = clients.KeystoneSession('neutron')
|
|
self.assertEqual(ksession.auth, auth)
|
|
self.assertEqual(cfg.CONF.neutron.endpoint_override, 'foo')
|
|
|
|
@mock.patch(
|
|
'keystoneauth1.loading.load_session_from_conf_options')
|
|
@mock.patch(
|
|
'keystoneauth1.loading.load_auth_from_conf_options')
|
|
def test_cached_session(self, kl_auth, kl_session):
|
|
ksession = clients.KeystoneSession('neutron')
|
|
self.assertIs(
|
|
ksession.session,
|
|
ksession.session)
|
|
kl_session.assert_called_once_with(
|
|
mock.ANY, 'neutron', auth=ksession.auth)
|
|
|
|
@mock.patch(
|
|
'keystoneauth1.loading.load_auth_from_conf_options')
|
|
def test_cached_auth(self, kl):
|
|
ksession = clients.KeystoneSession('neutron')
|
|
self.assertIs(
|
|
ksession.auth,
|
|
ksession.auth)
|
|
kl.assert_called_once_with(mock.ANY, 'neutron')
|
|
|
|
|
|
class TestNeutronAuth(base.BaseTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
config.register_opts()
|
|
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
|
|
self.conf.config(group='neutron', region_name='RegionOne',
|
|
valid_interfaces='internal')
|
|
self.mock_client = mock.patch(
|
|
'openstack.connection.Connection').start()
|
|
clients.Singleton._instances = {}
|
|
|
|
@mock.patch.object(clients, 'KeystoneSession')
|
|
def test_init(self, mock_ks):
|
|
clients.NeutronAuth()
|
|
self.mock_client.assert_called_once_with(
|
|
session=mock_ks().session, interface='internal',
|
|
region_name='RegionOne')
|
|
|
|
def test_singleton(self):
|
|
c1 = clients.NeutronAuth()
|
|
c2 = clients.NeutronAuth()
|
|
self.assertIs(c1, c2)
|
|
|
|
def test_singleton_exception(self):
|
|
mock_client = mock.Mock()
|
|
mock_client.network_proxy = 'foo'
|
|
with mock.patch(
|
|
'openstack.connection.Connection',
|
|
side_effect=[RuntimeError,
|
|
mock_client, mock_client]) as os_sdk:
|
|
self.assertRaises(
|
|
RuntimeError,
|
|
clients.NeutronAuth)
|
|
c2 = clients.NeutronAuth()
|
|
c3 = clients.NeutronAuth()
|
|
self.assertIs(c2, c3)
|
|
self.assertEqual(os_sdk._mock_call_count, 2)
|
|
|
|
@mock.patch.object(clients, 'KeystoneSession')
|
|
def test_get_client(self, mock_ks):
|
|
clients.get_neutron_client()
|
|
self.mock_client.assert_called_once_with(
|
|
session=mock_ks().session, interface='internal',
|
|
region_name='RegionOne')
|
|
|
|
@mock.patch.object(clients, 'NeutronAuth', side_effect=[RuntimeError])
|
|
def test_get_client_error(self, mock_ks):
|
|
msg = self.assertRaises(
|
|
driver_exceptions.DriverError,
|
|
clients.get_neutron_client)
|
|
self.assertEqual("An unknown driver error occurred.", str(msg))
|
|
|
|
|
|
class TestOctaviaAuth(base.BaseTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
config.register_opts()
|
|
self.mock_client = mock.patch(
|
|
'openstack.connection.Connection').start()
|
|
clients.Singleton._instances = {}
|
|
|
|
@mock.patch.object(clients, 'KeystoneSession')
|
|
@mock.patch('openstack.connection.Connection')
|
|
def test_init(self, mock_conn, mock_ks):
|
|
clients.OctaviaAuth()
|
|
mock_conn.assert_called_once_with(
|
|
session=mock_ks().session,
|
|
region_name=mock.ANY
|
|
)
|
|
|
|
def test_singleton(self):
|
|
c1 = clients.OctaviaAuth()
|
|
c2 = clients.OctaviaAuth()
|
|
self.assertIs(c1, c2)
|
|
|
|
def test_singleton_exception(self):
|
|
mock_client = mock.Mock()
|
|
mock_client.lbaas_proxy = 'foo'
|
|
with mock.patch(
|
|
'openstack.connection.Connection',
|
|
side_effect=[RuntimeError,
|
|
mock_client, mock_client]) as os_sdk:
|
|
self.assertRaises(
|
|
RuntimeError,
|
|
clients.OctaviaAuth)
|
|
c2 = clients.OctaviaAuth()
|
|
c3 = clients.OctaviaAuth()
|
|
self.assertIs(c2, c3)
|
|
self.assertEqual(os_sdk._mock_call_count, 2)
|
|
|
|
@mock.patch.object(clients, 'KeystoneSession')
|
|
@mock.patch('openstack.connection.Connection')
|
|
def test_get_client(self, mock_conn, mock_ks):
|
|
clients.get_octavia_client()
|
|
mock_conn.assert_called_once_with(
|
|
session=mock_ks().session,
|
|
region_name=mock.ANY
|
|
)
|
|
|
|
@mock.patch.object(clients, 'OctaviaAuth', side_effect=[RuntimeError])
|
|
def test_get_client_error(self, mock_ks):
|
|
msg = self.assertRaises(
|
|
driver_exceptions.DriverError,
|
|
clients.get_octavia_client)
|
|
self.assertEqual("An unknown driver error occurred.", str(msg))
|