Back to index

rabbitmq-server  2.8.4
rabbitmqadmin-test.py
Go to the documentation of this file.
00001 #!/usr/bin/env python
00002 
00003 import unittest
00004 import os
00005 import os.path
00006 import socket
00007 import subprocess
00008 import sys
00009 
00010 # TODO test: SSL, depth, config file, encodings(?), completion(???)
00011 
00012 class TestRabbitMQAdmin(unittest.TestCase):
00013     def test_no_args(self):
00014         self.run_fail([])
00015 
00016     def test_help(self):
00017         self.run_success(['--help'])
00018         self.run_success(['help', 'subcommands'])
00019         self.run_success(['help', 'config'])
00020         self.run_fail(['help', 'astronomy'])
00021 
00022     def test_host(self):
00023         self.run_success(['show', 'overview'])
00024         self.run_success(['--host', 'localhost', 'show', 'overview'])
00025         self.run_success(['--host', socket.gethostname(), 'show', 'overview'])
00026         self.run_fail(['--host', 'some-host-that-does-not-exist', 'show', 'overview'])
00027 
00028     def test_port(self):
00029         self.run_success(['--port', '55672', 'show', 'overview'])
00030         self.run_fail(['--port', '55673', 'show', 'overview'])
00031         self.run_fail(['--port', '5672', 'show', 'overview'])
00032 
00033     def test_user(self):
00034         self.run_success(['--user', 'guest', '--password', 'guest', 'show', 'overview'])
00035         self.run_fail(['--user', 'no', '--password', 'guest', 'show', 'overview'])
00036         self.run_fail(['--user', 'guest', '--password', 'no', 'show', 'overview'])
00037 
00038     def test_fmt_long(self):
00039         self.assert_output("""
00040 --------------------------------------------------------------------------------
00041 
00042    name: /
00043 tracing: False
00044 
00045 --------------------------------------------------------------------------------
00046 
00047 """, ['--format', 'long', 'list', 'vhosts'])
00048 
00049     def test_fmt_kvp(self):
00050         self.assert_output("""name="/" tracing="False"
00051 """, ['--format', 'kvp', 'list', 'vhosts'])
00052 
00053     def test_fmt_tsv(self):
00054         self.assert_output("""name tracing
00055 /      False
00056 """, ['--format', 'tsv', 'list', 'vhosts'])
00057 
00058     def test_fmt_table(self):
00059         out = """+------+---------+
00060 | name | tracing |
00061 +------+---------+
00062 | /    | False   |
00063 +------+---------+
00064 """
00065         self.assert_output(out, ['list', 'vhosts'])
00066         self.assert_output(out, ['--format', 'table', 'list', 'vhosts'])
00067 
00068     def test_fmt_bash(self):
00069         self.assert_output("""/
00070 """, ['--format', 'bash', 'list', 'vhosts'])
00071 
00072     def test_vhosts(self):
00073         self.assert_list(['/'], l('vhosts'))
00074         self.run_success(['declare', 'vhost', 'name=foo'])
00075         self.assert_list(['/', 'foo'], l('vhosts'))
00076         self.run_success(['delete', 'vhost', 'name=foo'])
00077         self.assert_list(['/'], l('vhosts'))
00078 
00079     def test_users(self):
00080         self.assert_list(['guest'], l('users'))
00081         self.run_fail(['declare', 'user', 'name=foo'])
00082         self.run_success(['declare', 'user', 'name=foo', 'password=pass', 'tags='])
00083         self.assert_list(['foo', 'guest'], l('users'))
00084         self.run_success(['delete', 'user', 'name=foo'])
00085         self.assert_list(['guest'], l('users'))
00086 
00087     def test_permissions(self):
00088         self.run_success(['declare', 'vhost', 'name=foo'])
00089         self.run_success(['declare', 'user', 'name=bar', 'password=pass', 'tags='])
00090         self.assert_table([['guest', '/']], ['list', 'permissions', 'user', 'vhost'])
00091         self.run_success(['declare', 'permission', 'user=bar', 'vhost=foo', 'configure=.*', 'write=.*', 'read=.*'])
00092         self.assert_table([['guest', '/'], ['bar', 'foo']], ['list', 'permissions', 'user', 'vhost'])
00093         self.run_success(['delete', 'user', 'name=bar'])
00094         self.run_success(['delete', 'vhost', 'name=foo'])
00095 
00096     def test_alt_vhost(self):
00097         self.run_success(['declare', 'vhost', 'name=foo'])
00098         self.run_success(['declare', 'permission', 'user=guest', 'vhost=foo', 'configure=.*', 'write=.*', 'read=.*'])
00099         self.run_success(['declare', 'queue', 'name=in_/'])
00100         self.run_success(['--vhost', 'foo', 'declare', 'queue', 'name=in_foo'])
00101         self.assert_table([['/', 'in_/'], ['foo', 'in_foo']], ['list', 'queues', 'vhost', 'name'])
00102         self.run_success(['--vhost', 'foo', 'delete', 'queue', 'name=in_foo'])
00103         self.run_success(['delete', 'queue', 'name=in_/'])
00104         self.run_success(['delete', 'vhost', 'name=foo'])
00105 
00106     def test_exchanges(self):
00107         self.run_success(['declare', 'exchange', 'name=foo', 'type=direct'])
00108         self.assert_list(['', 'amq.direct', 'amq.fanout', 'amq.headers', 'amq.match', 'amq.rabbitmq.log', 'amq.rabbitmq.trace', 'amq.topic', 'foo'], l('exchanges'))
00109         self.run_success(['delete', 'exchange', 'name=foo'])
00110 
00111     def test_queues(self):
00112         self.run_success(['declare', 'queue', 'name=foo'])
00113         self.assert_list(['foo'], l('queues'))
00114         self.run_success(['delete', 'queue', 'name=foo'])
00115 
00116     def test_bindings(self):
00117         self.run_success(['declare', 'queue', 'name=foo'])
00118         self.run_success(['declare', 'binding', 'source=amq.direct', 'destination=foo', 'destination_type=queue', 'routing_key=test'])
00119         self.assert_table([['', 'foo', 'queue', 'foo'], ['amq.direct', 'foo', 'queue', 'test']], ['list', 'bindings', 'source', 'destination', 'destination_type', 'routing_key'])
00120         self.run_success(['delete', 'queue', 'name=foo'])
00121 
00122     def test_publish(self):
00123         self.run_success(['declare', 'queue', 'name=test'])
00124         self.run_success(['publish', 'routing_key=test', 'payload=test_1'])
00125         self.run_success(['publish', 'routing_key=test', 'payload=test_2'])
00126         self.run_success(['publish', 'routing_key=test'], stdin='test_3')
00127         self.assert_table([exp_msg('test', 2, False, 'test_1')], ['get', 'queue=test', 'requeue=false'])
00128         self.assert_table([exp_msg('test', 1, False, 'test_2')], ['get', 'queue=test', 'requeue=true'])
00129         self.assert_table([exp_msg('test', 1, True,  'test_2')], ['get', 'queue=test', 'requeue=false'])
00130         self.assert_table([exp_msg('test', 0, False, 'test_3')], ['get', 'queue=test', 'requeue=false'])
00131         self.run_success(['publish', 'routing_key=test'], stdin='test_4')
00132         filename = '/tmp/rabbitmq-test/get.txt'
00133         self.run_success(['get', 'queue=test', 'requeue=false', 'payload_file=' + filename])
00134         with open(filename) as f:
00135             self.assertEqual('test_4', f.read())
00136         os.remove(filename)
00137         self.run_success(['delete', 'queue', 'name=test'])
00138 
00139     def test_ignore_vhost(self):
00140         self.run_success(['--vhost', '/', 'show', 'overview'])
00141         self.run_success(['--vhost', '/', 'list', 'users'])
00142         self.run_success(['--vhost', '/', 'list', 'vhosts'])
00143         self.run_success(['--vhost', '/', 'list', 'nodes'])
00144         self.run_success(['--vhost', '/', 'list', 'permissions'])
00145         self.run_success(['--vhost', '/', 'declare', 'user', 'name=foo', 'password=pass', 'tags='])
00146         self.run_success(['delete', 'user', 'name=foo'])
00147 
00148     # ---------------------------------------------------------------------------
00149 
00150     def run_success(self, args, **kwargs):
00151         self.assertEqual(0, run(args, **kwargs)[1])
00152 
00153     def run_fail(self, args):
00154         self.assertNotEqual(0, run(args)[1])
00155 
00156     def assert_output(self, expected, args):
00157         self.assertEqual(expected, run(args)[0])
00158 
00159     def assert_list(self, expected, args0):
00160         args = ['-f', 'tsv', '-q']
00161         args.extend(args0)
00162         self.assertEqual(expected, run(args)[0].splitlines())
00163 
00164     def assert_table(self, expected, args0):
00165         args = ['-f', 'tsv', '-q']
00166         args.extend(args0)
00167         self.assertEqual(expected, [l.split('\t') for l in run(args)[0].splitlines()])
00168 
00169 def run(args, stdin=None):
00170     path = os.path.normpath(os.path.join(os.getcwd(), sys.argv[0], '../../../bin/rabbitmqadmin'))
00171     cmdline = [path]
00172     cmdline.extend(args)
00173     proc = subprocess.Popen(cmdline, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
00174     (stdout, stderr) = proc.communicate(stdin)
00175     returncode = proc.returncode
00176     return (stdout, returncode)
00177 
00178 def l(thing):
00179     return ['list', thing, 'name']
00180 
00181 def exp_msg(key, count, redelivered, payload):
00182     # routing_key, exchange, message_count, payload, payload_bytes, payload_encoding, properties, redelivered
00183     return [key, '', str(count), payload, str(len(payload)), 'string', '', str(redelivered)]
00184 
00185 if __name__ == '__main__':
00186     print "\nrabbitmqadmin tests\n===================\n"
00187     suite = unittest.TestLoader().loadTestsFromTestCase(TestRabbitMQAdmin)
00188     unittest.TextTestRunner(verbosity=2).run(suite)