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
|
# Copyright 2009-2010 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Tests for the lp-serve plugin."""
__metaclass__ = type
from bzrlib import errors
from bzrlib.plugins.lpserve.test_lpserve import TestCaseWithSubprocess
from bzrlib.smart import medium
from bzrlib.transport import remote
from lp.testing.fixture import CaptureOops
from lp.testing.layers import LaunchpadLayer
class TestLaunchpadServe(TestCaseWithSubprocess):
"""Tests for the lp-serve plugin.
Most of the helper methods here are copied from bzrlib.tests and
bzrlib.tests.blackbox.test_serve in bzr.dev r4445. They have since been
modified for the Launchpad environment.
"""
# The oops tests need rabbit available
layer = LaunchpadLayer
def assertFinishedCleanly(self, result):
"""Assert that a server process finished cleanly."""
self.assertEqual((0, '', ''), tuple(result))
def finish_lpserve_subprocess(self, process):
"""Shut down the server process.
:return: A tuple of (retcode, stdout, stderr).
"""
# Shutdown the server: the server should shut down when it cannot read
# from stdin anymore.
process.stdin.close()
# Hide stdin from the subprocess module, so it won't fail to close it.
process.stdin = None
# Finish the process without asserting anything about the return code.
# We'll leave that to assertFinishedCleanly.
stdout_and_stderr = self.finish_bzr_subprocess(process, retcode=None)
return (
process.returncode,
stdout_and_stderr[0],
stdout_and_stderr[1],
)
def start_server_inet(self, user_id=None):
"""Start an lp-serve server subprocess.
Copied from `bzrlib.tests.test_serve`.
:param user_id: The database id of the user to run as. If not
provided, defaults to 1.
:return: a tuple with the bzr process handle for passing to
finish_lpserve_subprocess, a client for the server, and a
transport.
"""
# Serve from the current directory
if user_id is None:
user_id = 1
process = self.start_bzr_subprocess(
['lp-serve', '--inet', str(user_id)])
# Connect to the server.
# The transport needs a URL, but we don't have one for our server, so
# we're just going to give it this nearly-arbitrary-yet-well-formed
# one.
url = 'bzr://localhost/'
client_medium = medium.SmartSimplePipesClientMedium(
process.stdout, process.stdin, url)
transport = remote.RemoteTransport(url, medium=client_medium)
return process, transport
def test_successful_start_then_stop(self):
# We can start and stop the lpserve process.
process, transport = self.start_server_inet()
result = self.finish_lpserve_subprocess(process)
self.assertFinishedCleanly(result)
def test_successful_start_then_stop_logs_no_oops(self):
# Starting and stopping the lp-serve process leaves no OOPS.
capture = self.useFixture(CaptureOops())
process, transport = self.start_server_inet()
self.finish_lpserve_subprocess(process)
capture.sync()
self.assertEqual([], capture.oopses)
def test_unexpected_error_logs_oops(self):
# If an unexpected error is raised in the plugin, then an OOPS is
# recorded.
capture = self.useFixture(CaptureOops())
process, transport = self.start_server_inet()
# This will trigger an error, because the XML-RPC server is not
# running, and any filesystem access tries to get at the XML-RPC
# server. If this *doesn'* raise, then the test is no longer valid and
# we need a new way of triggering errors in the smart server.
self.assertRaises(
errors.UnknownErrorFromSmartServer,
transport.list_dir, 'foo/bar/baz')
result = self.finish_lpserve_subprocess(process)
self.assertFinishedCleanly(result)
capture.sync()
self.assertEqual(1, len(capture.oopses))
self.assertEqual(
'[Errno 111] Connection refused', capture.oopses[0]['value'],
capture.oopses)
def test_suite():
from bzrlib import tests
from bzrlib.plugins import lpserve
loader = tests.TestLoader()
suite = loader.loadTestsFromName(__name__)
suite = lpserve.load_tests(suite, lpserve, loader)
return suite
|