← 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: Ian Clatworthy
  • Date: 2009-09-02 16:03:51 UTC
  • mto: (4634.39.1 pdf-chm-docs)
  • mto: This revision was merged to the branch mainline in revision 4689.
  • Revision ID: ian.clatworthy@canonical.com-20090902160351-sxptcz3ttc1aencw
first cut at pdf docs via sphinx

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,
40
39
    )
41
40
from bzrlib.branch import Branch, BranchReferenceFormat
42
41
import bzrlib.smart.branch
43
42
import bzrlib.smart.bzrdir, bzrlib.smart.bzrdir as smart_dir
44
43
import bzrlib.smart.packrepository
45
44
import bzrlib.smart.repository
46
 
import bzrlib.smart.vfs
47
45
from bzrlib.smart.request import (
48
46
    FailedSmartServerResponse,
49
47
    SmartServerRequest,
53
51
from bzrlib.tests import (
54
52
    split_suite_by_re,
55
53
    )
56
 
from bzrlib.transport import chroot, get_transport, local, memory
 
54
from bzrlib.transport import chroot, get_transport
57
55
 
58
56
 
59
57
def load_tests(standard_tests, module, loader):
82
80
class TestCaseWithChrootedTransport(tests.TestCaseWithTransport):
83
81
 
84
82
    def setUp(self):
85
 
        self.vfs_transport_factory = memory.MemoryServer
86
83
        tests.TestCaseWithTransport.setUp(self)
87
84
        self._chroot_server = None
88
85
 
90
87
        if self._chroot_server is None:
91
88
            backing_transport = tests.TestCaseWithTransport.get_transport(self)
92
89
            self._chroot_server = chroot.ChrootServer(backing_transport)
93
 
            self.start_server(self._chroot_server)
 
90
            self._chroot_server.setUp()
 
91
            self.addCleanup(self._chroot_server.tearDown)
94
92
        t = get_transport(self._chroot_server.get_url())
95
93
        if relpath is not None:
96
94
            t = t.clone(relpath)
97
95
        return t
98
96
 
99
97
 
100
 
class TestCaseWithSmartMedium(tests.TestCaseWithMemoryTransport):
 
98
class TestCaseWithSmartMedium(tests.TestCaseWithTransport):
101
99
 
102
100
    def setUp(self):
103
101
        super(TestCaseWithSmartMedium, self).setUp()
115
113
        return self.get_transport().get_smart_medium()
116
114
 
117
115
 
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
 
 
137
116
class TestSmartServerResponse(tests.TestCase):
138
117
 
139
118
    def test__eq__(self):
171
150
        self.assertRaises(
172
151
            errors.PathNotChild, request.translate_client_path, 'bar/')
173
152
        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))
186
153
 
187
154
    def test_transport_from_client_path(self):
188
155
        transport = self.get_transport()
427
394
            'False', 'False', 'False', '', '', '', '', 'False')
428
395
 
429
396
 
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
 
 
497
397
class TestSmartServerRequestOpenBranch(TestCaseWithChrootedTransport):
498
398
 
499
399
    def test_no_branch(self):
514
414
 
515
415
    def test_branch_reference(self):
516
416
        """When there is a branch reference, the reference URL is returned."""
517
 
        self.vfs_transport_factory = local.LocalURLServer
518
417
        backing = self.get_transport()
519
418
        request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
520
419
        branch = self.make_branch('branch')
524
423
        self.assertEqual(SmartServerResponse(('ok', reference_url)),
525
424
            request.execute('reference'))
526
425
 
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
 
 
535
426
 
536
427
class TestSmartServerRequestOpenBranchV2(TestCaseWithChrootedTransport):
537
428
 
553
444
 
554
445
    def test_branch_reference(self):
555
446
        """When there is a branch reference, the reference URL is returned."""
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(''))
 
447
        backing = self.get_transport()
 
448
        request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
 
449
        branch = self.make_branch('branch')
 
450
        checkout = branch.create_checkout('reference',lightweight=True)
 
451
        reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
 
452
        self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
 
453
        self.assertEqual(SuccessfulSmartServerResponse(('ref', reference_url)),
 
454
            request.execute('reference'))
 
455
 
 
456
    def test_stacked_branch(self):
 
457
        """Opening a stacked branch does not open the stacked-on branch."""
 
458
        trunk = self.make_branch('trunk')
 
459
        feature = self.make_branch('feature')
 
460
        feature.set_stacked_on_url(trunk.base)
 
461
        opened_branches = []
 
462
        Branch.hooks.install_named_hook('open', opened_branches.append, None)
 
463
        backing = self.get_transport()
 
464
        request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
 
465
        request.setup_jail()
 
466
        try:
 
467
            response = request.execute('feature')
 
468
        finally:
 
469
            request.teardown_jail()
 
470
        expected_format = feature._format.network_name()
 
471
        self.assertEqual(
 
472
            SuccessfulSmartServerResponse(('branch', expected_format)),
 
473
            response)
 
474
        self.assertLength(1, opened_branches)
653
475
 
654
476
 
655
477
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
1779
1601
        self.assertEqual(SmartServerResponse(('ok',)), response)
1780
1602
 
1781
1603
 
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
 
 
1795
1604
class TestHandlers(tests.TestCase):
1796
1605
    """Tests for the request.request_handlers object."""
1797
1606
 
1847
1656
            smart.bzrdir.SmartServerRequestOpenBranch)
1848
1657
        self.assertHandlerEqual('BzrDir.open_branchV2',
1849
1658
            smart.bzrdir.SmartServerRequestOpenBranchV2)
1850
 
        self.assertHandlerEqual('BzrDir.open_branchV3',
1851
 
            smart.bzrdir.SmartServerRequestOpenBranchV3)
1852
1659
        self.assertHandlerEqual('PackRepository.autopack',
1853
1660
            smart.packrepository.SmartServerPackRepositoryAutopack)
1854
1661
        self.assertHandlerEqual('Repository.gather_stats',