/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/blackbox/test_serve.py

Merge --allow-writes from Robert

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/blackbox/test_serve.py
22
22
import subprocess
23
23
import threading
24
24
 
 
25
from bzrlib import errors
25
26
from bzrlib.branch import Branch
26
27
from bzrlib.bzrdir import BzrDir
27
28
from bzrlib.errors import ParamikoNotPresent
28
29
from bzrlib.tests import TestCaseWithTransport, TestSkipped
29
 
from bzrlib.transport import smart
 
30
from bzrlib.transport import get_transport, smart
30
31
 
31
32
 
32
33
class DoesNotCloseStdOutClient(smart.SmartStreamClient):
45
46
 
46
47
class TestBzrServe(TestCaseWithTransport):
47
48
 
48
 
    def test_bzr_serve_inet(self):
49
 
        # Make a branch
50
 
        self.make_branch('.')
51
 
 
52
 
        # Serve that branch from the current directory
53
 
        process = self.start_bzr_subprocess(['serve', '--inet'])
54
 
 
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)
61
 
 
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())
66
 
 
67
 
        # finish with the transport
68
 
        del transport
 
49
    def assertInetServerShutsdownCleanly(self, client, process):
 
50
        """Shutdown the server process looking for errors."""
69
51
        # Disconnect the client forcefully JUST IN CASE because of __del__'s use
70
52
        # in the smart module.
71
53
        client.disconnect()
79
61
        self.assertEqual('', result[0])
80
62
        self.assertEqual('', result[1])
81
63
    
82
 
    def test_bzr_serve_port(self):
83
 
        # Make a branch
84
 
        self.make_branch('.')
85
 
 
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)
 
64
    def assertServerFinishesCleanly(self, process):
 
65
        """Shutdown the bzr serve instance process looking for errors."""
 
66
        # Shutdown the server
 
67
        result = self.finish_bzr_subprocess(process, retcode=3,
 
68
                                            send_signal=signal.SIGINT)
 
69
        self.assertEqual('', result[0])
 
70
        self.assertEqual('bzr: interrupted\n', result[1])
 
71
 
 
72
    def start_server_inet(self, extra_options=()):
 
73
        """Start a bzr server subprocess using the --inet option.
 
74
 
 
75
        :param extra_options: extra options to give the server.
 
76
        :return: a tuple with the bzr process handle for passing to
 
77
            finish_bzr_subprocess, a client for the server, and a transport.
 
78
        """
 
79
        # Serve from the current directory
 
80
        process = self.start_bzr_subprocess(['serve', '--inet'])
 
81
 
 
82
        # Connect to the server
 
83
        # We use this url because while this is no valid URL to connect to this
 
84
        # 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
 
89
 
 
90
    def start_server_port(self, extra_options=()):
 
91
        """Start a bzr server subprocess.
 
92
 
 
93
        :param extra_options: extra options to give the server.
 
94
        :return: a tuple with the bzr process handle for passing to
 
95
            finish_bzr_subprocess, and the base url for the server.
 
96
        """
 
97
        # Serve from the current directory
 
98
        args = ['serve', '--port', 'localhost:0']
 
99
        args.extend(extra_options)
 
100
        process = self.start_bzr_subprocess(args, skip_if_plan_to_signal=True)
89
101
        port_line = process.stdout.readline()
90
102
        prefix = 'listening on port: '
91
103
        self.assertStartsWith(port_line, prefix)
92
104
        port = int(port_line[len(prefix):])
 
105
        return process,'bzr://localhost:%d/' % port
 
106
 
 
107
    def test_bzr_serve_inet_readonly(self):
 
108
        """bzr server should provide a read only filesystem by default."""
 
109
        process, client, transport = self.start_server_inet()
 
110
        self.assertRaises(errors.TransportNotPossible, transport.mkdir, 'adir')
 
111
        # finish with the transport
 
112
        del transport
 
113
        self.assertInetServerShutsdownCleanly(client, process)
 
114
 
 
115
    def test_bzr_serve_inet_readwrite(self):
 
116
        # Make a branch
 
117
        self.make_branch('.')
 
118
 
 
119
        process, client, transport = self.start_server_inet(['--allow-writes'])
 
120
 
 
121
        # We get a working branch
 
122
        branch = BzrDir.open_from_transport(transport).open_branch()
 
123
        branch.repository.get_revision_graph()
 
124
        self.assertEqual(None, branch.last_revision())
 
125
 
 
126
        # finish with the transport
 
127
        del transport
 
128
 
 
129
        self.assertInetServerShutsdownCleanly(client, process)
 
130
 
 
131
    def test_bzr_serve_port_readonly(self):
 
132
        """bzr server should provide a read only filesystem by default."""
 
133
        process, url = self.start_server_port()
 
134
        transport = get_transport(url)
 
135
        self.assertRaises(errors.TransportNotPossible, transport.mkdir, 'adir')
 
136
        self.assertServerFinishesCleanly(process)
 
137
 
 
138
    def test_bzr_serve_port_readwrite(self):
 
139
        # Make a branch
 
140
        self.make_branch('.')
 
141
 
 
142
        process, url = self.start_server_port(['--allow-writes'])
93
143
 
94
144
        # Connect to the server
95
 
        branch = Branch.open('bzr://localhost:%d/' % port)
 
145
        branch = Branch.open(url)
96
146
 
97
147
        # We get a working branch
98
148
        branch.repository.get_revision_graph()
99
149
        self.assertEqual(None, branch.last_revision())
100
150
 
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])
 
151
        self.assertServerFinishesCleanly(process)
106
152
 
107
153
    def test_bzr_serve_no_args(self):
108
154
        """'bzr serve' with no arguments or options should not traceback."""
182
228
            
183
229
            branch.repository.get_revision_graph()
184
230
            self.assertEqual(None, branch.last_revision())
 
231
            # Check we can perform write operations
 
232
            branch.bzrdir.root_transport.mkdir('foo')
185
233
        finally:
186
234
            # Restore the BZR_REMOTE_PATH environment variable back to its
187
235
            # original state.
191
239
                os.environ['BZR_REMOTE_PATH'] = orig_bzr_remote_path
192
240
 
193
241
        self.assertEqual(
194
 
            '%s serve --inet --directory=/' % self.get_bzr_path(),
 
242
            '%s serve --inet --directory=/ --allow-writes'
 
243
            % self.get_bzr_path(),
195
244
            self.command_executed)
196
245