← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_smart.py

  • Committer: Martin Pool
  • Date: 2010-01-29 14:09:05 UTC
  • mto: This revision was merged to the branch mainline in revision 4992.
  • Revision ID: mbp@sourcefrog.net-20100129140905-2uiarb6p8di1ywsr
Correction to url

from review: https://code.edge.launchpad.net/~mbp/bzr/doc/+merge/18250

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smart.py
36
36
    smart,
37
37
    tests,
38
38
    urlutils,
 
39
    versionedfile,
39
40
    )
40
41
from bzrlib.branch import Branch, BranchReferenceFormat
41
42
import bzrlib.smart.branch
42
43
import bzrlib.smart.bzrdir, bzrlib.smart.bzrdir as smart_dir
43
44
import bzrlib.smart.packrepository
44
45
import bzrlib.smart.repository
 
46
import bzrlib.smart.vfs
45
47
from bzrlib.smart.request import (
46
48
    FailedSmartServerResponse,
47
49
    SmartServerRequest,
51
53
from bzrlib.tests import (
52
54
    split_suite_by_re,
53
55
    )
54
 
from bzrlib.transport import chroot, get_transport
 
56
from bzrlib.transport import chroot, get_transport, local, memory
55
57
 
56
58
 
57
59
def load_tests(standard_tests, module, loader):
80
82
class TestCaseWithChrootedTransport(tests.TestCaseWithTransport):
81
83
 
82
84
    def setUp(self):
 
85
        self.vfs_transport_factory = memory.MemoryServer
83
86
        tests.TestCaseWithTransport.setUp(self)
84
87
        self._chroot_server = None
85
88
 
87
90
        if self._chroot_server is None:
88
91
            backing_transport = tests.TestCaseWithTransport.get_transport(self)
89
92
            self._chroot_server = chroot.ChrootServer(backing_transport)
90
 
            self._chroot_server.setUp()
91
 
            self.addCleanup(self._chroot_server.tearDown)
 
93
            self.start_server(self._chroot_server)
92
94
        t = get_transport(self._chroot_server.get_url())
93
95
        if relpath is not None:
94
96
            t = t.clone(relpath)
95
97
        return t
96
98
 
97
99
 
98
 
class TestCaseWithSmartMedium(tests.TestCaseWithTransport):
 
100
class TestCaseWithSmartMedium(tests.TestCaseWithMemoryTransport):
99
101
 
100
102
    def setUp(self):
101
103
        super(TestCaseWithSmartMedium, self).setUp()
113
115
        return self.get_transport().get_smart_medium()
114
116
 
115
117
 
 
118
class TestByteStreamToStream(tests.TestCase):
 
119
 
 
120
    def test_repeated_substreams_same_kind_are_one_stream(self):
 
121
        # Make a stream - an iterable of bytestrings.
 
122
        stream = [('text', [versionedfile.FulltextContentFactory(('k1',), None,
 
123
            None, 'foo')]),('text', [
 
124
            versionedfile.FulltextContentFactory(('k2',), None, None, 'bar')])]
 
125
        fmt = bzrdir.format_registry.get('pack-0.92')().repository_format
 
126
        bytes = smart.repository._stream_to_byte_stream(stream, fmt)
 
127
        streams = []
 
128
        # Iterate the resulting iterable; checking that we get only one stream
 
129
        # out.
 
130
        fmt, stream = smart.repository._byte_stream_to_stream(bytes)
 
131
        for kind, substream in stream:
 
132
            streams.append((kind, list(substream)))
 
133
        self.assertLength(1, streams)
 
134
        self.assertLength(2, streams[0][1])
 
135
 
 
136
 
116
137
class TestSmartServerResponse(tests.TestCase):
117
138
 
118
139
    def test__eq__(self):
150
171
        self.assertRaises(
151
172
            errors.PathNotChild, request.translate_client_path, 'bar/')
152
173
        self.assertEqual('./baz', request.translate_client_path('foo/baz'))
 
174
        e_acute = u'\N{LATIN SMALL LETTER E WITH ACUTE}'.encode('utf-8')
 
175
        self.assertEqual('./' + urlutils.escape(e_acute),
 
176
                         request.translate_client_path('foo/' + e_acute))
 
177
 
 
178
    def test_translate_client_path_vfs(self):
 
179
        """VfsRequests receive escaped paths rather than raw UTF-8."""
 
180
        transport = self.get_transport()
 
181
        request = smart.vfs.VfsRequest(transport, 'foo/')
 
182
        e_acute = u'\N{LATIN SMALL LETTER E WITH ACUTE}'.encode('utf-8')
 
183
        escaped = urlutils.escape('foo/' + e_acute)
 
184
        self.assertEqual('./' + urlutils.escape(e_acute),
 
185
                         request.translate_client_path(escaped))
153
186
 
154
187
    def test_transport_from_client_path(self):
155
188
        transport = self.get_transport()
202
235
        self.make_bzrdir('.')
203
236
        request_class = bzrlib.smart.bzrdir.SmartServerRequestCreateRepository
204
237
        request = request_class(backing)
205
 
        reference_bzrdir_format = bzrdir.format_registry.get('default')()
 
238
        reference_bzrdir_format = bzrdir.format_registry.get('pack-0.92')()
206
239
        reference_format = reference_bzrdir_format.repository_format
207
240
        network_name = reference_format.network_name()
208
241
        expected = SuccessfulSmartServerResponse(
247
280
            subtrees = 'yes'
248
281
        else:
249
282
            subtrees = 'no'
 
283
        if repo._format.supports_external_lookups:
 
284
            external = 'yes'
 
285
        else:
 
286
            external = 'no'
250
287
        if (smart.bzrdir.SmartServerRequestFindRepositoryV3 ==
251
288
            self._request_class):
252
289
            return SuccessfulSmartServerResponse(
253
 
                ('ok', '', rich_root, subtrees, 'no',
 
290
                ('ok', '', rich_root, subtrees, external,
254
291
                 repo._format.network_name()))
255
292
        elif (smart.bzrdir.SmartServerRequestFindRepositoryV2 ==
256
293
            self._request_class):
257
294
            # All tests so far are on formats, and for non-external
258
295
            # repositories.
259
296
            return SuccessfulSmartServerResponse(
260
 
                ('ok', '', rich_root, subtrees, 'no'))
 
297
                ('ok', '', rich_root, subtrees, external))
261
298
        else:
262
299
            return SuccessfulSmartServerResponse(('ok', '', rich_root, subtrees))
263
300
 
390
427
            'False', 'False', 'False', '', '', '', '', 'False')
391
428
 
392
429
 
 
430
class TestSmartServerRequestOpenBzrDir(tests.TestCaseWithMemoryTransport):
 
431
    
 
432
    def test_no_directory(self):
 
433
        backing = self.get_transport()
 
434
        request = smart.bzrdir.SmartServerRequestOpenBzrDir(backing)
 
435
        self.assertEqual(SmartServerResponse(('no', )),
 
436
            request.execute('does-not-exist'))
 
437
 
 
438
    def test_empty_directory(self):
 
439
        backing = self.get_transport()
 
440
        backing.mkdir('empty')
 
441
        request = smart.bzrdir.SmartServerRequestOpenBzrDir(backing)
 
442
        self.assertEqual(SmartServerResponse(('no', )),
 
443
            request.execute('empty'))
 
444
 
 
445
    def test_outside_root_client_path(self):
 
446
        backing = self.get_transport()
 
447
        request = smart.bzrdir.SmartServerRequestOpenBzrDir(backing,
 
448
            root_client_path='root')
 
449
        self.assertEqual(SmartServerResponse(('no', )),
 
450
            request.execute('not-root'))
 
451
 
 
452
    
 
453
class TestSmartServerRequestOpenBzrDir_2_1(tests.TestCaseWithMemoryTransport):
 
454
    
 
455
    def test_no_directory(self):
 
456
        backing = self.get_transport()
 
457
        request = smart.bzrdir.SmartServerRequestOpenBzrDir_2_1(backing)
 
458
        self.assertEqual(SmartServerResponse(('no', )),
 
459
            request.execute('does-not-exist'))
 
460
 
 
461
    def test_empty_directory(self):
 
462
        backing = self.get_transport()
 
463
        backing.mkdir('empty')
 
464
        request = smart.bzrdir.SmartServerRequestOpenBzrDir_2_1(backing)
 
465
        self.assertEqual(SmartServerResponse(('no', )),
 
466
            request.execute('empty'))
 
467
 
 
468
    def test_present_without_workingtree(self):
 
469
        backing = self.get_transport()
 
470
        request = smart.bzrdir.SmartServerRequestOpenBzrDir_2_1(backing)
 
471
        self.make_bzrdir('.')
 
472
        self.assertEqual(SmartServerResponse(('yes', 'no')),
 
473
            request.execute(''))
 
474
 
 
475
    def test_outside_root_client_path(self):
 
476
        backing = self.get_transport()
 
477
        request = smart.bzrdir.SmartServerRequestOpenBzrDir_2_1(backing,
 
478
            root_client_path='root')
 
479
        self.assertEqual(SmartServerResponse(('no',)),
 
480
            request.execute('not-root'))
 
481
 
 
482
    
 
483
class TestSmartServerRequestOpenBzrDir_2_1_disk(TestCaseWithChrootedTransport):
 
484
 
 
485
    def test_present_with_workingtree(self):
 
486
        self.vfs_transport_factory = local.LocalURLServer
 
487
        backing = self.get_transport()
 
488
        request = smart.bzrdir.SmartServerRequestOpenBzrDir_2_1(backing)
 
489
        bd = self.make_bzrdir('.')
 
490
        bd.create_repository()
 
491
        bd.create_branch()
 
492
        bd.create_workingtree()
 
493
        self.assertEqual(SmartServerResponse(('yes', 'yes')),
 
494
            request.execute(''))
 
495
 
 
496
 
393
497
class TestSmartServerRequestOpenBranch(TestCaseWithChrootedTransport):
394
498
 
395
499
    def test_no_branch(self):
410
514
 
411
515
    def test_branch_reference(self):
412
516
        """When there is a branch reference, the reference URL is returned."""
 
517
        self.vfs_transport_factory = local.LocalURLServer
413
518
        backing = self.get_transport()
414
519
        request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
415
520
        branch = self.make_branch('branch')
419
524
        self.assertEqual(SmartServerResponse(('ok', reference_url)),
420
525
            request.execute('reference'))
421
526
 
 
527
    def test_notification_on_branch_from_repository(self):
 
528
        """When there is a repository, the error should return details."""
 
529
        backing = self.get_transport()
 
530
        request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
 
531
        repo = self.make_repository('.')
 
532
        self.assertEqual(SmartServerResponse(('nobranch',)),
 
533
            request.execute(''))
 
534
 
422
535
 
423
536
class TestSmartServerRequestOpenBranchV2(TestCaseWithChrootedTransport):
424
537
 
440
553
 
441
554
    def test_branch_reference(self):
442
555
        """When there is a branch reference, the reference URL is returned."""
443
 
        backing = self.get_transport()
444
 
        request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
445
 
        branch = self.make_branch('branch')
446
 
        checkout = branch.create_checkout('reference',lightweight=True)
447
 
        reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
448
 
        self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
449
 
        self.assertEqual(SuccessfulSmartServerResponse(('ref', reference_url)),
450
 
            request.execute('reference'))
451
 
 
452
 
    def test_stacked_branch(self):
453
 
        """Opening a stacked branch does not open the stacked-on branch."""
454
 
        trunk = self.make_branch('trunk')
455
 
        feature = self.make_branch('feature', format='1.9')
456
 
        feature.set_stacked_on_url(trunk.base)
457
 
        opened_branches = []
458
 
        Branch.hooks.install_named_hook('open', opened_branches.append, None)
459
 
        backing = self.get_transport()
460
 
        request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
461
 
        request.setup_jail()
462
 
        try:
463
 
            response = request.execute('feature')
464
 
        finally:
465
 
            request.teardown_jail()
466
 
        expected_format = feature._format.network_name()
467
 
        self.assertEqual(
468
 
            SuccessfulSmartServerResponse(('branch', expected_format)),
469
 
            response)
470
 
        self.assertLength(1, opened_branches)
 
556
        self.vfs_transport_factory = local.LocalURLServer
 
557
        backing = self.get_transport()
 
558
        request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
 
559
        branch = self.make_branch('branch')
 
560
        checkout = branch.create_checkout('reference',lightweight=True)
 
561
        reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
 
562
        self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
 
563
        self.assertEqual(SuccessfulSmartServerResponse(('ref', reference_url)),
 
564
            request.execute('reference'))
 
565
 
 
566
    def test_stacked_branch(self):
 
567
        """Opening a stacked branch does not open the stacked-on branch."""
 
568
        trunk = self.make_branch('trunk')
 
569
        feature = self.make_branch('feature')
 
570
        feature.set_stacked_on_url(trunk.base)
 
571
        opened_branches = []
 
572
        Branch.hooks.install_named_hook('open', opened_branches.append, None)
 
573
        backing = self.get_transport()
 
574
        request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
 
575
        request.setup_jail()
 
576
        try:
 
577
            response = request.execute('feature')
 
578
        finally:
 
579
            request.teardown_jail()
 
580
        expected_format = feature._format.network_name()
 
581
        self.assertEqual(
 
582
            SuccessfulSmartServerResponse(('branch', expected_format)),
 
583
            response)
 
584
        self.assertLength(1, opened_branches)
 
585
 
 
586
    def test_notification_on_branch_from_repository(self):
 
587
        """When there is a repository, the error should return details."""
 
588
        backing = self.get_transport()
 
589
        request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
 
590
        repo = self.make_repository('.')
 
591
        self.assertEqual(SmartServerResponse(('nobranch',)),
 
592
            request.execute(''))
 
593
 
 
594
 
 
595
class TestSmartServerRequestOpenBranchV3(TestCaseWithChrootedTransport):
 
596
 
 
597
    def test_no_branch(self):
 
598
        """When there is no branch, ('nobranch', ) is returned."""
 
599
        backing = self.get_transport()
 
600
        self.make_bzrdir('.')
 
601
        request = smart.bzrdir.SmartServerRequestOpenBranchV3(backing)
 
602
        self.assertEqual(SmartServerResponse(('nobranch',)),
 
603
            request.execute(''))
 
604
 
 
605
    def test_branch(self):
 
606
        """When there is a branch, 'ok' is returned."""
 
607
        backing = self.get_transport()
 
608
        expected = self.make_branch('.')._format.network_name()
 
609
        request = smart.bzrdir.SmartServerRequestOpenBranchV3(backing)
 
610
        self.assertEqual(SuccessfulSmartServerResponse(('branch', expected)),
 
611
            request.execute(''))
 
612
 
 
613
    def test_branch_reference(self):
 
614
        """When there is a branch reference, the reference URL is returned."""
 
615
        self.vfs_transport_factory = local.LocalURLServer
 
616
        backing = self.get_transport()
 
617
        request = smart.bzrdir.SmartServerRequestOpenBranchV3(backing)
 
618
        branch = self.make_branch('branch')
 
619
        checkout = branch.create_checkout('reference',lightweight=True)
 
620
        reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
 
621
        self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
 
622
        self.assertEqual(SuccessfulSmartServerResponse(('ref', reference_url)),
 
623
            request.execute('reference'))
 
624
 
 
625
    def test_stacked_branch(self):
 
626
        """Opening a stacked branch does not open the stacked-on branch."""
 
627
        trunk = self.make_branch('trunk')
 
628
        feature = self.make_branch('feature')
 
629
        feature.set_stacked_on_url(trunk.base)
 
630
        opened_branches = []
 
631
        Branch.hooks.install_named_hook('open', opened_branches.append, None)
 
632
        backing = self.get_transport()
 
633
        request = smart.bzrdir.SmartServerRequestOpenBranchV3(backing)
 
634
        request.setup_jail()
 
635
        try:
 
636
            response = request.execute('feature')
 
637
        finally:
 
638
            request.teardown_jail()
 
639
        expected_format = feature._format.network_name()
 
640
        self.assertEqual(
 
641
            SuccessfulSmartServerResponse(('branch', expected_format)),
 
642
            response)
 
643
        self.assertLength(1, opened_branches)
 
644
 
 
645
    def test_notification_on_branch_from_repository(self):
 
646
        """When there is a repository, the error should return details."""
 
647
        backing = self.get_transport()
 
648
        request = smart.bzrdir.SmartServerRequestOpenBranchV3(backing)
 
649
        repo = self.make_repository('.')
 
650
        self.assertEqual(
 
651
            SmartServerResponse(('nobranch', 'location is a repository')),
 
652
            request.execute(''))
471
653
 
472
654
 
473
655
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
1597
1779
        self.assertEqual(SmartServerResponse(('ok',)), response)
1598
1780
 
1599
1781
 
 
1782
class TestSmartServerVfsGet(tests.TestCaseWithMemoryTransport):
 
1783
 
 
1784
    def test_unicode_path(self):
 
1785
        """VFS requests expect unicode paths to be escaped."""
 
1786
        filename = u'foo\N{INTERROBANG}'
 
1787
        filename_escaped = urlutils.escape(filename)
 
1788
        backing = self.get_transport()
 
1789
        request = smart.vfs.GetRequest(backing)
 
1790
        backing.put_bytes_non_atomic(filename_escaped, 'contents')
 
1791
        self.assertEqual(SmartServerResponse(('ok', ), 'contents'),
 
1792
            request.execute(filename_escaped))
 
1793
 
 
1794
 
1600
1795
class TestHandlers(tests.TestCase):
1601
1796
    """Tests for the request.request_handlers object."""
1602
1797
 
1652
1847
            smart.bzrdir.SmartServerRequestOpenBranch)
1653
1848
        self.assertHandlerEqual('BzrDir.open_branchV2',
1654
1849
            smart.bzrdir.SmartServerRequestOpenBranchV2)
 
1850
        self.assertHandlerEqual('BzrDir.open_branchV3',
 
1851
            smart.bzrdir.SmartServerRequestOpenBranchV3)
1655
1852
        self.assertHandlerEqual('PackRepository.autopack',
1656
1853
            smart.packrepository.SmartServerPackRepositoryAutopack)
1657
1854
        self.assertHandlerEqual('Repository.gather_stats',