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."""
30
from bzrlib.branch import Branch
31
from bzrlib.bzrdir import BzrDir
32
from bzrlib.errors import ParamikoNotPresent
33
from bzrlib.smart import medium
34
from bzrlib.tests import TestCaseWithTransport, TestSkipped
35
from bzrlib.transport import get_transport, remote
38
class TestBzrServe(TestCaseWithTransport):
40
def assertInetServerShutsdownCleanly(self, process):
41
"""Shutdown the server process looking for errors."""
42
# Shutdown the server: the server should shut down when it cannot read
45
# Hide stdin from the subprocess module, so it won't fail to close it.
47
result = self.finish_bzr_subprocess(process, retcode=0)
48
self.assertEqual('', result[0])
49
self.assertEqual('', result[1])
51
def assertServerFinishesCleanly(self, process):
52
"""Shutdown the bzr serve instance process looking for errors."""
54
result = self.finish_bzr_subprocess(process, retcode=3,
55
send_signal=signal.SIGINT)
56
self.assertEqual('', result[0])
57
self.assertEqual('bzr: interrupted\n', result[1])
59
def start_server_inet(self, extra_options=()):
60
"""Start a bzr server subprocess using the --inet option.
62
:param extra_options: extra options to give the server.
63
:return: a tuple with the bzr process handle for passing to
64
finish_bzr_subprocess, a client for the server, and a transport.
66
# Serve from the current directory
67
process = self.start_bzr_subprocess(['serve', '--inet'])
69
# Connect to the server
70
# We use this url because while this is no valid URL to connect to this
71
# server instance, the transport needs a URL.
72
client_medium = medium.SmartSimplePipesClientMedium(
73
process.stdout, process.stdin)
74
transport = remote.SmartTransport(
75
'bzr://localhost/', medium=client_medium)
76
return process, transport
78
def start_server_port(self, extra_options=()):
79
"""Start a bzr server subprocess.
81
:param extra_options: extra options to give the server.
82
:return: a tuple with the bzr process handle for passing to
83
finish_bzr_subprocess, and the base url for the server.
85
# Serve from the current directory
86
args = ['serve', '--port', 'localhost:0']
87
args.extend(extra_options)
88
process = self.start_bzr_subprocess(args, skip_if_plan_to_signal=True)
89
port_line = process.stdout.readline()
90
prefix = 'listening on port: '
91
self.assertStartsWith(port_line, prefix)
92
port = int(port_line[len(prefix):])
93
return process,'bzr://localhost:%d/' % port
95
def test_bzr_serve_inet_readonly(self):
96
"""bzr server should provide a read only filesystem by default."""
97
process, transport = self.start_server_inet()
98
self.assertRaises(errors.TransportNotPossible, transport.mkdir, 'adir')
99
self.assertInetServerShutsdownCleanly(process)
101
def test_bzr_serve_inet_readwrite(self):
103
self.make_branch('.')
105
process, transport = self.start_server_inet(['--allow-writes'])
107
# We get a working branch
108
branch = BzrDir.open_from_transport(transport).open_branch()
109
branch.repository.get_revision_graph()
110
self.assertEqual(None, branch.last_revision())
111
self.assertInetServerShutsdownCleanly(process)
113
def test_bzr_serve_port_readonly(self):
114
"""bzr server should provide a read only filesystem by default."""
115
process, url = self.start_server_port()
116
transport = get_transport(url)
117
self.assertRaises(errors.TransportNotPossible, transport.mkdir, 'adir')
118
self.assertServerFinishesCleanly(process)
120
def test_bzr_serve_port_readwrite(self):
122
self.make_branch('.')
124
process, url = self.start_server_port(['--allow-writes'])
126
# Connect to the server
127
branch = Branch.open(url)
129
# We get a working branch
130
branch.repository.get_revision_graph()
131
self.assertEqual(None, branch.last_revision())
133
self.assertServerFinishesCleanly(process)
135
def test_bzr_connect_to_bzr_ssh(self):
136
"""User acceptance that get_transport of a bzr+ssh:// behaves correctly.
138
bzr+ssh:// should cause bzr to run a remote bzr smart server over SSH.
141
from bzrlib.transport.sftp import SFTPServer
142
except ParamikoNotPresent:
143
raise TestSkipped('Paramiko not installed')
144
from bzrlib.tests.stub_sftp import StubServer
147
self.make_branch('a_branch')
149
# Start an SSH server
150
self.command_executed = []
151
# XXX: This is horrible -- we define a really dumb SSH server that
152
# executes commands, and manage the hooking up of stdin/out/err to the
153
# SSH channel ourselves. Surely this has already been implemented
155
class StubSSHServer(StubServer):
159
def check_channel_exec_request(self, channel, command):
160
self.test.command_executed.append(command)
161
proc = subprocess.Popen(
162
command, shell=True, stdin=subprocess.PIPE,
163
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
165
# XXX: horribly inefficient, not to mention ugly.
166
# Start a thread for each of stdin/out/err, and relay bytes from
167
# the subprocess to channel and vice versa.
168
def ferry_bytes(read, write, close):
177
(channel.recv, proc.stdin.write, proc.stdin.close),
178
(proc.stdout.read, channel.sendall, channel.close),
179
(proc.stderr.read, channel.sendall_stderr, channel.close)]
180
for read, write, close in file_functions:
181
t = threading.Thread(
182
target=ferry_bytes, args=(read, write, close))
187
ssh_server = SFTPServer(StubSSHServer)
188
# XXX: We *don't* want to override the default SSH vendor, so we set
189
# _vendor to what _get_ssh_vendor returns.
191
self.addCleanup(ssh_server.tearDown)
192
port = ssh_server._listener.port
194
# Access the branch via a bzr+ssh URL. The BZR_REMOTE_PATH environment
195
# variable is used to tell bzr what command to run on the remote end.
196
path_to_branch = osutils.abspath('a_branch')
198
orig_bzr_remote_path = os.environ.get('BZR_REMOTE_PATH')
199
bzr_remote_path = self.get_bzr_path()
200
if sys.platform == 'win32':
201
bzr_remote_path = sys.executable + ' ' + self.get_bzr_path()
202
os.environ['BZR_REMOTE_PATH'] = bzr_remote_path
204
if sys.platform == 'win32':
205
path_to_branch = os.path.splitdrive(path_to_branch)[1]
206
branch = Branch.open(
207
'bzr+ssh://fred:secret@localhost:%d%s' % (port, path_to_branch))
209
branch.repository.get_revision_graph()
210
self.assertEqual(None, branch.last_revision())
211
# Check we can perform write operations
212
branch.bzrdir.root_transport.mkdir('foo')
214
# Restore the BZR_REMOTE_PATH environment variable back to its
216
if orig_bzr_remote_path is None:
217
del os.environ['BZR_REMOTE_PATH']
219
os.environ['BZR_REMOTE_PATH'] = orig_bzr_remote_path
222
['%s serve --inet --directory=/ --allow-writes'
224
self.command_executed)
added
removed
bzrlib/tests/blackbox/test_serve.py
