1
# Copyright (C) 2007 Canonical Ltd
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.
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.
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
17
"""Tests for HTTP transports and servers implementations.
19
(transport, server) implementations tested here are supplied by
20
HTTPTestProviderAdapter. Note that a server is characterized by a request
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.
36
from bzrlib.tests import (
40
from bzrlib.transport.http._urllib import HttpTransport_urllib
44
from bzrlib.transport.http._pycurl import PyCurlTransport
46
except errors.DependencyNotPresent:
47
pycurl_present = False
49
class HTTPImplementationsTestProviderAdapter(tests.TestScenarioApplier):
52
transport_scenarios = [('urllib',
53
dict(_transport=HttpTransport_urllib,
54
_server=http_server.HttpServer_urllib,
55
_qualified_prefix='http+urllib',
58
transport_scenarios.append(
59
('pycurl', dict(_transport=PyCurlTransport,
60
_server=http_server.HttpServer_PyCurl,
61
_qualified_prefix='http+pycurl',
63
self.scenarios = transport_scenarios
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))
75
class TestHttpTransportUrls(tests.TestCase):
76
"""Test the http urls."""
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')
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,
93
'http://http://bazaar-vcs.org/bzr/bzr.dev/')
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')
102
def test_http_impl_urls(self):
103
"""There are servers which ask for particular clients to connect"""
104
server = self._server()
107
url = server.get_url()
108
self.assertTrue(url.startswith('%s://' % self._qualified_prefix))
113
class TestHttpConnections(http_utils.TestCaseWithWebserver):
114
"""Test the http connections."""
117
http_utils.TestCaseWithWebserver.setUp(self)
118
self.build_tree(['foo/', 'foo/bar'], line_endings='binary',
119
transport=self.get_transport())
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/')
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/')
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(
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)
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)
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()
161
socket.setdefaulttimeout(2)
163
s.bind(('localhost', 0))
164
t = self._transport('http://%s:%s/' % s.getsockname())
165
self.assertRaises(errors.ConnectionError, t.has, 'foo/bar')
167
socket.setdefaulttimeout(default_timeout)
170
class TestPost(tests.TestCase):
172
def test_post_body_is_received(self):
173
server = http_utils.RecordingServer(expect_body_tail='end-of-body')
175
self.addCleanup(server.tearDown)
176
scheme = self._qualified_prefix
177
url = '%s://%s:%s/' % (scheme, server.host, server.port)
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')
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.
190
server.received_bytes.endswith('\r\n\r\nabc def end-of-body'))