deb-ceilometer/tests/test_tools_notificationclient.py
Julien Danjou d9faf1c204 Combined fix to get past broken state of repo
This patch includes several separate sets of changes which
have been reviewed individually but must be submitted
together to get past the newly updated version of pep8
and changes to the nova code base.

1. Replace our CONF object with the one from openstack.common.cfg.
   There's no need to use our own, especially since some option
   are already registered on it for us.

   Signed-off-by: Julien Danjou <julien.danjou@enovance.com>

2. Instead of importing the RPC code from nova, use
   the openstack.common.rpc package. This change
   copies that code in from openstack-common, changes
   the imports throughout ceilometer, and fixes the
   way the configuration settings are initialized.

3. Resolve PEP-8 issues introduced by an even more
   pedantic version of pep8 (1.3.1). Some of the
   changes are fixed, and some warnings/errors are
   suppressed.

4. Ignore import errors in impl_qpid because the qpid
   package is not installed on the stackforge test server.

5. Fix missing imports from gettextutils in openstack.common.

Change-Id: I0ee7d4b3278d8ff1951ca27592e3be8a87fe4854
2012-06-22 11:31:53 -04:00

41 lines
1.2 KiB
Python

import os
import cPickle as pickle
from StringIO import StringIO
import sys
import types
import mox
from ceilometer.openstack.common.rpc import impl_kombu
# The module being tested is part of the tools directory,
# so make sure it is in our import path.
sys.path.insert(0, os.path.normpath(os.path.join(os.path.dirname(__file__),
'..', 'tools')))
import notificationclient
def test_send_messages():
message = {'timestamp': 'date goes here',
'event_type': 'compute.instance.exists',
# real messages have more fields...
}
input = StringIO(pickle.dumps(message))
conn = mox.MockObject(impl_kombu.Connection)
conn.topic_send('notifications.info', message)
mox.Replay(conn)
notificationclient.send_messages(conn, 'notifications.info', input)
mox.Verify(conn)
return
def test_record_messages():
conn = mox.MockObject(impl_kombu.Connection)
conn.declare_topic_consumer('notifications.info',
mox.IsA(types.FunctionType))
conn.consume()
mox.Replay(conn)
notificationclient.record_messages(conn, 'notifications.info', StringIO())
mox.Verify(conn)
return