1
# Copyright (C) 2006 by Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18
"""Tests of the bzr serve command."""
25
from bzrlib import errors
26
from bzrlib.branch import Branch
27
from bzrlib.bzrdir import BzrDir
28
from bzrlib.errors import ParamikoNotPresent
29
from bzrlib.tests import TestCaseWithTransport, TestSkipped
30
from bzrlib.transport import get_transport, smart
33
class DoesNotCloseStdOutClient(smart.SmartStreamClient):
34
"""A client that doesn't close stdout upon disconnect().
36
We wish to let stdout remain open so that we can see if the server writes
37
anything to stdout during its shutdown.
42
self._connected = False
43
# The client's out is the server's in.
47
class TestBzrServe(TestCaseWithTransport):
49
def assertInetServerShutsdownCleanly(self, client, process):
50
"""Shutdown the server process looking for errors."""
51
# Disconnect the client forcefully JUST IN CASE because of __del__'s use
52
# in the smart module.
55
# Shutdown the server: the client should have disconnected cleanly and
56
# closed stdin, so the server process should shut itself down.
57
self.assertTrue(process.stdin.closed)
58
# Hide stdin from the subprocess module, so it won't fail to close it.
60
result = self.finish_bzr_subprocess(process, retcode=0)
61
self.assertEqual('', result[0])
62
self.assertEqual('', result[1])
64
def assertServerFinishesCleanly(self, process):
65
"""Shutdown the bzr serve instance process looking for errors."""
67
result = self.finish_bzr_subprocess(process, retcode=3,
68
send_signal=signal.SIGINT)
69
self.assertEqual('', result[0])
70
self.assertEqual('bzr: interrupted\n', result[1])
72
def start_server_inet(self, extra_options=()):
73
"""Start a bzr server subprocess using the --inet option.
75
:param extra_options: extra options to give the server.
76
:return: a tuple with the bzr process handle for passing to
77
finish_bzr_subprocess, a client for the server, and a transport.
79
# Serve from the current directory
80
process = self.start_bzr_subprocess(['serve', '--inet'])
82
# Connect to the server
83
# We use this url because while this is no valid URL to connect to this
84
# server instance, the transport needs a URL.
85
client = DoesNotCloseStdOutClient(
86
lambda: (process.stdout, process.stdin))
87
transport = smart.SmartTransport('bzr://localhost/', client=client)
88
return process, client, transport
90
def start_server_port(self, extra_options=()):
91
"""Start a bzr server subprocess.
93
:param extra_options: extra options to give the server.
94
:return: a tuple with the bzr process handle for passing to
95
finish_bzr_subprocess, and the base url for the server.
97
# Serve from the current directory
98
args = ['serve', '--port', 'localhost:0']
99
args.extend(extra_options)
100
process = self.start_bzr_subprocess(args, skip_if_plan_to_signal=True)
101
port_line = process.stdout.readline()
102
prefix = 'listening on port: '
103
self.assertStartsWith(port_line, prefix)
104
port = int(port_line[len(prefix):])
105
return process,'bzr://localhost:%d/' % port
107
def test_bzr_serve_inet_readonly(self):
108
"""bzr server should provide a read only filesystem by default."""
109
process, client, transport = self.start_server_inet()
110
self.assertRaises(errors.TransportNotPossible, transport.mkdir, 'adir')
111
# finish with the transport
113
self.assertInetServerShutsdownCleanly(client, process)
115
def test_bzr_serve_inet_readwrite(self):
117
self.make_branch('.')
119
process, client, transport = self.start_server_inet(['--allow-writes'])
121
# We get a working branch
122
branch = BzrDir.open_from_transport(transport).open_branch()
123
branch.repository.get_revision_graph()
124
self.assertEqual(None, branch.last_revision())
126
# finish with the transport
129
self.assertInetServerShutsdownCleanly(client, process)
131
def test_bzr_serve_port_readonly(self):
132
"""bzr server should provide a read only filesystem by default."""
133
process, url = self.start_server_port()
134
transport = get_transport(url)
135
self.assertRaises(errors.TransportNotPossible, transport.mkdir, 'adir')
136
self.assertServerFinishesCleanly(process)
138
def test_bzr_serve_port_readwrite(self):
140
self.make_branch('.')
142
process, url = self.start_server_port(['--allow-writes'])
144
# Connect to the server
145
branch = Branch.open(url)
147
# We get a working branch
148
branch.repository.get_revision_graph()
149
self.assertEqual(None, branch.last_revision())
151
self.assertServerFinishesCleanly(process)
153
def test_bzr_serve_no_args(self):
154
"""'bzr serve' with no arguments or options should not traceback."""
155
out, err = self.run_bzr_error(
156
['bzr serve requires one of --inet or --port'], 'serve')
158
def test_bzr_connect_to_bzr_ssh(self):
159
"""User acceptance that get_transport of a bzr+ssh:// behaves correctly.
161
bzr+ssh:// should cause bzr to run a remote bzr smart server over SSH.
164
from bzrlib.transport.sftp import SFTPServer
165
except ParamikoNotPresent:
166
raise TestSkipped('Paramiko not installed')
168
from bzrlib.tests.stub_sftp import StubServer
171
self.make_branch('a_branch')
173
# Start an SSH server
174
# XXX: This is horrible -- we define a really dumb SSH server that
175
# executes commands, and manage the hooking up of stdin/out/err to the
176
# SSH channel ourselves. Surely this has already been implemented
178
class StubSSHServer(StubServer):
182
def check_channel_exec_request(self, channel, command):
183
self.test.command_executed = command
184
proc = subprocess.Popen(
185
command, shell=True, stdin=subprocess.PIPE,
186
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
188
# XXX: horribly inefficient, not to mention ugly.
190
# copy bytes from the channel to the subprocess's stdin
192
bytes = channel.recv(1)
196
proc.stdin.write(bytes)
198
threading.Thread(target=do_stdin).start()
200
def ferry_bytes(pipe):
206
channel.sendall(bytes)
207
threading.Thread(target=ferry_bytes, args=(proc.stdout,)).start()
208
threading.Thread(target=ferry_bytes, args=(proc.stderr,)).start()
212
ssh_server = SFTPServer(StubSSHServer)
213
# XXX: We *don't* want to override the default SSH vendor, so we set
214
# _vendor to what _get_ssh_vendor returns.
216
self.addCleanup(ssh_server.tearDown)
217
port = ssh_server._listener.port
219
# Access the branch via a bzr+ssh URL. The BZR_REMOTE_PATH environment
220
# variable is used to tell bzr what command to run on the remote end.
221
path_to_branch = os.path.abspath('a_branch')
223
orig_bzr_remote_path = os.environ.get('BZR_REMOTE_PATH')
224
os.environ['BZR_REMOTE_PATH'] = self.get_bzr_path()
226
branch = Branch.open(
227
'bzr+ssh://fred:secret@localhost:%d%s' % (port, path_to_branch))
229
branch.repository.get_revision_graph()
230
self.assertEqual(None, branch.last_revision())
231
# Check we can perform write operations
232
branch.bzrdir.root_transport.mkdir('foo')
234
# Restore the BZR_REMOTE_PATH environment variable back to its
236
if orig_bzr_remote_path is None:
237
del os.environ['BZR_REMOTE_PATH']
239
os.environ['BZR_REMOTE_PATH'] = orig_bzr_remote_path
242
'%s serve --inet --directory=/ --allow-writes'
243
% self.get_bzr_path(),
244
self.command_executed)
added
removed
bzrlib/tests/blackbox/test_serve.py
