← 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: Martin Pool
  • Date: 2009-08-14 12:08:08 UTC
  • mto: This revision was merged to the branch mainline in revision 4614.
  • Revision ID: mbp@sourcefrog.net-20090814120808-4gvx7fhg44z29cj3
Use platform(aliased=1)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_bundle.py
1
 
# Copyright (C) 2005-2010 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
114
114
            ie = InventoryDirectory(file_id, name, parent_id)
115
115
        elif kind == 'file':
116
116
            ie = InventoryFile(file_id, name, parent_id)
117
 
            ie.text_sha1 = text_sha_1
118
 
            ie.text_size = text_size
119
117
        elif kind == 'symlink':
120
118
            ie = InventoryLink(file_id, name, parent_id)
121
119
        else:
122
120
            raise errors.BzrError('unknown kind %r' % kind)
 
121
        ie.text_sha1 = text_sha_1
 
122
        ie.text_size = text_size
123
123
        return ie
124
124
 
125
125
    def add_dir(self, file_id, path):
1039
1039
        bundle = read_bundle(self.create_bundle_text('null:', 'rev2')[0])
1040
1040
        repo = self.make_repository('repo', format='dirstate-with-subtree')
1041
1041
        bundle.install_revisions(repo)
1042
 
        inv_text = repo._get_inventory_xml('rev2')
 
1042
        inv_text = repo.get_inventory_xml('rev2')
1043
1043
        self.assertNotContainsRe(inv_text, 'format="5"')
1044
1044
        self.assertContainsRe(inv_text, 'format="7"')
1045
1045
 
1065
1065
 
1066
1066
    def test_inv_hash_across_serializers(self):
1067
1067
        repo = self.make_repo_with_installed_revisions()
1068
 
        recorded_inv_sha1 = repo.get_revision('rev2').inventory_sha1
1069
 
        xml = repo._get_inventory_xml('rev2')
 
1068
        recorded_inv_sha1 = repo.get_inventory_sha1('rev2')
 
1069
        xml = repo.get_inventory_xml('rev2')
1070
1070
        self.assertEqual(osutils.sha_string(xml), recorded_inv_sha1)
1071
1071
 
1072
1072
    def test_across_models_incompatible(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