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
|
# Copyright (C) 2006 by Canonical Ltd
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""Tests of the bzr serve command."""
import signal
from bzrlib.branch import Branch
from bzrlib.bzrdir import BzrDir
from bzrlib.tests import TestCaseWithTransport
from bzrlib.transport import smart
class DoesNotCloseStdOutClient(smart.SmartStreamClient):
"""A client that doesn't close stdout upon disconnect().
We wish to let stdout remain open so that we can see if the server writes
anything to stdout during its shutdown.
"""
def disconnect(self):
if self._connected:
self._connected = False
# The client's out is the server's in.
self._out.close()
class TestBzrServe(TestCaseWithTransport):
def test_bzr_serve_inet(self):
# Make a branch
self.make_branch('.')
# Serve that branch from the current directory
process = self.start_bzr_subprocess(['serve', '--inet'])
# Connect to the server
# We use this url because while this is no valid URL to connect to this
# server instance, the transport needs a URL.
client = DoesNotCloseStdOutClient(
lambda: (process.stdout, process.stdin))
transport = smart.SmartTransport('bzr://localhost/', client=client)
# We get a working branch
branch = BzrDir.open_from_transport(transport).open_branch()
branch.repository.get_revision_graph()
self.assertEqual(None, branch.last_revision())
# finish with the transport
del transport
# Disconnect the client forcefully JUST IN CASE because of __del__'s use
# in the smart module.
client.disconnect()
# Shutdown the server: the client should have disconnected cleanly and
# closed stdin, so the server process should shut itself down.
self.assertTrue(process.stdin.closed)
# Hide stdin from the subprocess module, so it won't fail to close it.
process.stdin = None
result = self.finish_bzr_subprocess(process, retcode=0)
self.assertEqual('', result[0])
self.assertEqual('', result[1])
def test_bzr_serve_port(self):
# Make a branch
self.make_branch('.')
# Serve that branch from the current directory
process = self.start_bzr_subprocess(['serve', '--port', 'localhost:0'],
skip_if_plan_to_signal=True)
port_line = process.stdout.readline()
prefix = 'listening on port: '
self.assertStartsWith(port_line, prefix)
port = int(port_line[len(prefix):])
# Connect to the server
branch = Branch.open('bzr://localhost:%d/' % port)
# We get a working branch
branch.repository.get_revision_graph()
self.assertEqual(None, branch.last_revision())
# Shutdown the server
result = self.finish_bzr_subprocess(process, retcode=3,
send_signal=signal.SIGINT)
self.assertEqual('', result[0])
self.assertEqual('bzr: interrupted\n', result[1])
def test_bzr_serve_no_args(self):
"""'bzr serve' with no arguments or options should not traceback."""
out, err = self.run_bzr_error(
['bzr serve requires one of --inet or --port'], 'serve')
|