diff options
Diffstat (limited to 'funtests')
| -rwxr-xr-x | funtests/test_channel.py | 56 | ||||
| -rwxr-xr-x | funtests/test_serialization.py | 59 |
2 files changed, 33 insertions, 82 deletions
diff --git a/funtests/test_channel.py b/funtests/test_channel.py index 503a48e..10b860f 100755 --- a/funtests/test_channel.py +++ b/funtests/test_channel.py @@ -22,15 +22,8 @@ Test amqp.channel module import sys import unittest -try: - bytes -except NameError: - # Python 2.5 and lower - bytes = str - import settings - from amqp import ChannelError, Connection, Message, FrameSyntaxError @@ -163,14 +156,19 @@ class TestChannel(unittest.TestCase): self.assertTrue(isinstance(msg2.body, bytes)) self.assertEqual(msg2.body, u'hello w\xf6rld'.encode('latin_1')) - def test_exception(self): - """ - Check that Channel exceptions are actually raised as Python - exceptions. + def test_queue_delete_empty(self): + self.assertFalse( + self.ch.queue_delete('bogus_queue_that_does_not_exist') + ) - """ + def test_survives_channel_error(self): with self.assertRaises(ChannelError): - self.ch.queue_delete('bogus_queue_that_does_not_exist') + self.ch.queue_declare('krjqheewq_bogus', passive=True) + self.ch.queue_declare('funtest_survive') + self.ch.queue_declare('funtest_survive', passive=True) + self.assertEqual( + 0, self.ch.queue_delete('funtest_survive'), + ) def test_invalid_header(self): """ @@ -181,7 +179,7 @@ class TestChannel(unittest.TestCase): """ qname, _, _ = self.ch.queue_declare() - msg = Message(application_headers={'test': None}) + msg = Message(application_headers={'test': object()}) self.assertRaises( FrameSyntaxError, self.ch.basic_publish, msg, routing_key=qname, @@ -248,10 +246,10 @@ class TestChannel(unittest.TestCase): content_type='text/plain', application_headers={'foo': 7, 'bar': 'baz'}) - self.ch.basic_publish(msg, 'unittest.fanout') - self.ch.basic_publish(msg, 'unittest.fanout', mandatory=True) - self.ch.basic_publish(msg, 'unittest.fanout', mandatory=True) - self.ch.basic_publish(msg, 'unittest.fanout', mandatory=True) + self.ch.basic_publish(msg, 'funtest.fanout') + self.ch.basic_publish(msg, 'funtest.fanout', mandatory=True) + self.ch.basic_publish(msg, 'funtest.fanout', mandatory=True) + self.ch.basic_publish(msg, 'funtest.fanout', mandatory=True) self.ch.close() # @@ -264,38 +262,37 @@ class TestChannel(unittest.TestCase): Network configuration is as follows (-> is forwards to : source_exchange -> dest_exchange -> queue The test checks that once the message is publish to the - destination exchange(unittest.topic_dest) it is delivered to the queue. + destination exchange(funtest.topic_dest) it is delivered to the queue. """ test_routing_key = 'unit_test__key' - dest_exchange = 'unittest.topic_dest_bind' - source_exchange = 'unittest.topic_source_bind' + dest_exchange = 'funtest.topic_dest_bind' + source_exchange = 'funtest.topic_source_bind' self.ch.exchange_declare(dest_exchange, 'topic', auto_delete=True) self.ch.exchange_declare(source_exchange, 'topic', auto_delete=True) qname, _, _ = self.ch.queue_declare() - self.ch.exchange_bind(destination = dest_exchange, - source = source_exchange, - routing_key = test_routing_key) + self.ch.exchange_bind(destination=dest_exchange, + source=source_exchange, + routing_key=test_routing_key) self.ch.queue_bind(qname, dest_exchange, routing_key=test_routing_key) - msg = Message('unittest message', + msg = Message('funtest message', content_type='text/plain', application_headers={'foo': 7, 'bar': 'baz'}) - self.ch.basic_publish(msg, source_exchange, - routing_key = test_routing_key) + routing_key=test_routing_key) msg2 = self.ch.basic_get(qname, no_ack=True) self.assertEqual(msg, msg2) def test_exchange_unbind(self): - dest_exchange = 'unittest.topic_dest_unbind' - source_exchange = 'unittest.topic_source_unbind' + dest_exchange = 'funtest.topic_dest_unbind' + source_exchange = 'funtest.topic_source_unbind' test_routing_key = 'unit_test__key' self.ch.exchange_declare(dest_exchange, @@ -311,6 +308,7 @@ class TestChannel(unittest.TestCase): source=source_exchange, routing_key=test_routing_key) + def main(): suite = unittest.TestLoader().loadTestsFromTestCase(TestChannel) unittest.TextTestRunner(**settings.test_args).run(suite) diff --git a/funtests/test_serialization.py b/funtests/test_serialization.py index 9a8793c..ce70438 100755 --- a/funtests/test_serialization.py +++ b/funtests/test_serialization.py @@ -26,12 +26,6 @@ from random import randint import sys import unittest -try: - bytes -except NameError: - # Python 2.5 and lower - bytes = str - import settings from amqp.serialization import ( @@ -316,7 +310,7 @@ class TestSerialization(unittest.TestCase): Check that an un-serializable table entry raises a ValueError """ - val = {'test': None} + val = {'test': object()} w = AMQPWriter() self.assertRaises(FrameSyntaxError, w.write_table, val) @@ -327,6 +321,7 @@ class TestSerialization(unittest.TestCase): 'baz': 'this is some random string I typed', 'ubaz': u'And something in unicode', 'dday_aniv': datetime(1994, 6, 6), + 'nothing': None, 'more': { 'abc': -123, 'def': 'hello world', @@ -351,26 +346,26 @@ class TestSerialization(unittest.TestCase): # Array # def test_array_from_list(self): - val = [1, 'foo'] + val = [1, 'foo', None] w = AMQPWriter() w.write_array(val) s = w.getvalue() self.assertEqualBinary( - s, '\x00\x00\x00\x0DI\x00\x00\x00\x01S\x00\x00\x00\x03foo', + s, '\x00\x00\x00\x0EI\x00\x00\x00\x01S\x00\x00\x00\x03fooV', ) r = AMQPReader(s) self.assertEqual(r.read_array(), val) def test_array_from_tuple(self): - val = (1, 'foo') + val = (1, 'foo', None) w = AMQPWriter() w.write_array(val) s = w.getvalue() self.assertEqualBinary( - s, '\x00\x00\x00\x0DI\x00\x00\x00\x01S\x00\x00\x00\x03foo', + s, '\x00\x00\x00\x0EI\x00\x00\x00\x01S\x00\x00\x00\x03fooV', ) r = AMQPReader(s) @@ -394,48 +389,6 @@ class TestSerialization(unittest.TestCase): self.assertEqual(r.read_table(), val) # - # Array - # - def test_array_from_list(self): - val = [1, 'foo'] - w = AMQPWriter() - w.write_array(val) - s = w.getvalue() - - self.assertEqualBinary(s, '\x00\x00\x00\x0DI\x00\x00\x00\x01S\x00\x00\x00\x03foo') - - r = AMQPReader(s) - self.assertEqual(r.read_array(), val) - - def test_array_from_tuple(self): - val = (1, 'foo') - w = AMQPWriter() - w.write_array(val) - s = w.getvalue() - - self.assertEqualBinary(s, '\x00\x00\x00\x0DI\x00\x00\x00\x01S\x00\x00\x00\x03foo') - - r = AMQPReader(s) - self.assertEqual(r.read_array(), list(val)) - - def test_table_with_array(self): - val = { - 'foo': 7, - 'bar': Decimal('123345.1234'), - 'baz': 'this is some random string I typed', - 'blist': [1,2,3], - 'nlist': [1, [2,3,4]], - 'ndictl': {'nfoo': 8, 'nblist': [5,6,7] } - } - - w = AMQPWriter() - w.write_table(val) - s = w.getvalue() - - r = AMQPReader(s) - self.assertEqual(r.read_table(), val) - - # # GenericContent # def test_generic_content_eq(self): |