openstack/aodh
Code Issues Proposed changes

Replace tests.base part1

Browse Source 
It is the first step to replace using ceilometer.tests.base with
openstack.common.test.
Displacement method assertTimestampEqual() from TestCase to
BaseTestCase class in base.py because TestCase
will not be used in the future.

Change-Id: Ic56041516a142590e0ee869e60f939c5245ca977
This commit is contained in:
Vladislav Kuzmin 2013-10-08 08:19:42 -04:00
parent e4a1a4fcef
commit 9da986953d
4 changed files with 28 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
ceilometer/tests/base.py
View File

@ -25,6 +25,7 @@ import os.path
import testtools
from testtools import testcase
from ceilometer.openstack.common import test
from ceilometer.openstack.common import timeutils
from ceilometer.openstack.common.fixture import config
from ceilometer.openstack.common.fixture import moxstubout
@ -67,6 +68,8 @@ class TestCase(testtools.TestCase):
def temp_config_file_path(self, name='ceilometer.conf'):
return os.path.join(self.tempdir.path, name)
class BaseTestCase(test.BaseTestCase):
def assertTimestampEqual(self, first, second, msg=None):
"""Checks that two timestamps are equals.

9
nova_tests/test_notifier.py
View File

@ -50,8 +50,9 @@ config.cfg.CONF.import_opt('compute_manager', 'nova.service')
# HACK(jd) Import this first because of the second HACK below, and because
# of Nova not having these module yet as of this writing
import ceilometer.openstack.common.fixture.config
import ceilometer.openstack.common.fixture.moxstubout
from ceilometer.openstack.common import test
from ceilometer.openstack.common.fixture import config
from ceilometer.openstack.common.fixture import moxstubout
# HACK(dhellmann): Import this before any other ceilometer code
# because the notifier module messes with the import path to force
@ -60,13 +61,12 @@ from ceilometer.compute import nova_notifier
from ceilometer import sample
from ceilometer.compute.pollsters import util
from ceilometer.tests import base
LOG = logging.getLogger(__name__)
nova_CONF = config.cfg.CONF
class TestNovaNotifier(base.TestCase):
class TestNovaNotifier(test.BaseTestCase):
class Pollster(object):
instances = []
@ -107,6 +107,7 @@ class TestNovaNotifier(base.TestCase):
nova_CONF.spice.enabled = False
self.compute = importutils.import_object(nova_CONF.compute_manager)
self.context = context.get_admin_context()
self.stubs = self.useFixture(moxstubout.MoxStubout()).stubs
fake_network.set_stub_network_methods(self.stubs)
self.instance_data = {"display_name": "instance-1",

9
tests/api/v2/test_query.py
View File

@ -22,11 +22,13 @@ import wsme
from ceilometer import storage
from ceilometer.api.controllers import v2 as api
from ceilometer.api.controllers.v2 import Query
from ceilometer.openstack.common import test
from ceilometer.openstack.common import timeutils
from ceilometer.openstack.common.fixture import moxstubout
from ceilometer.tests import base as tests_base
class TestQuery(tests_base.TestCase):
class TestQuery(test.BaseTestCase):
def test_get_value_as_type_with_integer(self):
query = Query(field='metadata.size',
@ -137,7 +139,7 @@ class TestQuery(tests_base.TestCase):
self.assertEqual(query._get_value_as_type(), expected)
class TestValidateGroupByFields(tests_base.TestCase):
class TestValidateGroupByFields(test.BaseTestCase):
def test_valid_field(self):
result = api._validate_groupby_fields(['user_id'])
@ -166,9 +168,10 @@ class TestValidateGroupByFields(tests_base.TestCase):
self.assertEqual(result, set(['user_id', 'source']))
class TestQueryToKwArgs(tests_base.TestCase):
class TestQueryToKwArgs(tests_base.BaseTestCase):
def setUp(self):
super(TestQueryToKwArgs, self).setUp()
self.stubs = self.useFixture(moxstubout.MoxStubout()).stubs
self.stubs.Set(api, '_sanitize_query', lambda x, y, **z: x)
self.stubs.Set(api, '_verify_query_segregation', lambda x, **z: x)

25
tests/collector/test_service.py
View File

@ -24,14 +24,15 @@ import socket
from mock import patch
from mock import MagicMock
from oslo.config import cfg
from stevedore import extension
from stevedore.tests import manager as test_manager
from ceilometer import sample
from ceilometer.collector import service
from ceilometer.compute import notifications
from ceilometer.openstack.common import timeutils
from ceilometer import sample
from ceilometer.openstack.common.fixture import config
from ceilometer.openstack.common.fixture import moxstubout
from ceilometer.storage import models
from ceilometer.tests import base as tests_base
@ -86,18 +87,19 @@ TEST_NOTICE = {
}
class TestCollector(tests_base.TestCase):
class TestCollector(tests_base.BaseTestCase):
def setUp(self):
super(TestCollector, self).setUp()
cfg.CONF.set_override("connection", "log://", group='database')
self.CONF = self.useFixture(config.Config()).conf
self.CONF.set_override("connection", "log://", group='database')
class TestUDPCollectorService(TestCollector):
def _make_fake_socket(self, family, type):
udp_socket = self.mox.CreateMockAnything()
udp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
udp_socket.bind((cfg.CONF.collector.udp_address,
cfg.CONF.collector.udp_port))
udp_socket.bind((self.CONF.collector.udp_address,
self.CONF.collector.udp_port))
def stop_udp(anything):
# Make the loop stop
@ -114,6 +116,7 @@ class TestUDPCollectorService(TestCollector):
def setUp(self):
super(TestUDPCollectorService, self).setUp()
self.mox = self.useFixture(moxstubout.MoxStubout()).mox
self.srv = service.UDPCollectorService()
self.counter = sample.Sample(
name='foobar',
@ -205,7 +208,7 @@ class TestCollectorService(TestCollector):
# If we try to create a real RPC connection, init_host() never
# returns. Mock it out so we can establish the service
# configuration.
cfg.CONF.set_override("store_events", False, group="collector")
self.CONF.set_override("store_events", False, group="collector")
with patch('ceilometer.openstack.common.rpc.create_connection'):
self.srv.start()
self.srv.pipeline_manager.pipelines[0] = MagicMock()
@ -221,14 +224,14 @@ class TestCollectorService(TestCollector):
self.srv.pipeline_manager.publisher.called)
def test_process_notification_no_events(self):
cfg.CONF.set_override("store_events", False, group="collector")
self.CONF.set_override("store_events", False, group="collector")
self.srv.notification_manager = MagicMock()
with patch.object(self.srv, '_message_to_event') as fake_msg_to_event:
self.srv.process_notification({})
self.assertFalse(fake_msg_to_event.called)
def test_process_notification_with_events(self):
cfg.CONF.set_override("store_events", True, group="collector")
self.CONF.set_override("store_events", True, group="collector")
self.srv.notification_manager = MagicMock()
with patch.object(self.srv, '_message_to_event') as fake_msg_to_event:
self.srv.process_notification({})
@ -261,7 +264,7 @@ class TestCollectorService(TestCollector):
self.assertEqual(1, len(event.traits))
def test_message_to_event_duplicate(self):
cfg.CONF.set_override("store_events", True, group="collector")
self.CONF.set_override("store_events", True, group="collector")
mock_dispatcher = MagicMock()
self.srv.dispatcher_manager = test_manager.TestExtensionManager(
[extension.Extension('test',
@ -276,7 +279,7 @@ class TestCollectorService(TestCollector):
self.srv._message_to_event(message) # Should return silently.
def test_message_to_event_bad_event(self):
cfg.CONF.set_override("store_events", True, group="collector")
self.CONF.set_override("store_events", True, group="collector")
mock_dispatcher = MagicMock()
self.srv.dispatcher_manager = test_manager.TestExtensionManager(
[extension.Extension('test',