1
# Copyright (C) 2006 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, remote
31
from bzrlib.smart import medium
34
class TestBzrServe(TestCaseWithTransport):
36
def assertInetServerShutsdownCleanly(self, process):
37
"""Shutdown the server process looking for errors."""
38
# Shutdown the server: the server should shut down when it cannot read
41
# Hide stdin from the subprocess module, so it won't fail to close it.
43
result = self.finish_bzr_subprocess(process, retcode=0)
44
self.assertEqual('', result[0])
45
self.assertEqual('', result[1])
47
def assertServerFinishesCleanly(self, process):
48
"""Shutdown the bzr serve instance process looking for errors."""
50
result = self.finish_bzr_subprocess(process, retcode=3,
51
send_signal=signal.SIGINT)
52
self.assertEqual('', result[0])
53
self.assertEqual('bzr: interrupted\n', result[1])
55
def start_server_inet(self, extra_options=()):
56
"""Start a bzr server subprocess using the --inet option.
58
:param extra_options: extra options to give the server.
59
:return: a tuple with the bzr process handle for passing to
60
finish_bzr_subprocess, a client for the server, and a transport.
62
# Serve from the current directory
63
process = self.start_bzr_subprocess(['serve', '--inet'])
65
# Connect to the server
66
# We use this url because while this is no valid URL to connect to this
67
# server instance, the transport needs a URL.
68
client_medium = medium.SmartSimplePipesClientMedium(
69
process.stdout, process.stdin)
70
transport = remote.RemoteTransport(
71
'bzr://localhost/', medium=client_medium)
72
return process, transport
74
def start_server_port(self, extra_options=()):
75
"""Start a bzr server subprocess.
77
:param extra_options: extra options to give the server.
78
:return: a tuple with the bzr process handle for passing to
79
finish_bzr_subprocess, and the base url for the server.
81
# Serve from the current directory
82
args = ['serve', '--port', 'localhost:0']
83
args.extend(extra_options)
84
process = self.start_bzr_subprocess(args, skip_if_plan_to_signal=True)
85
port_line = process.stdout.readline()
86
prefix = 'listening on port: '
87
self.assertStartsWith(port_line, prefix)
88
port = int(port_line[len(prefix):])
89
return process,'bzr://localhost:%d/' % port
91
def test_bzr_serve_inet_readonly(self):
92
"""bzr server should provide a read only filesystem by default."""
93
process, transport = self.start_server_inet()
94
self.assertRaises(errors.TransportNotPossible, transport.mkdir, 'adir')
95
self.assertInetServerShutsdownCleanly(process)
97
def test_bzr_serve_inet_readwrite(self):
101
process, transport = self.start_server_inet(['--allow-writes'])
103
# We get a working branch
104
branch = BzrDir.open_from_transport(transport).open_branch()
105
branch.repository.get_revision_graph()
106
self.assertEqual(None, branch.last_revision())
107
self.assertInetServerShutsdownCleanly(process)
109
def test_bzr_serve_port_readonly(self):
110
"""bzr server should provide a read only filesystem by default."""
111
process, url = self.start_server_port()
112
transport = get_transport(url)
113
self.assertRaises(errors.TransportNotPossible, transport.mkdir, 'adir')
114
self.assertServerFinishesCleanly(process)
116
def test_bzr_serve_port_readwrite(self):
118
self.make_branch('.')
120
process, url = self.start_server_port(['--allow-writes'])
122
# Connect to the server
123
branch = Branch.open(url)
125
# We get a working branch
126
branch.repository.get_revision_graph()
127
self.assertEqual(None, branch.last_revision())
129
self.assertServerFinishesCleanly(process)
131
def test_bzr_serve_no_args(self):
132
"""'bzr serve' with no arguments or options should not traceback."""
133
out, err = self.run_bzr_error(
134
['bzr serve requires one of --inet or --port'], 'serve')
136
def test_bzr_connect_to_bzr_ssh(self):
137
"""User acceptance that get_transport of a bzr+ssh:// behaves correctly.
139
bzr+ssh:// should cause bzr to run a remote bzr smart server over SSH.
142
from bzrlib.transport.sftp import SFTPServer
143
except ParamikoNotPresent:
144
raise TestSkipped('Paramiko not installed')
145
from bzrlib.tests.stub_sftp import StubServer
148
self.make_branch('a_branch')
150
# Start an SSH server
151
self.command_executed = []
152
# XXX: This is horrible -- we define a really dumb SSH server that
153
# executes commands, and manage the hooking up of stdin/out/err to the
154
# SSH channel ourselves. Surely this has already been implemented
156
class StubSSHServer(StubServer):
160
def check_channel_exec_request(self, channel, command):
161
self.test.command_executed.append(command)
162
proc = subprocess.Popen(
163
command, shell=True, stdin=subprocess.PIPE,
164
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
166
# XXX: horribly inefficient, not to mention ugly.
167
# Start a thread for each of stdin/out/err, and relay bytes from
168
# the subprocess to channel and vice versa.
169
def ferry_bytes(read, write, close):
178
(channel.recv, proc.stdin.write, proc.stdin.close),
179
(proc.stdout.read, channel.sendall, channel.close),
180
(proc.stderr.read, channel.sendall_stderr, channel.close)]
181
for read, write, close in file_functions:
182
t = threading.Thread(
183
target=ferry_bytes, args=(read, write, close))
188
ssh_server = SFTPServer(StubSSHServer)
189
# XXX: We *don't* want to override the default SSH vendor, so we set
190
# _vendor to what _get_ssh_vendor returns.
192
self.addCleanup(ssh_server.tearDown)
193
port = ssh_server._listener.port
195
# Access the branch via a bzr+ssh URL. The BZR_REMOTE_PATH environment
196
# variable is used to tell bzr what command to run on the remote end.
197
path_to_branch = os.path.abspath('a_branch')
199
orig_bzr_remote_path = os.environ.get('BZR_REMOTE_PATH')
200
os.environ['BZR_REMOTE_PATH'] = self.get_bzr_path()
202
branch = Branch.open(
203
'bzr+ssh://fred:secret@localhost:%d%s' % (port, path_to_branch))
205
branch.repository.get_revision_graph()
206
self.assertEqual(None, branch.last_revision())
207
# Check we can perform write operations
208
branch.bzrdir.root_transport.mkdir('foo')
210
# Restore the BZR_REMOTE_PATH environment variable back to its
212
if orig_bzr_remote_path is None:
213
del os.environ['BZR_REMOTE_PATH']
215
os.environ['BZR_REMOTE_PATH'] = orig_bzr_remote_path
218
['%s serve --inet --directory=/ --allow-writes'
219
% self.get_bzr_path()],
220
self.command_executed)
added
removed
bzrlib/tests/blackbox/test_serve.py
