← 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: Andrew Bennetts
  • Date: 2010-10-13 00:26:41 UTC
  • mto: This revision was merged to the branch mainline in revision 5498.
  • Revision ID: andrew.bennetts@canonical.com-20101013002641-9tlh9k89mlj1666m
Keep docs-plain working.

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) 2005-2010 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
16
16
 
17
17
from cStringIO import StringIO
18
18
import os
19
 
import SocketServer
 
19
import socket
20
20
import sys
 
21
import threading
21
22
 
22
23
from bzrlib import (
23
24
    bzrdir,
26
27
    inventory,
27
28
    merge,
28
29
    osutils,
 
30
    repository,
29
31
    revision as _mod_revision,
30
32
    tests,
31
33
    treebuilder,
33
35
from bzrlib.bundle import read_mergeable_from_url
34
36
from bzrlib.bundle.apply_bundle import install_bundle, merge_bundle
35
37
from bzrlib.bundle.bundle_data import BundleTree
 
38
from bzrlib.bzrdir import BzrDir
36
39
from bzrlib.directory_service import directories
37
40
from bzrlib.bundle.serializer import write_bundle, read_bundle, v09, v4
38
41
from bzrlib.bundle.serializer.v08 import BundleSerializerV08
39
42
from bzrlib.bundle.serializer.v09 import BundleSerializerV09
40
43
from bzrlib.bundle.serializer.v4 import BundleSerializerV4
 
44
from bzrlib.branch import Branch
41
45
from bzrlib.repofmt import knitrepo
42
46
from bzrlib.tests import (
43
 
    features,
 
47
    test_read_bundle,
44
48
    test_commit,
45
 
    test_read_bundle,
46
 
    test_server,
47
49
    )
48
50
from bzrlib.transform import TreeTransform
49
51
 
65
67
 
66
68
 
67
69
class MockTree(object):
68
 
 
69
70
    def __init__(self):
70
71
        from bzrlib.inventory import InventoryDirectory, ROOT_ID
71
72
        object.__init__(self)
76
77
 
77
78
    inventory = property(lambda x:x)
78
79
 
79
 
    def all_file_ids(self):
80
 
        return set(self.paths.keys())
 
80
    def __iter__(self):
 
81
        return self.paths.iterkeys()
81
82
 
82
83
    def __getitem__(self, file_id):
83
84
        if file_id == self.root.file_id:
144
145
        result.seek(0,0)
145
146
        return result
146
147
 
147
 
    def get_file_revision(self, file_id):
148
 
        return self.inventory[file_id].revision
149
 
 
150
148
    def contents_stats(self, file_id):
151
149
        if file_id not in self.contents:
152
150
            return None, None
494
492
                                 % (ancestor,))
495
493
 
496
494
                # Now check that the file contents are all correct
497
 
                for inventory_id in old.all_file_ids():
 
495
                for inventory_id in old:
498
496
                    try:
499
497
                        old_file = old.get_file(inventory_id)
500
498
                    except errors.NoSuchFile:
507
505
                new.unlock()
508
506
                old.unlock()
509
507
        if not _mod_revision.is_null(rev_id):
510
 
            tree.branch.generate_revision_history(rev_id)
 
508
            rh = self.b1.revision_history()
 
509
            tree.branch.set_revision_history(rh[:rh.index(rev_id)+1])
511
510
            tree.update()
512
511
            delta = tree.changes_from(self.b1.repository.revision_tree(rev_id))
513
512
            self.assertFalse(delta.has_changed(),
680
679
    def _test_symlink_bundle(self, link_name, link_target, new_link_target):
681
680
        link_id = 'link-1'
682
681
 
683
 
        self.requireFeature(features.SymlinkFeature)
 
682
        self.requireFeature(tests.SymlinkFeature)
684
683
        self.tree1 = self.make_branch_and_tree('b1')
685
684
        self.b1 = self.tree1.branch
686
685
 
727
726
        self._test_symlink_bundle('link', 'bar/foo', 'mars')
728
727
 
729
728
    def test_unicode_symlink_bundle(self):
730
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
729
        self.requireFeature(tests.UnicodeFilenameFeature)
731
730
        self._test_symlink_bundle(u'\N{Euro Sign}link',
732
731
                                  u'bar/\N{Euro Sign}foo',
733
732
                                  u'mars\N{Euro Sign}')
834
833
        return bundle_file.getvalue()
835
834
 
836
835
    def test_unicode_bundle(self):
837
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
836
        self.requireFeature(tests.UnicodeFilenameFeature)
838
837
        # Handle international characters
839
838
        os.mkdir('b1')
840
839
        f = open(u'b1/with Dod\N{Euro Sign}', 'wb')
1413
1412
        branch = tree_a.branch
1414
1413
        repo_a = branch.repository
1415
1414
        tree_a.commit("base", allow_pointless=True, rev_id='A')
1416
 
        self.assertFalse(branch.repository.has_signature_for_revision_id('A'))
 
1415
        self.failIf(branch.repository.has_signature_for_revision_id('A'))
1417
1416
        try:
1418
1417
            from bzrlib.testament import Testament
1419
1418
            # monkey patch gpg signing mechanism
1420
1419
            bzrlib.gpg.GPGStrategy = bzrlib.gpg.LoopbackGPGStrategy
1421
 
            new_config = test_commit.MustSignConfig()
1422
 
            commit.Commit(config_stack=new_config).commit(message="base",
 
1420
            new_config = test_commit.MustSignConfig(branch)
 
1421
            commit.Commit(config=new_config).commit(message="base",
1423
1422
                                                    allow_pointless=True,
1424
1423
                                                    rev_id='B',
1425
1424
                                                    working_tree=tree_a)
1443
1442
        install_bundle(repo_b, serializer.read(s))
1444
1443
 
1445
1444
 
 
1445
class V4WeaveBundleTester(V4BundleTester):
 
1446
 
 
1447
    def bzrdir_format(self):
 
1448
        return 'metaweave'
 
1449
 
 
1450
 
1446
1451
class V4_2aBundleTester(V4BundleTester):
1447
1452
 
1448
1453
    def bzrdir_format(self):
1834
1839
        bundle, then the ConnectionReset error should be propagated.
1835
1840
        """
1836
1841
        # Instantiate a server that will provoke a ConnectionReset
1837
 
        sock_server = DisconnectingServer()
 
1842
        sock_server = _DisconnectingTCPServer()
1838
1843
        self.start_server(sock_server)
1839
1844
        # We don't really care what the url is since the server will close the
1840
1845
        # connection without interpreting it
1842
1847
        self.assertRaises(errors.ConnectionReset, read_mergeable_from_url, url)
1843
1848
 
1844
1849
 
1845
 
class DisconnectingHandler(SocketServer.BaseRequestHandler):
1846
 
    """A request handler that immediately closes any connection made to it."""
1847
 
 
1848
 
    def handle(self):
1849
 
        self.request.close()
1850
 
 
1851
 
 
1852
 
class DisconnectingServer(test_server.TestingTCPServerInAThread):
1853
 
 
1854
 
    def __init__(self):
1855
 
        super(DisconnectingServer, self).__init__(
1856
 
            ('127.0.0.1', 0),
1857
 
            test_server.TestingTCPServer,
1858
 
            DisconnectingHandler)
 
1850
class _DisconnectingTCPServer(object):
 
1851
    """A TCP server that immediately closes any connection made to it."""
 
1852
 
 
1853
    def start_server(self):
 
1854
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
1855
        self.sock.bind(('127.0.0.1', 0))
 
1856
        self.sock.listen(1)
 
1857
        self.port = self.sock.getsockname()[1]
 
1858
        self.thread = threading.Thread(
 
1859
            name='%s (port %d)' % (self.__class__.__name__, self.port),
 
1860
            target=self.accept_and_close)
 
1861
        self.thread.start()
 
1862
 
 
1863
    def accept_and_close(self):
 
1864
        conn, addr = self.sock.accept()
 
1865
        conn.shutdown(socket.SHUT_RDWR)
 
1866
        conn.close()
1859
1867
 
1860
1868
    def get_url(self):
1861
 
        """Return the url of the server"""
1862
 
        return "bzr://%s:%d/" % self.server.server_address
 
1869
        return 'bzr://127.0.0.1:%d/' % (self.port,)
 
1870
 
 
1871
    def stop_server(self):
 
1872
        try:
 
1873
            # make sure the thread dies by connecting to the listening socket,
 
1874
            # just in case the test failed to do so.
 
1875
            conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
1876
            conn.connect(self.sock.getsockname())
 
1877
            conn.close()
 
1878
        except socket.error:
 
1879
            pass
 
1880
        self.sock.close()
 
1881
        self.thread.join()