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.branch import Branch
26
from bzrlib.bzrdir import BzrDir
27
from bzrlib.errors import ParamikoNotPresent
28
from bzrlib.tests import TestCaseWithTransport, TestSkipped
29
from bzrlib.transport import smart
32
class DoesNotCloseStdOutClient(smart.SmartStreamClient):
33
"""A client that doesn't close stdout upon disconnect().
35
We wish to let stdout remain open so that we can see if the server writes
36
anything to stdout during its shutdown.
41
self._connected = False
42
# The client's out is the server's in.
46
class TestBzrServe(TestCaseWithTransport):
48
def test_bzr_serve_inet(self):
52
# Serve that branch from the current directory
53
process = self.start_bzr_subprocess(['serve', '--inet'])
55
# Connect to the server
56
# We use this url because while this is no valid URL to connect to this
57
# server instance, the transport needs a URL.
58
client = DoesNotCloseStdOutClient(
59
lambda: (process.stdout, process.stdin))
60
transport = smart.SmartTransport('bzr://localhost/', client=client)
62
# We get a working branch
63
branch = BzrDir.open_from_transport(transport).open_branch()
64
branch.repository.get_revision_graph()
65
self.assertEqual(None, branch.last_revision())
67
# finish with the transport
69
# Disconnect the client forcefully JUST IN CASE because of __del__'s use
70
# in the smart module.
73
# Shutdown the server: the client should have disconnected cleanly and
74
# closed stdin, so the server process should shut itself down.
75
self.assertTrue(process.stdin.closed)
76
# Hide stdin from the subprocess module, so it won't fail to close it.
78
result = self.finish_bzr_subprocess(process, retcode=0)
79
self.assertEqual('', result[0])
80
self.assertEqual('', result[1])
82
def test_bzr_serve_port(self):
86
# Serve that branch from the current directory
87
process = self.start_bzr_subprocess(['serve', '--port', 'localhost:0'],
88
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):])
94
# Connect to the server
95
branch = Branch.open('bzr://localhost:%d/' % port)
97
# We get a working branch
98
branch.repository.get_revision_graph()
99
self.assertEqual(None, branch.last_revision())
101
# Shutdown the server
102
result = self.finish_bzr_subprocess(process, retcode=3,
103
send_signal=signal.SIGINT)
104
self.assertEqual('', result[0])
105
self.assertEqual('bzr: interrupted\n', result[1])
107
def test_bzr_serve_no_args(self):
108
"""'bzr serve' with no arguments or options should not traceback."""
109
out, err = self.run_bzr_error(
110
['bzr serve requires one of --inet or --port'], 'serve')
112
def test_bzr_connect_to_bzr_ssh(self):
113
"""User acceptance that get_transport of a bzr+ssh:// behaves correctly.
115
bzr+ssh:// should cause bzr to run a remote bzr smart server over SSH.
118
from bzrlib.transport.sftp import SFTPServer
119
except ParamikoNotPresent:
120
raise TestSkipped('Paramiko not installed')
122
from bzrlib.tests.stub_sftp import StubServer
123
from paramiko.common import AUTH_SUCCESSFUL
125
# XXX: Eventually, all SSH vendor classes should implement connect_ssh,
126
# and this TestSkipped can be removed.
127
from bzrlib.transport.ssh import _get_ssh_vendor, SSHVendor
128
vendor = _get_ssh_vendor()
129
if vendor.connect_ssh.im_func == SSHVendor.connect_ssh.im_func:
131
'connect_ssh is not yet implemented on %r' % vendor)
134
self.make_branch('a_branch')
136
# Start an SSH server
137
# XXX: This is horrible -- we define a really dumb SSH server that
138
# executes commands, and manage the hooking up of stdin/out/err to the
139
# SSH channel ourselves. Surely this has already been implemented
141
class StubSSHServer(StubServer):
145
def get_allowed_auths(self, username):
148
def check_auth_none(self, username):
149
return AUTH_SUCCESSFUL
151
def check_channel_exec_request(self, channel, command):
152
self.test.command_executed = command
153
proc = subprocess.Popen(
154
command, shell=True, stdin=subprocess.PIPE,
155
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
157
# XXX: horribly inefficient, not to mention ugly.
159
# copy bytes from the channel to the subprocess's stdin
161
bytes = channel.recv(1)
165
proc.stdin.write(bytes)
167
threading.Thread(target=do_stdin).start()
169
def ferry_bytes(pipe):
175
channel.sendall(bytes)
176
threading.Thread(target=ferry_bytes, args=(proc.stdout,)).start()
177
threading.Thread(target=ferry_bytes, args=(proc.stderr,)).start()
181
ssh_server = SFTPServer(StubSSHServer)
182
# XXX: We *don't* want to override the default SSH vendor, so we set
183
# _vendor to what _get_ssh_vendor returns.
184
ssh_server._vendor = vendor
186
self.addCleanup(ssh_server.tearDown)
187
port = ssh_server._listener.port
189
# Access the branch via a bzr+ssh URL. The BZR_REMOTE_PATH environment
190
# variable is used to tell bzr what command to run on the remote end.
191
path_to_branch = os.path.abspath('a_branch')
193
orig_bzr_remote_path = os.environ.get('BZR_REMOTE_PATH')
194
os.environ['BZR_REMOTE_PATH'] = self.get_bzr_path()
196
branch = Branch.open(
197
'bzr+ssh://localhost:%d%s' % (port, path_to_branch))
199
branch.repository.get_revision_graph()
200
self.assertEqual(None, branch.last_revision())
202
# Restore the BZR_REMOTE_PATH environment variable back to its
204
if orig_bzr_remote_path is None:
205
del os.environ['BZR_REMOTE_PATH']
207
os.environ['BZR_REMOTE_PATH'] = orig_bzr_remote_path
210
'%s serve --inet --directory=/' % self.get_bzr_path(),
211
self.command_executed)
added
removed
bzrlib/tests/blackbox/test_serve.py
