10
11
from amqplib import client_0_8 as amqp
11
12
from fixtures import EnvironmentVariableFixture
13
from testtools.content import Content
13
15
from lp.services.rabbit.testing.server import RabbitServer
14
16
from lp.testing import TestCase
19
# copy_content() and gather_details() have been copied from
20
# lp:~allenap/testtools/gather-details. If/when that branch lands the copies
21
# here should be removed.
25
def copy_content(content_object):
26
"""Make a copy of the given content object.
28
The content within `content_object` is iterated and saved. This is useful
29
when the source of the content is volatile, a log file in a temporary
30
directory for example.
32
:param content_object: A `content.Content` instance.
33
:return: A `content.Content` instance with the same mime-type as
34
`content_object` and a non-volatile copy of its content.
36
content_bytes = list(content_object.iter_bytes())
37
content_callback = lambda: content_bytes
38
return Content(content_object.content_type, content_callback)
41
def gather_details(source, target):
42
"""Merge the details from `source` into `target`.
44
:param source: A *detailed* object from which details will be gathered.
45
:param target: A *detailed* object into which details will be gathered.
47
source_details = source.getDetails()
48
target_details = target.getDetails()
49
for name, content_object in source_details.items():
51
disambiguator = itertools.count(1)
52
while new_name in target_details:
53
new_name = '%s-%d' % (name, next(disambiguator))
55
target.addDetail(name, copy_content(content_object))
17
58
class TestRabbitFixture(TestCase):
25
66
fixture = RabbitServer()
27
# Work around failures-in-setup-not-attaching-details (if they did we
28
# could use self.useFixture).
29
self.addCleanup(self._gather_details, fixture.getDetails)
34
"host": 'localhost:%s' % fixture.config.port,
35
"userid": "guest", "password": "guest",
36
"virtual_host": "/", "insist": False,
38
amqp.Connection(**connect_arguments).close()
40
log = fixture.runner.getDetails()["rabbit.log"]
41
# Which shouldn't blow up on iteration.
72
"host": 'localhost:%s' % fixture.config.port,
73
"userid": "guest", "password": "guest",
74
"virtual_host": "/", "insist": False,
76
amqp.Connection(**connect_arguments).close()
78
log = fixture.runner.getDetails()["rabbit.log"]
79
# Which shouldn't blow up on iteration.
82
# Work around failures-in-setup-not-attaching-details (if they did
83
# we could use self.useFixture).
84
gather_details(fixture.runner.environment, fixture.runner)
85
gather_details(fixture.runner, fixture)
86
gather_details(fixture.config, fixture)
87
gather_details(fixture, self)
44
90
# The daemon should be closed now.
45
91
self.assertRaises(socket.error, amqp.Connection, **connect_arguments)