← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_souk_transport.py

Reorganize SoukServer; implement basic get_bundle command.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_souk_transport.py
16
16
 
17
17
"""Tests for souk transport"""
18
18
 
19
 
# TODO: Try sending multiple requests; they should all get answers.
20
 
 
21
 
# TODO: If the server raises an error within its processing that should be
22
 
# caught and propagated back to the client.
23
 
 
24
19
# all of this deals with byte strings so this is safe
25
20
from cStringIO import StringIO
26
21
import subprocess
34
29
    
35
30
    def test_souk_query_version(self):
36
31
        """Feed a canned query version to a server"""
37
 
        to_server = StringIO('hello\0011\n')
 
32
        to_server = StringIO('hello\n')
38
33
        from_server = StringIO()
39
34
        server = souk.SoukStreamServer(to_server, from_server, local.LocalTransport('file:///'))
40
35
        server._serve_one_request()
43
38
 
44
39
    def test_canned_get_response(self):
45
40
        transport = memory.MemoryTransport('memory:///')
46
 
        transport.put('hello', StringIO('contents\nof\nfile\n'))
47
 
        to_server = StringIO('get\001./hello\n')
 
41
        transport.put('testfile', StringIO('contents\nof\nfile\n'))
 
42
        to_server = StringIO('get\001./testfile\n')
48
43
        from_server = StringIO()
49
44
        server = souk.SoukStreamServer(to_server, from_server, transport)
50
45
        server._serve_one_request()
180
175
        t = self.backing_transport
181
176
        bzrdir.BzrDirFormat.get_default_format().initialize_on_transport(t)
182
177
        result_dir = bzrdir.BzrDir.open_containing_from_transport(conn)
 
178
 
 
179
 
 
180
class SoukServerTests(tests.TestCaseWithTransport):
 
181
    """Test that call directly into the server logic, bypassing the network."""
 
182
 
 
183
    def test_hello(self):
 
184
        server = souk.SoukServer(None)
 
185
        response = server.dispatch_command('hello', ())
 
186
        self.assertEqual(('bzr server', '1'), response.args)
 
187
        self.assertEqual(None, response.body)
 
188
        
 
189
    def test_get_bundle(self):
 
190
        wt = self.make_branch_and_tree('.')
 
191
        b = wt.branch
 
192
        file('hello', 'w').write('hello world')
 
193
        wt.add('hello')
 
194
        wt.commit(message='add hello', rev_id='rev-1')
 
195
        
 
196
        server = souk.SoukServer(self.get_transport())
 
197
        response = server.dispatch_command('get_bundle', ('.', 'rev-1'))
 
198
        self.assert_(response.body.startswith('# Bazaar revision bundle '),
 
199
                repr(response.body))
 
200