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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
18
"""Tests of the bzr serve command."""
29
revision as _mod_revision,
31
from bzrlib.branch import Branch
32
from bzrlib.bzrdir import BzrDir
33
from bzrlib.errors import ParamikoNotPresent
34
from bzrlib.smart import medium
35
from bzrlib.tests import TestCaseWithTransport, TestSkipped
36
from bzrlib.transport import get_transport, remote
39
class TestBzrServe(TestCaseWithTransport):
41
def assertInetServerShutsdownCleanly(self, process):
42
"""Shutdown the server process looking for errors."""
43
# Shutdown the server: the server should shut down when it cannot read
46
# Hide stdin from the subprocess module, so it won't fail to close it.
48
result = self.finish_bzr_subprocess(process)
49
self.assertEqual('', result[0])
50
self.assertEqual('', result[1])
52
def assertServerFinishesCleanly(self, process):
53
"""Shutdown the bzr serve instance process looking for errors."""
55
result = self.finish_bzr_subprocess(process, retcode=3,
56
send_signal=signal.SIGINT)
57
self.assertEqual('', result[0])
58
self.assertEqual('bzr: interrupted\n', result[1])
60
def make_read_requests(self, branch):
61
"""Do some read only requests."""
64
branch.repository.all_revision_ids()
65
self.assertEqual(_mod_revision.NULL_REVISION,
66
_mod_revision.ensure_null(branch.last_revision()))
70
def start_server_inet(self, extra_options=()):
71
"""Start a bzr server subprocess using the --inet option.
73
:param extra_options: extra options to give the server.
74
:return: a tuple with the bzr process handle for passing to
75
finish_bzr_subprocess, a client for the server, and a transport.
77
# Serve from the current directory
78
process = self.start_bzr_subprocess(['serve', '--inet'])
80
# Connect to the server
81
# We use this url because while this is no valid URL to connect to this
82
# server instance, the transport needs a URL.
83
url = 'bzr://localhost/'
84
client_medium = medium.SmartSimplePipesClientMedium(
85
process.stdout, process.stdin, url)
86
transport = remote.RemoteTransport(url, medium=client_medium)
87
return process, transport
89
def start_server_port(self, extra_options=()):
90
"""Start a bzr server subprocess.
92
:param extra_options: extra options to give the server.
93
:return: a tuple with the bzr process handle for passing to
94
finish_bzr_subprocess, and the base url for the server.
96
# Serve from the current directory
97
args = ['serve', '--port', 'localhost:0']
98
args.extend(extra_options)
99
process = self.start_bzr_subprocess(args, skip_if_plan_to_signal=True)
100
port_line = process.stderr.readline()
101
prefix = 'listening on port: '
102
self.assertStartsWith(port_line, prefix)
103
port = int(port_line[len(prefix):])
104
return process,'bzr://localhost:%d/' % port
106
def test_bzr_serve_inet_readonly(self):
107
"""bzr server should provide a read only filesystem by default."""
108
process, transport = self.start_server_inet()
109
self.assertRaises(errors.TransportNotPossible, transport.mkdir, 'adir')
110
self.assertInetServerShutsdownCleanly(process)
112
def test_bzr_serve_inet_readwrite(self):
114
self.make_branch('.')
116
process, transport = self.start_server_inet(['--allow-writes'])
118
# We get a working branch
119
branch = BzrDir.open_from_transport(transport).open_branch()
120
self.make_read_requests(branch)
121
self.assertInetServerShutsdownCleanly(process)
123
def test_bzr_serve_port_readonly(self):
124
"""bzr server should provide a read only filesystem by default."""
125
process, url = self.start_server_port()
126
transport = get_transport(url)
127
self.assertRaises(errors.TransportNotPossible, transport.mkdir, 'adir')
128
self.assertServerFinishesCleanly(process)
130
def test_bzr_serve_port_readwrite(self):
132
self.make_branch('.')
134
process, url = self.start_server_port(['--allow-writes'])
136
# Connect to the server
137
branch = Branch.open(url)
138
self.make_read_requests(branch)
139
self.assertServerFinishesCleanly(process)
141
def test_bzr_serve_supports_protocol(self):
143
self.make_branch('.')
145
process, url = self.start_server_port(['--allow-writes',
148
# Connect to the server
149
branch = Branch.open(url)
150
self.make_read_requests(branch)
151
self.assertServerFinishesCleanly(process)
153
def test_bzr_connect_to_bzr_ssh(self):
154
"""User acceptance that get_transport of a bzr+ssh:// behaves correctly.
156
bzr+ssh:// should cause bzr to run a remote bzr smart server over SSH.
159
from bzrlib.transport.sftp import SFTPServer
160
except ParamikoNotPresent:
161
raise TestSkipped('Paramiko not installed')
162
from bzrlib.tests.stub_sftp import StubServer
165
self.make_branch('a_branch')
167
# Start an SSH server
168
self.command_executed = []
169
# XXX: This is horrible -- we define a really dumb SSH server that
170
# executes commands, and manage the hooking up of stdin/out/err to the
171
# SSH channel ourselves. Surely this has already been implemented
173
class StubSSHServer(StubServer):
177
def check_channel_exec_request(self, channel, command):
178
self.test.command_executed.append(command)
179
proc = subprocess.Popen(
180
command, shell=True, stdin=subprocess.PIPE,
181
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
183
# XXX: horribly inefficient, not to mention ugly.
184
# Start a thread for each of stdin/out/err, and relay bytes from
185
# the subprocess to channel and vice versa.
186
def ferry_bytes(read, write, close):
195
(channel.recv, proc.stdin.write, proc.stdin.close),
196
(proc.stdout.read, channel.sendall, channel.close),
197
(proc.stderr.read, channel.sendall_stderr, channel.close)]
198
for read, write, close in file_functions:
199
t = threading.Thread(
200
target=ferry_bytes, args=(read, write, close))
205
ssh_server = SFTPServer(StubSSHServer)
206
# XXX: We *don't* want to override the default SSH vendor, so we set
207
# _vendor to what _get_ssh_vendor returns.
209
self.addCleanup(ssh_server.tearDown)
210
port = ssh_server._listener.port
212
# Access the branch via a bzr+ssh URL. The BZR_REMOTE_PATH environment
213
# variable is used to tell bzr what command to run on the remote end.
214
path_to_branch = osutils.abspath('a_branch')
216
orig_bzr_remote_path = os.environ.get('BZR_REMOTE_PATH')
217
bzr_remote_path = self.get_bzr_path()
218
if sys.platform == 'win32':
219
bzr_remote_path = sys.executable + ' ' + self.get_bzr_path()
220
os.environ['BZR_REMOTE_PATH'] = bzr_remote_path
222
if sys.platform == 'win32':
223
path_to_branch = os.path.splitdrive(path_to_branch)[1]
224
branch = Branch.open(
225
'bzr+ssh://fred:secret@localhost:%d%s' % (port, path_to_branch))
226
self.make_read_requests(branch)
227
# Check we can perform write operations
228
branch.bzrdir.root_transport.mkdir('foo')
230
# Restore the BZR_REMOTE_PATH environment variable back to its
232
if orig_bzr_remote_path is None:
233
del os.environ['BZR_REMOTE_PATH']
235
os.environ['BZR_REMOTE_PATH'] = orig_bzr_remote_path
238
['%s serve --inet --directory=/ --allow-writes'
240
self.command_executed)
added
removed
bzrlib/tests/blackbox/test_serve.py
