openstack/aodh
Code Issues Proposed changes

Merge "Replace mox with mock in tests.collector"

Browse Source
This commit is contained in:
Jenkins 2013-11-21 05:34:48 +00:00 committed by Gerrit Code Review
commit d7c0e68342
2 changed files with 40 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
ceilometer/tests/collector/dispatcher/test_db.py
View File

@ -19,12 +19,12 @@
"""
from datetime import datetime
import mock
from ceilometer.collector.dispatcher import database
from ceilometer.openstack.common.fixture import config
from ceilometer.openstack.common.fixture import moxstubout
from ceilometer.openstack.common import test
from ceilometer.publisher import rpc
from ceilometer.storage import base
class TestDispatcherDB(test.BaseTestCase):
@ -34,7 +34,6 @@ class TestDispatcherDB(test.BaseTestCase):
self.CONF = self.useFixture(config.Config()).conf
self.dispatcher = database.DatabaseDispatcher(self.CONF)
self.ctx = None
self.mox = self.useFixture(moxstubout.MoxStubout()).mox
def test_valid_message(self):
msg = {'counter_name': 'test',
@ -46,12 +45,11 @@ class TestDispatcherDB(test.BaseTestCase):
self.CONF.publisher_rpc.metering_secret,
)
self.dispatcher.storage_conn = self.mox.CreateMock(base.Connection)
self.dispatcher.storage_conn.record_metering_data(msg)
self.mox.ReplayAll()
with mock.patch.object(self.dispatcher.storage_conn,
'record_metering_data') as record_metering_data:
self.dispatcher.record_metering_data(self.ctx, msg)
self.dispatcher.record_metering_data(self.ctx, msg)
self.mox.VerifyAll()
record_metering_data.assert_called_once_with(msg)
def test_invalid_message(self):
msg = {'counter_name': 'test',
@ -85,15 +83,14 @@ class TestDispatcherDB(test.BaseTestCase):
self.CONF.publisher_rpc.metering_secret,
)
expected = {}
expected.update(msg)
expected = msg.copy()
expected['timestamp'] = datetime(2012, 7, 2, 13, 53, 40)
self.dispatcher.storage_conn = self.mox.CreateMock(base.Connection)
self.dispatcher.storage_conn.record_metering_data(expected)
self.mox.ReplayAll()
with mock.patch.object(self.dispatcher.storage_conn,
'record_metering_data') as record_metering_data:
self.dispatcher.record_metering_data(self.ctx, msg)
self.dispatcher.record_metering_data(self.ctx, msg)
record_metering_data.assert_called_once_with(expected)
def test_timestamp_tzinfo_conversion(self):
msg = {'counter_name': 'test',
@ -106,12 +103,11 @@ class TestDispatcherDB(test.BaseTestCase):
self.CONF.publisher_rpc.metering_secret,
)
expected = {}
expected.update(msg)
expected = msg.copy()
expected['timestamp'] = datetime(2012, 9, 30, 23, 31, 50, 262000)
self.dispatcher.storage_conn = self.mox.CreateMock(base.Connection)
self.dispatcher.storage_conn.record_metering_data(expected)
self.mox.ReplayAll()
with mock.patch.object(self.dispatcher.storage_conn,
'record_metering_data') as record_metering_data:
self.dispatcher.record_metering_data(self.ctx, msg)
self.dispatcher.record_metering_data(self.ctx, msg)
record_metering_data.assert_called_once_with(expected)

43
ceilometer/tests/collector/test_service.py
View File

@ -30,7 +30,6 @@ from stevedore.tests import manager as test_manager
from ceilometer.collector import service
from ceilometer.compute import notifications
from ceilometer.openstack.common.fixture import config
from ceilometer.openstack.common.fixture import moxstubout
from ceilometer.openstack.common import timeutils
from ceilometer import sample
from ceilometer.storage import models
@ -95,28 +94,25 @@ class TestCollector(tests_base.BaseTestCase):
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((self.CONF.collector.udp_address,
self.CONF.collector.udp_port))
def stop_udp(anything):
def _make_fake_socket(self):
def recvfrom(size):
# Make the loop stop
self.srv.stop()
return (msgpack.dumps(self.counter), ('127.0.0.1', 12345))
udp_socket.recvfrom(64 * 1024).WithSideEffects(
stop_udp).AndReturn(
(msgpack.dumps(self.counter),
('127.0.0.1', 12345)))
sock = mock.Mock()
sock.recvfrom = recvfrom
return sock
self.mox.ReplayAll()
return udp_socket
def _verify_udp_socket(self, udp_socket):
conf = self.CONF.collector
udp_socket.setsockopt.assert_called_once_with(socket.SOL_SOCKET,
socket.SO_REUSEADDR, 1)
udp_socket.bind.assert_called_once_with((conf.udp_address,
conf.udp_port))
def setUp(self):
super(TestUDPCollectorService, self).setUp()
self.mox = self.useFixture(moxstubout.MoxStubout()).mox
self.srv = service.UDPCollectorService()
self.counter = sample.Sample(
name='foobar',
@ -145,9 +141,12 @@ class TestUDPCollectorService(TestCollector):
self.counter['counter_type'] = self.counter['type']
self.counter['counter_unit'] = self.counter['unit']
with patch('socket.socket', self._make_fake_socket):
udp_socket = self._make_fake_socket()
with patch('socket.socket', return_value=udp_socket):
self.srv.start()
self._verify_udp_socket(udp_socket)
mock_dispatcher.record_metering_data.assert_called_once_with(
self.counter)
@ -168,9 +167,12 @@ class TestUDPCollectorService(TestCollector):
self.counter['counter_type'] = self.counter['type']
self.counter['counter_unit'] = self.counter['unit']
with patch('socket.socket', self._make_fake_socket):
udp_socket = self._make_fake_socket()
with patch('socket.socket', return_value=udp_socket):
self.srv.start()
self._verify_udp_socket(udp_socket)
mock_dispatcher.record_metering_data.assert_called_once_with(
self.counter)
@ -179,10 +181,13 @@ class TestUDPCollectorService(TestCollector):
raise Exception
def test_udp_receive_bad_decoding(self):
with patch('socket.socket', self._make_fake_socket):
udp_socket = self._make_fake_socket()
with patch('socket.socket', return_value=udp_socket):
with patch('msgpack.loads', self._raise_error):
self.srv.start()
self._verify_udp_socket(udp_socket)
class MyException(Exception):
pass