/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/test_remote.py

  • Committer: Alexander Belchenko
  • Date: 2008-02-20 10:36:15 UTC
  • mfrom: (3228 +trunk)
  • mto: This revision was merged to the branch mainline in revision 3231.
  • Revision ID: bialix@ukr.net-20080220103615-uxw9jc9m6l63ea27
merge bzr.dev; update patch for 1.3

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_remote.py
23
23
These tests correspond to tests.test_smart, which exercises the server side.
24
24
"""
25
25
 
 
26
import bz2
26
27
from cStringIO import StringIO
27
28
 
28
29
from bzrlib import (
139
140
        self.responses = responses
140
141
        self._calls = []
141
142
        self.expecting_body = False
142
 
        _SmartClient.__init__(self, FakeMedium(fake_medium_base))
 
143
        _SmartClient.__init__(self, FakeMedium(fake_medium_base, self._calls))
143
144
 
144
145
    def call(self, method, *args):
145
146
        self._calls.append(('call', method, args))
161
162
 
162
163
class FakeMedium(object):
163
164
 
164
 
    def __init__(self, base):
 
165
    def __init__(self, base, client_calls):
165
166
        self.base = base
 
167
        self.connection = FakeConnection(client_calls)
 
168
        self._client_calls = client_calls
 
169
 
 
170
 
 
171
class FakeConnection(object):
 
172
 
 
173
    def __init__(self, client_calls):
 
174
        self._remote_is_at_least_1_2 = True
 
175
        self._client_calls = client_calls
 
176
 
 
177
    def disconnect(self):
 
178
        self._client_calls.append(('disconnect medium',))
166
179
 
167
180
 
168
181
class TestVfsHas(tests.TestCase):
184
197
        transport = MemoryTransport()
185
198
        transport.mkdir('quack')
186
199
        transport = transport.clone('quack')
187
 
        client = FakeClient([(('ok', ''), ), (('ok', '', 'no', 'no'), )],
 
200
        client = FakeClient([(('ok', ''), ), (('ok', '', 'no', 'no', 'no'), )],
188
201
                            transport.base)
189
202
        bzrdir = RemoteBzrDir(transport, _client=client)
190
203
        result = bzrdir.open_branch()
191
204
        self.assertEqual(
192
205
            [('call', 'BzrDir.open_branch', ('quack/',)),
193
 
             ('call', 'BzrDir.find_repository', ('quack/',))],
 
206
             ('call', 'BzrDir.find_repositoryV2', ('quack/',))],
194
207
            client._calls)
195
208
        self.assertIsInstance(result, RemoteBranch)
196
209
        self.assertEqual(bzrdir, result.bzrdir)
206
219
            [('call', 'BzrDir.open_branch', ('quack/',))],
207
220
            client._calls)
208
221
 
 
222
    def test__get_tree_branch(self):
 
223
        # _get_tree_branch is a form of open_branch, but it should only ask for
 
224
        # branch opening, not any other network requests.
 
225
        calls = []
 
226
        def open_branch():
 
227
            calls.append("Called")
 
228
            return "a-branch"
 
229
        transport = MemoryTransport()
 
230
        # no requests on the network - catches other api calls being made.
 
231
        client = FakeClient([], transport.base)
 
232
        bzrdir = RemoteBzrDir(transport, _client=client)
 
233
        # patch the open_branch call to record that it was called.
 
234
        bzrdir.open_branch = open_branch
 
235
        self.assertEqual((None, "a-branch"), bzrdir._get_tree_branch())
 
236
        self.assertEqual(["Called"], calls)
 
237
        self.assertEqual([], client._calls)
 
238
 
209
239
    def test_url_quoting_of_path(self):
210
240
        # Relpaths on the wire should not be URL-escaped.  So "~" should be
211
241
        # transmitted as "~", not "%7E".
212
242
        transport = RemoteTransport('bzr://localhost/~hello/')
213
 
        client = FakeClient([(('ok', ''), ), (('ok', '', 'no', 'no'), )],
 
243
        client = FakeClient([(('ok', ''), ), (('ok', '', 'no', 'no', 'no'), )],
214
244
                            transport.base)
215
245
        bzrdir = RemoteBzrDir(transport, _client=client)
216
246
        result = bzrdir.open_branch()
217
247
        self.assertEqual(
218
248
            [('call', 'BzrDir.open_branch', ('~hello/',)),
219
 
             ('call', 'BzrDir.find_repository', ('~hello/',))],
 
249
             ('call', 'BzrDir.find_repositoryV2', ('~hello/',))],
220
250
            client._calls)
221
251
 
222
 
    def check_open_repository(self, rich_root, subtrees):
 
252
    def check_open_repository(self, rich_root, subtrees, external_lookup='no'):
223
253
        transport = MemoryTransport()
224
254
        transport.mkdir('quack')
225
255
        transport = transport.clone('quack')
231
261
            subtree_response = 'yes'
232
262
        else:
233
263
            subtree_response = 'no'
234
 
        client = FakeClient([(('ok', '', rich_response, subtree_response), ),],
235
 
                            transport.base)
 
264
        client = FakeClient(
 
265
            [(('ok', '', rich_response, subtree_response, external_lookup), ),],
 
266
            transport.base)
236
267
        bzrdir = RemoteBzrDir(transport, _client=client)
237
268
        result = bzrdir.open_repository()
238
269
        self.assertEqual(
239
 
            [('call', 'BzrDir.find_repository', ('quack/',))],
 
270
            [('call', 'BzrDir.find_repositoryV2', ('quack/',))],
240
271
            client._calls)
241
272
        self.assertIsInstance(result, RemoteRepository)
242
273
        self.assertEqual(bzrdir, result.bzrdir)
248
279
        self.check_open_repository(False, True)
249
280
        self.check_open_repository(True, False)
250
281
        self.check_open_repository(False, False)
 
282
        self.check_open_repository(False, False, 'yes')
251
283
 
252
284
    def test_old_server(self):
253
285
        """RemoteBzrDirFormat should fail to probe if the server version is too
608
640
        r1 = u'\u0e33'.encode('utf8')
609
641
        r2 = u'\u0dab'.encode('utf8')
610
642
        lines = [' '.join([r2, r1]), r1]
611
 
        encoded_body = '\n'.join(lines)
 
643
        encoded_body = bz2.compress('\n'.join(lines))
612
644
        responses = [(('ok', ), encoded_body), (('ok', ), encoded_body)]
613
645
 
614
646
        transport_path = 'quack'
624
656
        parents = graph.get_parent_map([r1])
625
657
        self.assertEqual({r1: (NULL_REVISION,)}, parents)
626
658
        self.assertEqual(
627
 
            [('call_expecting_body', 'Repository.get_parent_map',
628
 
             ('quack/', r2))],
 
659
            [('call_with_body_bytes_expecting_body',
 
660
              'Repository.get_parent_map', ('quack/', r2), '\n\n0')],
629
661
            client._calls)
630
662
        repo.unlock()
631
663
        # now we call again, and it should use the second response.
634
666
        parents = graph.get_parent_map([r1])
635
667
        self.assertEqual({r1: (NULL_REVISION,)}, parents)
636
668
        self.assertEqual(
637
 
            [('call_expecting_body', 'Repository.get_parent_map',
638
 
              ('quack/', r2)),
639
 
             ('call_expecting_body', 'Repository.get_parent_map',
640
 
              ('quack/', r1))
 
669
            [('call_with_body_bytes_expecting_body',
 
670
              'Repository.get_parent_map', ('quack/', r2), '\n\n0'),
 
671
             ('call_with_body_bytes_expecting_body',
 
672
              'Repository.get_parent_map', ('quack/', r1), '\n\n0'),
641
673
            ],
642
674
            client._calls)
643
675
        repo.unlock()
644
676
 
 
677
    def test_get_parent_map_reconnects_if_unknown_method(self):
 
678
        error_msg = (
 
679
            "Generic bzr smart protocol error: "
 
680
            "bad request 'Repository.get_parent_map'")
 
681
        responses = [
 
682
            (('error', error_msg), ''),
 
683
            (('ok',), '')]
 
684
        transport_path = 'quack'
 
685
        repo, client = self.setup_fake_client_and_repository(
 
686
            responses, transport_path)
 
687
        rev_id = 'revision-id'
 
688
        parents = repo.get_parent_map([rev_id])
 
689
        self.assertEqual(
 
690
            [('call_with_body_bytes_expecting_body',
 
691
              'Repository.get_parent_map', ('quack/', rev_id), '\n\n0'),
 
692
             ('disconnect medium',),
 
693
             ('call_expecting_body', 'Repository.get_revision_graph',
 
694
              ('quack/', ''))],
 
695
            client._calls)
 
696
 
 
697
 
645
698
 
646
699
class TestRepositoryGetRevisionGraph(TestRemoteRepository):
647
700