← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_bundle.py

  • Committer: Benoît Pierre
  • Date: 2009-11-02 22:24:29 UTC
  • mto: (4634.96.1 integration-2.0)
  • mto: This revision was merged to the branch mainline in revision 4798.
  • Revision ID: benoit.pierre@gmail.com-20091102222429-xqdyo6n8odh3xbbd
Small fix for handling of short option names in shellcomplete_on_options.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_bundle.py
1
 
# Copyright (C) 2005-2011 Canonical Ltd
 
1
# Copyright (C) 2004, 2005, 2006, 2007 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
27
27
    inventory,
28
28
    merge,
29
29
    osutils,
 
30
    repository,
30
31
    revision as _mod_revision,
31
 
    symbol_versioning,
32
32
    tests,
33
33
    treebuilder,
34
34
    )
35
35
from bzrlib.bundle import read_mergeable_from_url
36
36
from bzrlib.bundle.apply_bundle import install_bundle, merge_bundle
37
37
from bzrlib.bundle.bundle_data import BundleTree
 
38
from bzrlib.bzrdir import BzrDir
38
39
from bzrlib.directory_service import directories
39
40
from bzrlib.bundle.serializer import write_bundle, read_bundle, v09, v4
40
41
from bzrlib.bundle.serializer.v08 import BundleSerializerV08
41
42
from bzrlib.bundle.serializer.v09 import BundleSerializerV09
42
43
from bzrlib.bundle.serializer.v4 import BundleSerializerV4
 
44
from bzrlib.branch import Branch
43
45
from bzrlib.repofmt import knitrepo
44
46
from bzrlib.tests import (
45
47
    test_read_bundle,
46
48
    test_commit,
47
49
    )
48
50
from bzrlib.transform import TreeTransform
49
 
from bzrlib.tests import (
50
 
    features,
51
 
    )
52
51
 
53
52
 
54
53
def get_text(vf, key):
68
67
 
69
68
 
70
69
class MockTree(object):
71
 
 
72
70
    def __init__(self):
73
71
        from bzrlib.inventory import InventoryDirectory, ROOT_ID
74
72
        object.__init__(self)
79
77
 
80
78
    inventory = property(lambda x:x)
81
79
 
82
 
    def all_file_ids(self):
83
 
        return set(self.paths.keys())
 
80
    def __iter__(self):
 
81
        return self.paths.iterkeys()
84
82
 
85
83
    def __getitem__(self, file_id):
86
84
        if file_id == self.root.file_id:
116
114
            ie = InventoryDirectory(file_id, name, parent_id)
117
115
        elif kind == 'file':
118
116
            ie = InventoryFile(file_id, name, parent_id)
119
 
            ie.text_sha1 = text_sha_1
120
 
            ie.text_size = text_size
121
117
        elif kind == 'symlink':
122
118
            ie = InventoryLink(file_id, name, parent_id)
123
119
        else:
124
120
            raise errors.BzrError('unknown kind %r' % kind)
 
121
        ie.text_sha1 = text_sha_1
 
122
        ie.text_size = text_size
125
123
        return ie
126
124
 
127
125
    def add_dir(self, file_id, path):
147
145
        result.seek(0,0)
148
146
        return result
149
147
 
150
 
    def get_file_revision(self, file_id):
151
 
        return self.inventory[file_id].revision
152
 
 
153
148
    def contents_stats(self, file_id):
154
149
        if file_id not in self.contents:
155
150
            return None, None
497
492
                                 % (ancestor,))
498
493
 
499
494
                # Now check that the file contents are all correct
500
 
                for inventory_id in old.all_file_ids():
 
495
                for inventory_id in old:
501
496
                    try:
502
497
                        old_file = old.get_file(inventory_id)
503
498
                    except errors.NoSuchFile:
511
506
                old.unlock()
512
507
        if not _mod_revision.is_null(rev_id):
513
508
            rh = self.b1.revision_history()
514
 
            self.applyDeprecated(symbol_versioning.deprecated_in((2, 4, 0)),
515
 
                tree.branch.set_revision_history, rh[:rh.index(rev_id)+1])
 
509
            tree.branch.set_revision_history(rh[:rh.index(rev_id)+1])
516
510
            tree.update()
517
511
            delta = tree.changes_from(self.b1.repository.revision_tree(rev_id))
518
512
            self.assertFalse(delta.has_changed(),
685
679
    def _test_symlink_bundle(self, link_name, link_target, new_link_target):
686
680
        link_id = 'link-1'
687
681
 
688
 
        self.requireFeature(features.SymlinkFeature)
 
682
        self.requireFeature(tests.SymlinkFeature)
689
683
        self.tree1 = self.make_branch_and_tree('b1')
690
684
        self.b1 = self.tree1.branch
691
685
 
732
726
        self._test_symlink_bundle('link', 'bar/foo', 'mars')
733
727
 
734
728
    def test_unicode_symlink_bundle(self):
735
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
729
        self.requireFeature(tests.UnicodeFilenameFeature)
736
730
        self._test_symlink_bundle(u'\N{Euro Sign}link',
737
731
                                  u'bar/\N{Euro Sign}foo',
738
732
                                  u'mars\N{Euro Sign}')
839
833
        return bundle_file.getvalue()
840
834
 
841
835
    def test_unicode_bundle(self):
842
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
836
        self.requireFeature(tests.UnicodeFilenameFeature)
843
837
        # Handle international characters
844
838
        os.mkdir('b1')
845
839
        f = open(u'b1/with Dod\N{Euro Sign}', 'wb')
1045
1039
        bundle = read_bundle(self.create_bundle_text('null:', 'rev2')[0])
1046
1040
        repo = self.make_repository('repo', format='dirstate-with-subtree')
1047
1041
        bundle.install_revisions(repo)
1048
 
        inv_text = repo._get_inventory_xml('rev2')
 
1042
        inv_text = repo.get_inventory_xml('rev2')
1049
1043
        self.assertNotContainsRe(inv_text, 'format="5"')
1050
1044
        self.assertContainsRe(inv_text, 'format="7"')
1051
1045
 
1071
1065
 
1072
1066
    def test_inv_hash_across_serializers(self):
1073
1067
        repo = self.make_repo_with_installed_revisions()
1074
 
        recorded_inv_sha1 = repo.get_revision('rev2').inventory_sha1
1075
 
        xml = repo._get_inventory_xml('rev2')
 
1068
        recorded_inv_sha1 = repo.get_inventory_sha1('rev2')
 
1069
        xml = repo.get_inventory_xml('rev2')
1076
1070
        self.assertEqual(osutils.sha_string(xml), recorded_inv_sha1)
1077
1071
 
1078
1072
    def test_across_models_incompatible(self):
1418
1412
        branch = tree_a.branch
1419
1413
        repo_a = branch.repository
1420
1414
        tree_a.commit("base", allow_pointless=True, rev_id='A')
1421
 
        self.assertFalse(branch.repository.has_signature_for_revision_id('A'))
 
1415
        self.failIf(branch.repository.has_signature_for_revision_id('A'))
1422
1416
        try:
1423
1417
            from bzrlib.testament import Testament
1424
1418
            # monkey patch gpg signing mechanism
1448
1442
        install_bundle(repo_b, serializer.read(s))
1449
1443
 
1450
1444
 
 
1445
class V4WeaveBundleTester(V4BundleTester):
 
1446
 
 
1447
    def bzrdir_format(self):
 
1448
        return 'metaweave'
 
1449
 
 
1450
 
1451
1451
class V4_2aBundleTester(V4BundleTester):
1452
1452
 
1453
1453
    def bzrdir_format(self):
1820
1820
            def look_up(self, name, url):
1821
1821
                return 'source'
1822
1822
        directories.register('foo:', FooService, 'Testing directory service')
1823
 
        self.addCleanup(directories.remove, 'foo:')
 
1823
        self.addCleanup(lambda: directories.remove('foo:'))
1824
1824
        self.build_tree_contents([('./foo:bar', out.getvalue())])
1825
1825
        self.assertRaises(errors.NotABundle, read_mergeable_from_url,
1826
1826
                          'foo:bar')
1830
1830
        """
1831
1831
        from bzrlib.tests.blackbox.test_push import RedirectingMemoryServer
1832
1832
        server = RedirectingMemoryServer()
1833
 
        self.start_server(server)
 
1833
        server.setUp()
1834
1834
        url = server.get_url() + 'infinite-loop'
 
1835
        self.addCleanup(server.tearDown)
1835
1836
        self.assertRaises(errors.NotABundle, read_mergeable_from_url, url)
1836
1837
 
1837
1838
    def test_smart_server_connection_reset(self):
1840
1841
        """
1841
1842
        # Instantiate a server that will provoke a ConnectionReset
1842
1843
        sock_server = _DisconnectingTCPServer()
1843
 
        self.start_server(sock_server)
 
1844
        sock_server.setUp()
 
1845
        self.addCleanup(sock_server.tearDown)
1844
1846
        # We don't really care what the url is since the server will close the
1845
1847
        # connection without interpreting it
1846
1848
        url = sock_server.get_url()
1850
1852
class _DisconnectingTCPServer(object):
1851
1853
    """A TCP server that immediately closes any connection made to it."""
1852
1854
 
1853
 
    def start_server(self):
 
1855
    def setUp(self):
1854
1856
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1855
1857
        self.sock.bind(('127.0.0.1', 0))
1856
1858
        self.sock.listen(1)
1868
1870
    def get_url(self):
1869
1871
        return 'bzr://127.0.0.1:%d/' % (self.port,)
1870
1872
 
1871
 
    def stop_server(self):
 
1873
    def tearDown(self):
1872
1874
        try:
1873
1875
            # make sure the thread dies by connecting to the listening socket,
1874
1876
            # just in case the test failed to do so.
1879
1881
            pass
1880
1882
        self.sock.close()
1881
1883
        self.thread.join()
 
1884