128
127
('urllib,http', dict(_activity_server=ActivityHTTPServer,
129
128
_transport=_urllib.HttpTransport_urllib,)),
130
if features.pycurl.available():
131
activity_scenarios.append(
132
('pycurl,http', dict(_activity_server=ActivityHTTPServer,
133
_transport=PyCurlTransport,)),)
131
134
if features.HTTPSServerFeature.available():
135
# FIXME: Until we have a better way to handle self-signed certificates
136
# (like allowing them in a test specific authentication.conf for
137
# example), we need some specialized pycurl/urllib transport for tests.
139
from bzrlib.tests import (
142
class HTTPS_urllib_transport(_urllib.HttpTransport_urllib):
144
def __init__(self, base, _from_transport=None):
145
super(HTTPS_urllib_transport, self).__init__(
146
base, _from_transport=_from_transport,
147
ca_certs=ssl_certs.build_path('ca.crt'))
132
149
activity_scenarios.append(
133
150
('urllib,https', dict(_activity_server=ActivityHTTPSServer,
134
_transport=_urllib.HttpTransport_urllib,)),)
135
if features.pycurl.available():
136
activity_scenarios.append(
137
('pycurl,http', dict(_activity_server=ActivityHTTPServer,
138
_transport=PyCurlTransport,)),)
139
if features.HTTPSServerFeature.available():
140
from bzrlib.tests import (
143
# FIXME: Until we have a better way to handle self-signed
144
# certificates (like allowing them in a test specific
145
# authentication.conf for example), we need some specialized pycurl
146
# transport for tests.
151
_transport=HTTPS_urllib_transport,)),)
152
if features.pycurl.available():
147
153
class HTTPS_pycurl_transport(PyCurlTransport):
149
155
def __init__(self, base, _from_transport=None):
378
384
_transport = property(_get_pycurl_maybe)
381
class TestHttpUrls(tests.TestCase):
383
# TODO: This should be moved to authorization tests once they
386
def test_url_parsing(self):
388
url = http.extract_auth('http://example.com', f)
389
self.assertEqual('http://example.com', url)
390
self.assertEqual(0, len(f.credentials))
391
url = http.extract_auth(
392
'http://user:pass@example.com/bzr/bzr.dev', f)
393
self.assertEqual('http://example.com/bzr/bzr.dev', url)
394
self.assertEqual(1, len(f.credentials))
395
self.assertEqual([None, 'example.com', 'user', 'pass'],
399
387
class TestHttpTransportUrls(tests.TestCase):
400
388
"""Test the http urls."""
1157
1145
protocol_version=self._protocol_version)
1159
1147
def setUp(self):
1160
http_utils.TestCaseWithWebserver.setUp(self)
1148
super(TestLimitedRangeRequestServer, self).setUp()
1161
1149
# We need to manipulate ranges that correspond to real chunks in the
1162
1150
# response, so we build a content appropriately.
1163
1151
filler = ''.join(['abcdefghij' for x in range(102)])
1905
1893
server._url_protocol = self._url_protocol
1908
def test_open_bzrdir(self):
1896
def test_open_controldir(self):
1909
1897
branch = self.make_branch('relpath')
1910
1898
url = self.http_server.get_url() + 'relpath'
1911
bd = bzrdir.BzrDir.open(url)
1899
bd = controldir.ControlDir.open(url)
1912
1900
self.addCleanup(bd.transport.disconnect)
1913
1901
self.assertIsInstance(bd, _mod_remote.RemoteBzrDir)
2119
2107
def setUp(self):
2120
tests.TestCase.setUp(self)
2121
2108
self.server = self._activity_server(self._protocol_version)
2122
2109
self.server.start_server()
2110
self.addCleanup(self.server.stop_server)
2123
2111
_activities = {} # Don't close over self and create a cycle
2124
2112
def report_activity(t, bytes, direction):
2125
2113
count = _activities.get(direction, 0)
2127
2115
_activities[direction] = count
2128
2116
self.activities = _activities
2130
2117
# We override at class level because constructors may propagate the
2131
2118
# bound method and render instance overriding ineffective (an
2132
2119
# alternative would be to define a specific ui factory instead...)
2133
2120
self.overrideAttr(self._transport, '_report_activity', report_activity)
2134
self.addCleanup(self.server.stop_server)
2136
2122
def get_transport(self):
2137
2123
t = self._transport(self.server.get_url())