/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 bzrlib/tests/test_http_implementations.py

Merge bzr.dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2007 Canonical Ltd
 
2
#
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
16
 
 
17
"""Tests for HTTP transports and servers implementations.
 
18
 
 
19
(transport, server) implementations tested here are supplied by
 
20
HTTPTestProviderAdapter. Note that a server is characterized by a request
 
21
handler class.
 
22
 
 
23
Transport implementations are normally tested via
 
24
test_transport_implementations. The tests here are about the variations in HTTP
 
25
protocol implementation to guarantee the robustness of our transports.
 
26
"""
 
27
 
 
28
import socket
 
29
 
 
30
import bzrlib
 
31
from bzrlib import (
 
32
    errors,
 
33
    tests,
 
34
    transport,
 
35
    )
 
36
from bzrlib.tests import (
 
37
    http_server,
 
38
    http_utils,
 
39
    )
 
40
from bzrlib.transport.http._urllib import HttpTransport_urllib
 
41
 
 
42
 
 
43
try:
 
44
    from bzrlib.transport.http._pycurl import PyCurlTransport
 
45
    pycurl_present = True
 
46
except errors.DependencyNotPresent:
 
47
    pycurl_present = False
 
48
 
 
49
class HTTPImplementationsTestProviderAdapter(tests.TestScenarioApplier):
 
50
 
 
51
    def __init__(self):
 
52
        transport_scenarios = [('urllib',
 
53
                                dict(_transport=HttpTransport_urllib,
 
54
                                     _server=http_server.HttpServer_urllib,
 
55
                                     _qualified_prefix='http+urllib',
 
56
                                     )),]
 
57
        if pycurl_present:
 
58
            transport_scenarios.append(
 
59
                ('pycurl', dict(_transport=PyCurlTransport,
 
60
                                _server=http_server.HttpServer_PyCurl,
 
61
                                _qualified_prefix='http+pycurl',
 
62
                                )))
 
63
        self.scenarios = transport_scenarios
 
64
 
 
65
 
 
66
def load_tests(standard_tests, module, loader):
 
67
    """Multiply tests for http clients and protocol versions."""
 
68
    adapter = HTTPImplementationsTestProviderAdapter()
 
69
    result = loader.suiteClass()
 
70
    for test in tests.iter_suite_tests(standard_tests):
 
71
        result.addTests(adapter.adapt(test))
 
72
    return result
 
73
 
 
74
 
 
75
class TestHttpTransportUrls(tests.TestCase):
 
76
    """Test the http urls."""
 
77
 
 
78
    def test_abs_url(self):
 
79
        """Construction of absolute http URLs"""
 
80
        t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
 
81
        eq = self.assertEqualDiff
 
82
        eq(t.abspath('.'), 'http://bazaar-vcs.org/bzr/bzr.dev')
 
83
        eq(t.abspath('foo/bar'), 'http://bazaar-vcs.org/bzr/bzr.dev/foo/bar')
 
84
        eq(t.abspath('.bzr'), 'http://bazaar-vcs.org/bzr/bzr.dev/.bzr')
 
85
        eq(t.abspath('.bzr/1//2/./3'),
 
86
           'http://bazaar-vcs.org/bzr/bzr.dev/.bzr/1/2/3')
 
87
 
 
88
    def test_invalid_http_urls(self):
 
89
        """Trap invalid construction of urls"""
 
90
        t = self._transport('http://bazaar-vcs.org/bzr/bzr.dev/')
 
91
        self.assertRaises(errors.InvalidURL,
 
92
                          self._transport,
 
93
                          'http://http://bazaar-vcs.org/bzr/bzr.dev/')
 
94
 
 
95
    def test_http_root_urls(self):
 
96
        """Construction of URLs from server root"""
 
97
        t = self._transport('http://bzr.ozlabs.org/')
 
98
        eq = self.assertEqualDiff
 
99
        eq(t.abspath('.bzr/tree-version'),
 
100
           'http://bzr.ozlabs.org/.bzr/tree-version')
 
101
 
 
102
    def test_http_impl_urls(self):
 
103
        """There are servers which ask for particular clients to connect"""
 
104
        server = self._server()
 
105
        try:
 
106
            server.setUp()
 
107
            url = server.get_url()
 
108
            self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
 
109
        finally:
 
110
            server.tearDown()
 
111
 
 
112
 
 
113
class TestHttpConnections(http_utils.TestCaseWithWebserver):
 
114
    """Test the http connections."""
 
115
 
 
116
    def setUp(self):
 
117
        http_utils.TestCaseWithWebserver.setUp(self)
 
118
        self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
 
119
                        transport=self.get_transport())
 
120
 
 
121
    def test_http_has(self):
 
122
        server = self.get_readonly_server()
 
123
        t = self._transport(server.get_url())
 
124
        self.assertEqual(t.has('foo/bar'), True)
 
125
        self.assertEqual(len(server.logs), 1)
 
126
        self.assertContainsRe(server.logs[0],
 
127
            r'"HEAD /foo/bar HTTP/1.." (200|302) - "-" "bzr/')
 
128
 
 
129
    def test_http_has_not_found(self):
 
130
        server = self.get_readonly_server()
 
131
        t = self._transport(server.get_url())
 
132
        self.assertEqual(t.has('not-found'), False)
 
133
        self.assertContainsRe(server.logs[1],
 
134
            r'"HEAD /not-found HTTP/1.." 404 - "-" "bzr/')
 
135
 
 
136
    def test_http_get(self):
 
137
        server = self.get_readonly_server()
 
138
        t = self._transport(server.get_url())
 
139
        fp = t.get('foo/bar')
 
140
        self.assertEqualDiff(
 
141
            fp.read(),
 
142
            'contents of foo/bar\n')
 
143
        self.assertEqual(len(server.logs), 1)
 
144
        self.assertTrue(server.logs[0].find(
 
145
            '"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
 
146
            % bzrlib.__version__) > -1)
 
147
 
 
148
    def test_get_smart_medium(self):
 
149
        # For HTTP, get_smart_medium should return the transport object.
 
150
        server = self.get_readonly_server()
 
151
        http_transport = self._transport(server.get_url())
 
152
        medium = http_transport.get_smart_medium()
 
153
        self.assertIs(medium, http_transport)
 
154
 
 
155
    def test_has_on_bogus_host(self):
 
156
        # Get a free address and don't 'accept' on it, so that we
 
157
        # can be sure there is no http handler there, but set a
 
158
        # reasonable timeout to not slow down tests too much.
 
159
        default_timeout = socket.getdefaulttimeout()
 
160
        try:
 
161
            socket.setdefaulttimeout(2)
 
162
            s = socket.socket()
 
163
            s.bind(('localhost', 0))
 
164
            t = self._transport('http://%s:%s/' % s.getsockname())
 
165
            self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
 
166
        finally:
 
167
            socket.setdefaulttimeout(default_timeout)
 
168
 
 
169
 
 
170
class TestPost(tests.TestCase):
 
171
 
 
172
    def test_post_body_is_received(self):
 
173
        server = http_utils.RecordingServer(expect_body_tail='end-of-body')
 
174
        server.setUp()
 
175
        self.addCleanup(server.tearDown)
 
176
        scheme = self._qualified_prefix
 
177
        url = '%s://%s:%s/' % (scheme, server.host, server.port)
 
178
        try:
 
179
            http_transport = transport.get_transport(url)
 
180
        except errors.UnsupportedProtocol:
 
181
            raise tests.TestSkipped('%s not available' % scheme)
 
182
        code, response = http_transport._post('abc def end-of-body')
 
183
        self.assertTrue(
 
184
            server.received_bytes.startswith('POST /.bzr/smart HTTP/1.'))
 
185
        self.assertTrue('content-length: 19\r' in server.received_bytes.lower())
 
186
        # The transport should not be assuming that the server can accept
 
187
        # chunked encoding the first time it connects, because HTTP/1.1, so we
 
188
        # check for the literal string.
 
189
        self.assertTrue(
 
190
            server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))
 
191
 
 
192