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
"""bzr+ssh should work in ordinary bzr commands without blowing up."""
115
from bzrlib.transport.sftp import SFTPServer
116
except ParamikoNotPresent:
117
raise TestSkipped('Paramiko not installed')
119
from bzrlib.tests.stub_sftp import StubServer
120
from paramiko.common import AUTH_SUCCESSFUL
122
# XXX: Eventually, all SSH vendor classes should implement this.
123
from bzrlib.transport.ssh import _get_ssh_vendor, SSHVendor
124
vendor = _get_ssh_vendor()
125
if vendor.connect_ssh.im_func == SSHVendor.connect_ssh.im_func:
127
'connect_ssh is not yet implemented on %r' % vendor)
130
self.make_branch('a_branch')
132
# Start an SSH server
133
# XXX: This is horrible -- we define a really dumb SSH server that
134
# executes commands, and manage the hooking up of stdin/out/err to the
135
# SSH channel ourselves. Surely this has already been implemented
137
class StubSSHServer(StubServer):
141
def get_allowed_auths(self, username):
144
def check_auth_none(self, username):
145
return AUTH_SUCCESSFUL
147
def check_channel_exec_request(self, channel, command):
148
self.test.command_executed = command
149
proc = subprocess.Popen(
150
command, shell=True, stdin=subprocess.PIPE,
151
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
153
# XXX: horribly inefficient, not to mention ugly.
155
# copy bytes from the channel to the subprocess's stdin
157
bytes = channel.recv(1)
161
proc.stdin.write(bytes)
163
threading.Thread(target=do_stdin).start()
165
def ferry_bytes(pipe):
171
channel.sendall(bytes)
172
threading.Thread(target=ferry_bytes, args=(proc.stdout,)).start()
173
threading.Thread(target=ferry_bytes, args=(proc.stderr,)).start()
177
ssh_server = SFTPServer(StubSSHServer)
179
self.addCleanup(ssh_server.tearDown)
180
port = ssh_server._listener.port
182
# Access the branch via a bzr+ssh URL. The BZR_REMOTE_PATH environment
183
# variable is used to tell bzr what command to run on the remote end.
184
path_to_branch = os.path.abspath('.')
185
self.run_bzr_subprocess(
186
'log', 'bzr+ssh://localhost:%d/%s' % (port, path_to_branch),
187
env_changes={'BZR_REMOTE_PATH': self.get_bzr_path()})
190
'%s serve --inet --directory=/' % self.get_bzr_path(),
191
self.command_executed)
added
removed
bzrlib/tests/blackbox/test_serve.py
