/usr/lib/python2.7/dist-packages/oslo_messaging/tests/test_opts.py is in python-oslo.messaging 4.6.1-2ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 | # Copyright 2014 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.
import stevedore
import testtools
import mock
from oslo_messaging import server
try:
from oslo_messaging import opts
except ImportError:
opts = None
from oslo_messaging.tests import utils as test_utils
class OptsTestCase(test_utils.BaseTestCase):
@testtools.skipIf(opts is None, "Options not importable")
def setUp(self):
super(OptsTestCase, self).setUp()
def _test_list_opts(self, result):
self.assertEqual(5, len(result))
groups = [g for (g, l) in result]
self.assertIn(None, groups)
self.assertIn('matchmaker_redis', groups)
self.assertIn('oslo_messaging_amqp', groups)
self.assertIn('oslo_messaging_notifications', groups)
self.assertIn('oslo_messaging_rabbit', groups)
opt_names = [o.name for (g, l) in result for o in l]
self.assertIn('rpc_backend', opt_names)
def test_list_opts(self):
self._test_list_opts(opts.list_opts())
def test_entry_point(self):
result = None
for ext in stevedore.ExtensionManager('oslo.config.opts',
invoke_on_load=True):
if ext.name == "oslo.messaging":
result = ext.obj
break
self.assertIsNotNone(result)
self._test_list_opts(result)
def test_defaults(self):
transport = mock.Mock()
transport.conf = self.conf
server.MessageHandlingServer(transport, mock.Mock())
opts.set_defaults(self.conf, executor_thread_pool_size=100)
self.assertEqual(100, self.conf.executor_thread_pool_size)
|