30
30
from bzrlib.transport import get_transport, smart
33
class DoesNotCloseStdOutClient(smart.SmartStreamClient):
34
"""A client that doesn't close stdout upon disconnect().
36
We wish to let stdout remain open so that we can see if the server writes
37
anything to stdout during its shutdown.
42
self._connected = False
43
# The client's out is the server's in.
47
33
class TestBzrServe(TestCaseWithTransport):
49
def assertInetServerShutsdownCleanly(self, client, process):
35
def assertInetServerShutsdownCleanly(self, process):
50
36
"""Shutdown the server process looking for errors."""
51
# Disconnect the client forcefully JUST IN CASE because of __del__'s use
52
# in the smart module.
55
# Shutdown the server: the client should have disconnected cleanly and
56
# closed stdin, so the server process should shut itself down.
57
self.assertTrue(process.stdin.closed)
37
# Shutdown the server: the server should shut down when it cannot read
58
40
# Hide stdin from the subprocess module, so it won't fail to close it.
59
41
process.stdin = None
60
42
result = self.finish_bzr_subprocess(process, retcode=0)
82
64
# Connect to the server
83
65
# We use this url because while this is no valid URL to connect to this
84
66
# server instance, the transport needs a URL.
85
client = DoesNotCloseStdOutClient(
86
lambda: (process.stdout, process.stdin))
87
transport = smart.SmartTransport('bzr://localhost/', client=client)
88
return process, client, transport
67
medium = smart.SmartSimplePipesClientMedium(
68
process.stdout, process.stdin)
69
transport = smart.SmartTransport('bzr://localhost/', medium=medium)
70
return process, transport
90
72
def start_server_port(self, extra_options=()):
91
73
"""Start a bzr server subprocess.
107
89
def test_bzr_serve_inet_readonly(self):
108
90
"""bzr server should provide a read only filesystem by default."""
109
process, client, transport = self.start_server_inet()
91
process, transport = self.start_server_inet()
110
92
self.assertRaises(errors.TransportNotPossible, transport.mkdir, 'adir')
111
# finish with the transport
113
self.assertInetServerShutsdownCleanly(client, process)
93
self.assertInetServerShutsdownCleanly(process)
115
95
def test_bzr_serve_inet_readwrite(self):
117
97
self.make_branch('.')
119
process, client, transport = self.start_server_inet(['--allow-writes'])
99
process, transport = self.start_server_inet(['--allow-writes'])
121
101
# We get a working branch
122
102
branch = BzrDir.open_from_transport(transport).open_branch()
123
103
branch.repository.get_revision_graph()
124
104
self.assertEqual(None, branch.last_revision())
126
# finish with the transport
129
self.assertInetServerShutsdownCleanly(client, process)
105
self.assertInetServerShutsdownCleanly(process)
131
107
def test_bzr_serve_port_readonly(self):
132
108
"""bzr server should provide a read only filesystem by default."""