Add a unit testing configuration fixture

The configuration options registered by oslo.messaging should not be
directly relied upon by users of the library, since those config option
names could change in future.

Add an API which allows API users to override specific configuration
options e.g.

    self.messaging_conf = self.useFixture(messaging.ConfFixture(cfg.CONF))
    self.messaging_conf.transport_driver = 'fake'

Change-Id: If0d837e1b86e3b04237fde522551cfb81505a543
This commit is contained in:
Mark McLoughlin 2013-08-13 12:46:36 +01:00
parent 5aa7c37144
commit fbe3192d9a
7 changed files with 106 additions and 22 deletions

View File

@ -17,6 +17,7 @@ Contents
notifier
serializer
exceptions
conffixture
Release Notes
=============

View File

@ -13,6 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from .conffixture import *
from .exceptions import *
from .localcontext import *
from .notify import *

View File

@ -0,0 +1,87 @@
# Copyright 2013 Red Hat, Inc.
#
# 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.
__all__ = ['ConfFixture']
import sys
import fixtures
def _import_opts(conf, module, opts):
__import__(module)
conf.register_opts(getattr(sys.modules[module], opts))
class ConfFixture(fixtures.Fixture):
"""Tweak configuration options for unit testing.
oslo.messaging registers a number of configuration options, but rather than
directly referencing those options, users of the API should use this
interface for querying and overriding certain configuration options.
An example usage::
self.messaging_conf = self.useFixture(messaging.ConfFixture(cfg.CONF))
self.messaging_conf.transport_driver = 'fake'
:param conf: a ConfigOpts instance
:type conf: oslo.config.cfg.ConfigOpts
"""
def __init__(self, conf):
self.conf = conf
_import_opts(self.conf,
'oslo.messaging._drivers.impl_rabbit', 'rabbit_opts')
_import_opts(self.conf, 'oslo.messaging.rpc.client', '_client_opts')
_import_opts(self.conf, 'oslo.messaging.transport', '_transport_opts')
def setUp(self):
super(ConfFixture, self).setUp()
self.addCleanup(self.conf.reset)
@property
def transport_driver(self):
"""The transport driver - e.g. 'rabbit', 'qpid' or 'fake'."""
return self.conf.rpc_backend
@transport_driver.setter
def transport_driver(self, value):
self.conf.set_override('rpc_backend', value)
@property
def in_memory(self):
"""Use an in-memory transport; currently supported by rabbit driver."""
if (('rabbit' in self.transport_driver or
'kombu' in self.transport_driver)):
return self.conf.fake_rabbit
else:
return False
@in_memory.setter
def in_memory(self, value):
if (('rabbit' in self.transport_driver or
'kombu' in self.transport_driver)):
self.conf.set_override('fake_rabbit', value)
@property
def response_timeout(self):
"""Default number of seconds to wait for a response from a call."""
return self.conf.rpc_response_timeout
@response_timeout.setter
def response_timeout(self, value):
self.conf.set_override('rpc_response_timeout', value)

View File

@ -1,6 +1,9 @@
oslo.config
stevedore
# for oslo.messging.conffixture
fixtures>=0.3.12
# for timeutils
iso8601

View File

@ -5,7 +5,6 @@ flake8==2.0
hacking>=0.5.6,<0.7
discover
fixtures>=0.3.12
mox>=0.5.3
python-subunit
testrepository>=0.0.13

View File

@ -27,7 +27,6 @@ from oslo import messaging
from oslo.messaging._drivers import common as driver_common
from oslo.messaging._drivers import impl_rabbit as rabbit_driver
from oslo.messaging.openstack.common import jsonutils
from oslo.messaging import transport as msg_transport
from tests import utils as test_utils
load_tests = testscenarios.load_tests_apply_scenarios
@ -37,10 +36,8 @@ class TestRabbitDriverLoad(test_utils.BaseTestCase):
def setUp(self):
super(TestRabbitDriverLoad, self).setUp()
self.conf.register_opts(msg_transport._transport_opts)
self.conf.register_opts(rabbit_driver.rabbit_opts)
self.config(rpc_backend='rabbit')
self.config(fake_rabbit=True)
self.messaging_conf.transport_driver = 'rabbit'
self.messaging_conf.in_memory = True
def test_driver_load(self):
transport = messaging.get_transport(self.conf)
@ -81,10 +78,8 @@ class TestRabbitTransportURL(test_utils.BaseTestCase):
def setUp(self):
super(TestRabbitTransportURL, self).setUp()
self.conf.register_opts(msg_transport._transport_opts)
self.conf.register_opts(rabbit_driver.rabbit_opts)
self.config(rpc_backend='rabbit')
self.config(fake_rabbit=True)
self.messaging_conf.transport_driver = 'rabbit'
self.messaging_conf.in_memory = True
def test_transport_url(self):
cnx_init = rabbit_driver.Connection.__init__
@ -139,10 +134,8 @@ class TestSendReceive(test_utils.BaseTestCase):
def setUp(self):
super(TestSendReceive, self).setUp()
self.conf.register_opts(msg_transport._transport_opts)
self.conf.register_opts(rabbit_driver.rabbit_opts)
self.config(rpc_backend='rabbit')
self.config(fake_rabbit=True)
self.messaging_conf.transport_driver = 'rabbit'
self.messaging_conf.in_memory = True
def test_send_receive(self):
transport = messaging.get_transport(self.conf)
@ -313,10 +306,8 @@ class TestRequestWireFormat(test_utils.BaseTestCase):
def setUp(self):
super(TestRequestWireFormat, self).setUp()
self.conf.register_opts(msg_transport._transport_opts)
self.conf.register_opts(rabbit_driver.rabbit_opts)
self.config(rpc_backend='rabbit')
self.config(fake_rabbit=True)
self.messaging_conf.transport_driver = 'rabbit'
self.messaging_conf.in_memory = True
self.uuids = []
self.orig_uuid4 = uuid.uuid4
@ -462,10 +453,8 @@ class TestReplyWireFormat(test_utils.BaseTestCase):
def setUp(self):
super(TestReplyWireFormat, self).setUp()
self.conf.register_opts(msg_transport._transport_opts)
self.conf.register_opts(rabbit_driver.rabbit_opts)
self.config(rpc_backend='rabbit')
self.config(fake_rabbit=True)
self.messaging_conf.transport_driver = 'rabbit'
self.messaging_conf.in_memory = True
def test_reply_wire_format(self):
if hasattr(self, 'skip_msg'):

View File

@ -58,6 +58,10 @@ class BaseTestCase(testtools.TestCase):
self.conf = conf
self.addCleanup(self.conf.reset)
from oslo.messaging import conffixture
self.messaging_conf = self.useFixture(
conffixture.ConfFixture(self.conf))
moxfixture = self.useFixture(moxstubout.MoxStubout())
self.mox = moxfixture.mox
self.stubs = moxfixture.stubs