/usr/lib/python2.7/dist-packages/celery/tests/bin/test_amqp.py is in python-celery 3.1.20-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | from __future__ import absolute_import
from celery.bin.amqp import (
AMQPAdmin,
AMQShell,
dump_message,
amqp,
main,
)
from celery.tests.case import AppCase, Mock, WhateverIO, patch
class test_AMQShell(AppCase):
def setup(self):
self.fh = WhateverIO()
self.adm = self.create_adm()
self.shell = AMQShell(connect=self.adm.connect, out=self.fh)
def create_adm(self, *args, **kwargs):
return AMQPAdmin(app=self.app, out=self.fh, *args, **kwargs)
def test_queue_declare(self):
self.shell.onecmd('queue.declare foo')
self.assertIn('ok', self.fh.getvalue())
def test_missing_command(self):
self.shell.onecmd('foo foo')
self.assertIn('unknown syntax', self.fh.getvalue())
def RV(self):
raise Exception(self.fh.getvalue())
def test_spec_format_response(self):
spec = self.shell.amqp['exchange.declare']
self.assertEqual(spec.format_response(None), 'ok.')
self.assertEqual(spec.format_response('NO'), 'NO')
def test_missing_namespace(self):
self.shell.onecmd('ns.cmd arg')
self.assertIn('unknown syntax', self.fh.getvalue())
def test_help(self):
self.shell.onecmd('help')
self.assertIn('Example:', self.fh.getvalue())
def test_help_command(self):
self.shell.onecmd('help queue.declare')
self.assertIn('passive:no', self.fh.getvalue())
def test_help_unknown_command(self):
self.shell.onecmd('help foo.baz')
self.assertIn('unknown syntax', self.fh.getvalue())
def test_onecmd_error(self):
self.shell.dispatch = Mock()
self.shell.dispatch.side_effect = MemoryError()
self.shell.say = Mock()
self.assertFalse(self.shell.needs_reconnect)
self.shell.onecmd('hello')
self.assertTrue(self.shell.say.called)
self.assertTrue(self.shell.needs_reconnect)
def test_exit(self):
with self.assertRaises(SystemExit):
self.shell.onecmd('exit')
self.assertIn("don't leave!", self.fh.getvalue())
def test_note_silent(self):
self.shell.silent = True
self.shell.note('foo bar')
self.assertNotIn('foo bar', self.fh.getvalue())
def test_reconnect(self):
self.shell.onecmd('queue.declare foo')
self.shell.needs_reconnect = True
self.shell.onecmd('queue.delete foo')
def test_completenames(self):
self.assertEqual(
self.shell.completenames('queue.dec'),
['queue.declare'],
)
self.assertEqual(
sorted(self.shell.completenames('declare')),
sorted(['queue.declare', 'exchange.declare']),
)
def test_empty_line(self):
self.shell.emptyline = Mock()
self.shell.default = Mock()
self.shell.onecmd('')
self.shell.emptyline.assert_called_with()
self.shell.onecmd('foo')
self.shell.default.assert_called_with('foo')
def test_respond(self):
self.shell.respond({'foo': 'bar'})
self.assertIn('foo', self.fh.getvalue())
def test_prompt(self):
self.assertTrue(self.shell.prompt)
def test_no_returns(self):
self.shell.onecmd('queue.declare foo')
self.shell.onecmd('exchange.declare bar direct yes')
self.shell.onecmd('queue.bind foo bar baz')
self.shell.onecmd('basic.ack 1')
def test_dump_message(self):
m = Mock()
m.body = 'the quick brown fox'
m.properties = {'a': 1}
m.delivery_info = {'exchange': 'bar'}
self.assertTrue(dump_message(m))
def test_dump_message_no_message(self):
self.assertIn('No messages in queue', dump_message(None))
def test_note(self):
self.adm.silent = True
self.adm.note('FOO')
self.assertNotIn('FOO', self.fh.getvalue())
def test_run(self):
a = self.create_adm('queue.declare foo')
a.run()
self.assertIn('ok', self.fh.getvalue())
def test_run_loop(self):
a = self.create_adm()
a.Shell = Mock()
shell = a.Shell.return_value = Mock()
shell.cmdloop = Mock()
a.run()
shell.cmdloop.assert_called_with()
shell.cmdloop.side_effect = KeyboardInterrupt()
a.run()
self.assertIn('bibi', self.fh.getvalue())
@patch('celery.bin.amqp.amqp')
def test_main(self, Command):
c = Command.return_value = Mock()
main()
c.execute_from_commandline.assert_called_with()
@patch('celery.bin.amqp.AMQPAdmin')
def test_command(self, cls):
x = amqp(app=self.app)
x.run()
self.assertIs(cls.call_args[1]['app'], self.app)
|