/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to breezy/tests/test_bundle.py

  • Committer: Jelmer Vernooij
  • Date: 2018-11-13 01:08:02 UTC
  • mto: This revision was merged to the branch mainline in revision 7305.
  • Revision ID: jelmer@jelmer.uk-20181113010802-fw4f16k9hd38u2fa
Move mergeable functions to breezy.mergeable.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_bundle.py
17
17
import bz2
18
18
from io import BytesIO
19
19
import os
20
 
try:
21
 
    import socketserver
22
 
except ImportError:
23
 
    import SocketServer as socketserver
24
20
import sys
25
21
 
26
22
from .. import (
36
32
    bzrdir,
37
33
    inventory,
38
34
    )
39
 
from ..bundle import read_mergeable_from_url
40
35
from ..bundle.apply_bundle import install_bundle, merge_bundle
41
36
from ..bundle.bundle_data import BundleTree
42
 
from ..directory_service import directories
43
37
from ..bundle.serializer import write_bundle, read_bundle, v09, v4
44
38
from ..bundle.serializer.v08 import BundleSerializerV08
45
39
from ..bundle.serializer.v09 import BundleSerializerV09
48
42
from . import (
49
43
    features,
50
44
    test_commit,
51
 
    test_read_bundle,
52
 
    test_server,
53
45
    )
54
46
from ..transform import TreeTransform
55
47
 
1847
1839
            'info', None, None), record)
1848
1840
        self.assertRaises(errors.BadBundle, next, record_iter)
1849
1841
 
1850
 
 
1851
 
class TestReadMergeableFromUrl(tests.TestCaseWithTransport):
1852
 
 
1853
 
    def test_read_mergeable_skips_local(self):
1854
 
        """A local bundle named like the URL should not be read.
1855
 
        """
1856
 
        out, wt = test_read_bundle.create_bundle_file(self)
1857
 
        class FooService(object):
1858
 
            """A directory service that always returns source"""
1859
 
 
1860
 
            def look_up(self, name, url):
1861
 
                return 'source'
1862
 
        directories.register('foo:', FooService, 'Testing directory service')
1863
 
        self.addCleanup(directories.remove, 'foo:')
1864
 
        self.build_tree_contents([('./foo:bar', out.getvalue())])
1865
 
        self.assertRaises(errors.NotABundle, read_mergeable_from_url,
1866
 
                          'foo:bar')
1867
 
 
1868
 
    def test_infinite_redirects_are_not_a_bundle(self):
1869
 
        """If a URL causes TooManyRedirections then NotABundle is raised.
1870
 
        """
1871
 
        from .blackbox.test_push import RedirectingMemoryServer
1872
 
        server = RedirectingMemoryServer()
1873
 
        self.start_server(server)
1874
 
        url = server.get_url() + 'infinite-loop'
1875
 
        self.assertRaises(errors.NotABundle, read_mergeable_from_url, url)
1876
 
 
1877
 
    def test_smart_server_connection_reset(self):
1878
 
        """If a smart server connection fails during the attempt to read a
1879
 
        bundle, then the ConnectionReset error should be propagated.
1880
 
        """
1881
 
        # Instantiate a server that will provoke a ConnectionReset
1882
 
        sock_server = DisconnectingServer()
1883
 
        self.start_server(sock_server)
1884
 
        # We don't really care what the url is since the server will close the
1885
 
        # connection without interpreting it
1886
 
        url = sock_server.get_url()
1887
 
        self.assertRaises(errors.ConnectionReset, read_mergeable_from_url, url)
1888
 
 
1889
 
 
1890
 
class DisconnectingHandler(socketserver.BaseRequestHandler):
1891
 
    """A request handler that immediately closes any connection made to it."""
1892
 
 
1893
 
    def handle(self):
1894
 
        self.request.close()
1895
 
 
1896
 
 
1897
 
class DisconnectingServer(test_server.TestingTCPServerInAThread):
1898
 
 
1899
 
    def __init__(self):
1900
 
        super(DisconnectingServer, self).__init__(
1901
 
            ('127.0.0.1', 0),
1902
 
            test_server.TestingTCPServer,
1903
 
            DisconnectingHandler)
1904
 
 
1905
 
    def get_url(self):
1906
 
        """Return the url of the server"""
1907
 
        return "bzr://%s:%d/" % self.server.server_address