← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_http.py

  • Committer: John Arbash Meinel
  • Author(s): Mark Hammond
  • Date: 2008-09-09 17:02:21 UTC
  • mto: This revision was merged to the branch mainline in revision 3697.
  • Revision ID: john@arbash-meinel.com-20080909170221-svim3jw2mrz0amp3
An updated transparent icon for bzr.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_http.py
1
 
# Copyright (C) 2005, 2006, 2007, 2008, 2009 Canonical Ltd
 
1
# Copyright (C) 2005, 2006 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
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Tests for HTTP implementations.
18
18
 
44
44
    ui,
45
45
    urlutils,
46
46
    )
47
 
from bzrlib.symbol_versioning import (
48
 
    deprecated_in,
49
 
    )
50
47
from bzrlib.tests import (
51
48
    http_server,
52
49
    http_utils,
68
65
    pycurl_present = False
69
66
 
70
67
 
 
68
class TransportAdapter(tests.TestScenarioApplier):
 
69
    """Generate the same test for each transport implementation."""
 
70
 
 
71
    def __init__(self):
 
72
        transport_scenarios = [
 
73
            ('urllib', dict(_transport=_urllib.HttpTransport_urllib,
 
74
                            _server=http_server.HttpServer_urllib,
 
75
                            _qualified_prefix='http+urllib',)),
 
76
            ]
 
77
        if pycurl_present:
 
78
            transport_scenarios.append(
 
79
                ('pycurl', dict(_transport=PyCurlTransport,
 
80
                                _server=http_server.HttpServer_PyCurl,
 
81
                                _qualified_prefix='http+pycurl',)))
 
82
        self.scenarios = transport_scenarios
 
83
 
 
84
 
 
85
class TransportProtocolAdapter(TransportAdapter):
 
86
    """Generate the same test for each protocol implementation.
 
87
 
 
88
    In addition to the transport adaptatation that we inherit from.
 
89
    """
 
90
 
 
91
    def __init__(self):
 
92
        super(TransportProtocolAdapter, self).__init__()
 
93
        protocol_scenarios = [
 
94
            ('HTTP/1.0',  dict(_protocol_version='HTTP/1.0')),
 
95
            ('HTTP/1.1',  dict(_protocol_version='HTTP/1.1')),
 
96
            ]
 
97
        self.scenarios = tests.multiply_scenarios(self.scenarios,
 
98
                                                  protocol_scenarios)
 
99
 
 
100
 
 
101
class TransportProtocolAuthenticationAdapter(TransportProtocolAdapter):
 
102
    """Generate the same test for each authentication scheme implementation.
 
103
 
 
104
    In addition to the protocol adaptatation that we inherit from.
 
105
    """
 
106
 
 
107
    def __init__(self):
 
108
        super(TransportProtocolAuthenticationAdapter, self).__init__()
 
109
        auth_scheme_scenarios = [
 
110
            ('basic', dict(_auth_scheme='basic')),
 
111
            ('digest', dict(_auth_scheme='digest')),
 
112
            ]
 
113
 
 
114
        self.scenarios = tests.multiply_scenarios(self.scenarios,
 
115
                                                  auth_scheme_scenarios)
 
116
 
71
117
def load_tests(standard_tests, module, loader):
72
118
    """Multiply tests for http clients and protocol versions."""
 
119
    # one for each transport
 
120
    t_adapter = TransportAdapter()
 
121
    t_classes= (TestHttpTransportRegistration,
 
122
                TestHttpTransportUrls,
 
123
                )
 
124
    is_testing_for_transports = tests.condition_isinstance(t_classes)
 
125
 
 
126
    # multiplied by one for each protocol version
 
127
    tp_adapter = TransportProtocolAdapter()
 
128
    tp_classes= (SmartHTTPTunnellingTest,
 
129
                 TestDoCatchRedirections,
 
130
                 TestHTTPConnections,
 
131
                 TestHTTPRedirections,
 
132
                 TestHTTPSilentRedirections,
 
133
                 TestLimitedRangeRequestServer,
 
134
                 TestPost,
 
135
                 TestProxyHttpServer,
 
136
                 TestRanges,
 
137
                 TestSpecificRequestHandler,
 
138
                 )
 
139
    is_also_testing_for_protocols = tests.condition_isinstance(tp_classes)
 
140
 
 
141
    # multiplied by one for each authentication scheme
 
142
    tpa_adapter = TransportProtocolAuthenticationAdapter()
 
143
    tpa_classes = (TestAuth,
 
144
                   )
 
145
    is_also_testing_for_authentication = tests.condition_isinstance(
 
146
        tpa_classes)
 
147
 
73
148
    result = loader.suiteClass()
74
 
 
75
 
    # one for each transport implementation
76
 
    t_tests, remaining_tests = tests.split_suite_by_condition(
77
 
        standard_tests, tests.condition_isinstance((
78
 
                TestHttpTransportRegistration,
79
 
                TestHttpTransportUrls,
80
 
                Test_redirected_to,
81
 
                )))
82
 
    transport_scenarios = [
83
 
        ('urllib', dict(_transport=_urllib.HttpTransport_urllib,
84
 
                        _server=http_server.HttpServer_urllib,
85
 
                        _qualified_prefix='http+urllib',)),
86
 
        ]
87
 
    if pycurl_present:
88
 
        transport_scenarios.append(
89
 
            ('pycurl', dict(_transport=PyCurlTransport,
90
 
                            _server=http_server.HttpServer_PyCurl,
91
 
                            _qualified_prefix='http+pycurl',)))
92
 
    tests.multiply_tests(t_tests, transport_scenarios, result)
93
 
 
94
 
    # each implementation tested with each HTTP version
95
 
    tp_tests, remaining_tests = tests.split_suite_by_condition(
96
 
        remaining_tests, tests.condition_isinstance((
97
 
                SmartHTTPTunnellingTest,
98
 
                TestDoCatchRedirections,
99
 
                TestHTTPConnections,
100
 
                TestHTTPRedirections,
101
 
                TestHTTPSilentRedirections,
102
 
                TestLimitedRangeRequestServer,
103
 
                TestPost,
104
 
                TestProxyHttpServer,
105
 
                TestRanges,
106
 
                TestSpecificRequestHandler,
107
 
                )))
108
 
    protocol_scenarios = [
109
 
            ('HTTP/1.0',  dict(_protocol_version='HTTP/1.0')),
110
 
            ('HTTP/1.1',  dict(_protocol_version='HTTP/1.1')),
111
 
            ]
112
 
    tp_scenarios = tests.multiply_scenarios(transport_scenarios,
113
 
                                            protocol_scenarios)
114
 
    tests.multiply_tests(tp_tests, tp_scenarios, result)
115
 
 
116
 
    # auth: each auth scheme on all http versions on all implementations.
117
 
    tpa_tests, remaining_tests = tests.split_suite_by_condition(
118
 
        remaining_tests, tests.condition_isinstance((
119
 
                TestAuth,
120
 
                )))
121
 
    auth_scheme_scenarios = [
122
 
        ('basic', dict(_auth_scheme='basic')),
123
 
        ('digest', dict(_auth_scheme='digest')),
124
 
        ]
125
 
    tpa_scenarios = tests.multiply_scenarios(tp_scenarios,
126
 
        auth_scheme_scenarios)
127
 
    tests.multiply_tests(tpa_tests, tpa_scenarios, result)
128
 
 
129
 
    # activity: activity on all http versions on all implementations
130
 
    tpact_tests, remaining_tests = tests.split_suite_by_condition(
131
 
        remaining_tests, tests.condition_isinstance((
132
 
                TestActivity,
133
 
                )))
134
 
    activity_scenarios = [
135
 
        ('http', dict(_activity_server=ActivityHTTPServer)),
136
 
        ]
137
 
    if tests.HTTPSServerFeature.available():
138
 
        activity_scenarios.append(
139
 
            ('https', dict(_activity_server=ActivityHTTPSServer)))
140
 
    tpact_scenarios = tests.multiply_scenarios(tp_scenarios,
141
 
        activity_scenarios)
142
 
    tests.multiply_tests(tpact_tests, tpact_scenarios, result)
143
 
 
144
 
    # No parametrization for the remaining tests
145
 
    result.addTests(remaining_tests)
146
 
 
 
149
    for test_class in tests.iter_suite_tests(standard_tests):
 
150
        # Each test class is either standalone or testing for some combination
 
151
        # of transport, protocol version, authentication scheme. Use the right
 
152
        # adpater (or none) depending on the class.
 
153
        if is_testing_for_transports(test_class):
 
154
            result.addTests(t_adapter.adapt(test_class))
 
155
        elif is_also_testing_for_protocols(test_class):
 
156
            result.addTests(tp_adapter.adapt(test_class))
 
157
        elif is_also_testing_for_authentication(test_class):
 
158
            result.addTests(tpa_adapter.adapt(test_class))
 
159
        else:
 
160
            result.addTest(test_class)
147
161
    return result
148
162
 
149
163
 
158
172
 
159
173
class RecordingServer(object):
160
174
    """A fake HTTP server.
161
 
 
 
175
    
162
176
    It records the bytes sent to it, and replies with a 200.
163
177
    """
164
178
 
213
227
        self.port = None
214
228
 
215
229
 
216
 
class TestAuthHeader(tests.TestCase):
217
 
 
218
 
    def parse_header(self, header):
219
 
        ah =  _urllib2_wrappers.AbstractAuthHandler()
220
 
        return ah._parse_auth_header(header)
221
 
 
222
 
    def test_empty_header(self):
223
 
        scheme, remainder = self.parse_header('')
224
 
        self.assertEquals('', scheme)
225
 
        self.assertIs(None, remainder)
226
 
 
227
 
    def test_negotiate_header(self):
228
 
        scheme, remainder = self.parse_header('Negotiate')
229
 
        self.assertEquals('negotiate', scheme)
230
 
        self.assertIs(None, remainder)
231
 
 
232
 
    def test_basic_header(self):
233
 
        scheme, remainder = self.parse_header(
234
 
            'Basic realm="Thou should not pass"')
235
 
        self.assertEquals('basic', scheme)
236
 
        self.assertEquals('realm="Thou should not pass"', remainder)
237
 
 
238
 
    def test_digest_header(self):
239
 
        scheme, remainder = self.parse_header(
240
 
            'Digest realm="Thou should not pass"')
241
 
        self.assertEquals('digest', scheme)
242
 
        self.assertEquals('realm="Thou should not pass"', remainder)
243
 
 
244
 
 
245
230
class TestHTTPServer(tests.TestCase):
246
231
    """Test the HTTP servers implementations."""
247
232
 
464
449
            '"GET /foo/bar HTTP/1.1" 200 - "-" "bzr/%s'
465
450
            % bzrlib.__version__) > -1)
466
451
 
 
452
    def test_get_smart_medium(self):
 
453
        # For HTTP, get_smart_medium should return the transport object.
 
454
        server = self.get_readonly_server()
 
455
        http_transport = self._transport(server.get_url())
 
456
        medium = http_transport.get_smart_medium()
 
457
        self.assertIs(medium, http_transport)
 
458
 
467
459
    def test_has_on_bogus_host(self):
468
460
        # Get a free address and don't 'accept' on it, so that we
469
461
        # can be sure there is no http handler there, but set a
826
818
        # bytes on the socket
827
819
        ireadv = iter(t.readv('a', ((0, 1), (1, 1), (2, 4), (6, 4))))
828
820
        self.assertEqual((0, '0'), ireadv.next())
829
 
        # The server should have issued one request so far
 
821
        # The server should have issued one request so far 
830
822
        self.assertEqual(1, server.GET_request_nb)
831
823
        self.assertEqual('0123456789', t.get_bytes('a'))
832
824
        # get_bytes issued an additional request, the readv pending ones are
1153
1145
        url = self.server.get_url()
1154
1146
        t = self._transport(url)
1155
1147
        try:
1156
 
            self.assertEqual('proxied contents of foo\n', t.get('foo').read())
 
1148
            self.assertEqual(t.get('foo').read(), 'proxied contents of foo\n')
1157
1149
        finally:
1158
1150
            self._restore_env()
1159
1151
 
1162
1154
        url = self.server.get_url()
1163
1155
        t = self._transport(url)
1164
1156
        try:
1165
 
            self.assertEqual('contents of foo\n', t.get('foo').read())
 
1157
            self.assertEqual(t.get('foo').read(), 'contents of foo\n')
1166
1158
        finally:
1167
1159
            self._restore_env()
1168
1160
 
1278
1270
                                  ('bundle',
1279
1271
                                  '# Bazaar revision bundle v0.9\n#\n')
1280
1272
                                  ],)
1281
 
        # The requests to the old server will be redirected to the new server
 
1273
 
1282
1274
        self.old_transport = self._transport(self.old_server.get_url())
1283
1275
 
1284
1276
    def test_redirected(self):
1289
1281
    def test_read_redirected_bundle_from_url(self):
1290
1282
        from bzrlib.bundle import read_bundle_from_url
1291
1283
        url = self.old_transport.abspath('bundle')
1292
 
        bundle = self.applyDeprecated(deprecated_in((1, 12, 0)),
1293
 
                read_bundle_from_url, url)
 
1284
        bundle = read_bundle_from_url(url)
1294
1285
        # If read_bundle_from_url was successful we get an empty bundle
1295
1286
        self.assertEqual([], bundle.revisions)
1296
1287
 
1462
1453
    def _testing_pycurl(self):
1463
1454
        return pycurl_present and self._transport == PyCurlTransport
1464
1455
 
1465
 
    def get_user_url(self, user, password):
 
1456
    def get_user_url(self, user=None, password=None):
1466
1457
        """Build an url embedding user and password"""
1467
1458
        url = '%s://' % self.server._url_protocol
1468
1459
        if user is not None:
1473
1464
        url += '%s:%s/' % (self.server.host, self.server.port)
1474
1465
        return url
1475
1466
 
1476
 
    def get_user_transport(self, user, password):
 
1467
    def get_user_transport(self, user=None, password=None):
1477
1468
        return self._transport(self.get_user_url(user, password))
1478
1469
 
1479
1470
    def test_no_user(self):
1480
1471
        self.server.add_user('joe', 'foo')
1481
 
        t = self.get_user_transport(None, None)
 
1472
        t = self.get_user_transport()
1482
1473
        self.assertRaises(errors.InvalidHttpResponse, t.get, 'a')
1483
1474
        # Only one 'Authentication Required' error should occur
1484
1475
        self.assertEqual(1, self.server.auth_required_errors)
1572
1563
        # Only one 'Authentication Required' error should occur
1573
1564
        self.assertEqual(1, self.server.auth_required_errors)
1574
1565
 
1575
 
    def test_user_from_auth_conf(self):
1576
 
        if self._testing_pycurl():
1577
 
            raise tests.TestNotApplicable(
1578
 
                'pycurl does not support authentication.conf')
1579
 
        user = 'joe'
1580
 
        password = 'foo'
1581
 
        self.server.add_user(user, password)
1582
 
        # Create a minimal config file with the right password
1583
 
        conf = config.AuthenticationConfig()
1584
 
        conf._get_config().update(
1585
 
            {'httptest': {'scheme': 'http', 'port': self.server.port,
1586
 
                          'user': user, 'password': password}})
1587
 
        conf._save()
1588
 
        t = self.get_user_transport(None, None)
1589
 
        # Issue a request to the server to connect
1590
 
        self.assertEqual('contents of a\n', t.get('a').read())
1591
 
        # Only one 'Authentication Required' error should occur
1592
 
        self.assertEqual(1, self.server.auth_required_errors)
1593
 
 
1594
1566
    def test_changing_nonce(self):
1595
1567
        if self._auth_scheme != 'digest':
1596
1568
            raise tests.TestNotApplicable('HTTP auth digest only test')
1642
1614
                protocol_version=self._protocol_version)
1643
1615
        return server
1644
1616
 
1645
 
    def get_user_transport(self, user, password):
 
1617
    def get_user_transport(self, user=None, password=None):
1646
1618
        self._install_env({'all_proxy': self.get_user_url(user, password)})
1647
1619
        return self._transport(self.server.get_url())
1648
1620
 
1775
1747
        # No need to build a valid smart request here, the server will not even
1776
1748
        # try to interpret it.
1777
1749
        self.assertRaises(errors.SmartProtocolError,
1778
 
                          t.get_smart_medium().send_http_smart_request,
1779
 
                          'whatever')
1780
 
 
1781
 
class Test_redirected_to(tests.TestCase):
1782
 
 
1783
 
    def test_redirected_to_subdir(self):
1784
 
        t = self._transport('http://www.example.com/foo')
1785
 
        r = t._redirected_to('http://www.example.com/foo',
1786
 
                             'http://www.example.com/foo/subdir')
1787
 
        self.assertIsInstance(r, type(t))
1788
 
        # Both transports share the some connection
1789
 
        self.assertEquals(t._get_connection(), r._get_connection())
1790
 
 
1791
 
    def test_redirected_to_self_with_slash(self):
1792
 
        t = self._transport('http://www.example.com/foo')
1793
 
        r = t._redirected_to('http://www.example.com/foo',
1794
 
                             'http://www.example.com/foo/')
1795
 
        self.assertIsInstance(r, type(t))
1796
 
        # Both transports share the some connection (one can argue that we
1797
 
        # should return the exact same transport here, but that seems
1798
 
        # overkill).
1799
 
        self.assertEquals(t._get_connection(), r._get_connection())
1800
 
 
1801
 
    def test_redirected_to_host(self):
1802
 
        t = self._transport('http://www.example.com/foo')
1803
 
        r = t._redirected_to('http://www.example.com/foo',
1804
 
                             'http://foo.example.com/foo/subdir')
1805
 
        self.assertIsInstance(r, type(t))
1806
 
 
1807
 
    def test_redirected_to_same_host_sibling_protocol(self):
1808
 
        t = self._transport('http://www.example.com/foo')
1809
 
        r = t._redirected_to('http://www.example.com/foo',
1810
 
                             'https://www.example.com/foo')
1811
 
        self.assertIsInstance(r, type(t))
1812
 
 
1813
 
    def test_redirected_to_same_host_different_protocol(self):
1814
 
        t = self._transport('http://www.example.com/foo')
1815
 
        r = t._redirected_to('http://www.example.com/foo',
1816
 
                             'ftp://www.example.com/foo')
1817
 
        self.assertNotEquals(type(r), type(t))
1818
 
 
1819
 
    def test_redirected_to_different_host_same_user(self):
1820
 
        t = self._transport('http://joe@www.example.com/foo')
1821
 
        r = t._redirected_to('http://www.example.com/foo',
1822
 
                             'https://foo.example.com/foo')
1823
 
        self.assertIsInstance(r, type(t))
1824
 
        self.assertEquals(t._user, r._user)
1825
 
 
1826
 
 
1827
 
class PredefinedRequestHandler(http_server.TestingHTTPRequestHandler):
1828
 
    """Request handler for a unique and pre-defined request.
1829
 
 
1830
 
    The only thing we care about here is how many bytes travel on the wire. But
1831
 
    since we want to measure it for a real http client, we have to send it
1832
 
    correct responses.
1833
 
 
1834
 
    We expect to receive a *single* request nothing more (and we won't even
1835
 
    check what request it is, we just measure the bytes read until an empty
1836
 
    line.
1837
 
    """
1838
 
 
1839
 
    def handle_one_request(self):
1840
 
        tcs = self.server.test_case_server
1841
 
        requestline = self.rfile.readline()
1842
 
        headers = self.MessageClass(self.rfile, 0)
1843
 
        # We just read: the request, the headers, an empty line indicating the
1844
 
        # end of the headers.
1845
 
        bytes_read = len(requestline)
1846
 
        for line in headers.headers:
1847
 
            bytes_read += len(line)
1848
 
        bytes_read += len('\r\n')
1849
 
        if requestline.startswith('POST'):
1850
 
            # The body should be a single line (or we don't know where it ends
1851
 
            # and we don't want to issue a blocking read)
1852
 
            body = self.rfile.readline()
1853
 
            bytes_read += len(body)
1854
 
        tcs.bytes_read = bytes_read
1855
 
 
1856
 
        # We set the bytes written *before* issuing the write, the client is
1857
 
        # supposed to consume every produced byte *before* checking that value.
1858
 
 
1859
 
        # Doing the oppposite may lead to test failure: we may be interrupted
1860
 
        # after the write but before updating the value. The client can then
1861
 
        # continue and read the value *before* we can update it. And yes,
1862
 
        # this has been observed -- vila 20090129
1863
 
        tcs.bytes_written = len(tcs.canned_response)
1864
 
        self.wfile.write(tcs.canned_response)
1865
 
 
1866
 
 
1867
 
class ActivityServerMixin(object):
1868
 
 
1869
 
    def __init__(self, protocol_version):
1870
 
        super(ActivityServerMixin, self).__init__(
1871
 
            request_handler=PredefinedRequestHandler,
1872
 
            protocol_version=protocol_version)
1873
 
        # Bytes read and written by the server
1874
 
        self.bytes_read = 0
1875
 
        self.bytes_written = 0
1876
 
        self.canned_response = None
1877
 
 
1878
 
 
1879
 
class ActivityHTTPServer(ActivityServerMixin, http_server.HttpServer):
1880
 
    pass
1881
 
 
1882
 
 
1883
 
if tests.HTTPSServerFeature.available():
1884
 
    from bzrlib.tests import https_server
1885
 
    class ActivityHTTPSServer(ActivityServerMixin, https_server.HTTPSServer):
1886
 
        pass
1887
 
 
1888
 
 
1889
 
class TestActivity(tests.TestCase):
1890
 
    """Test socket activity reporting.
1891
 
 
1892
 
    We use a special purpose server to control the bytes sent and received and
1893
 
    be able to predict the activity on the client socket.
1894
 
    """
1895
 
 
1896
 
    def setUp(self):
1897
 
        tests.TestCase.setUp(self)
1898
 
        self.server = self._activity_server(self._protocol_version)
1899
 
        self.server.setUp()
1900
 
        self.activities = {}
1901
 
        def report_activity(t, bytes, direction):
1902
 
            count = self.activities.get(direction, 0)
1903
 
            count += bytes
1904
 
            self.activities[direction] = count
1905
 
 
1906
 
        # We override at class level because constructors may propagate the
1907
 
        # bound method and render instance overriding ineffective (an
1908
 
        # alternative would be be to define a specific ui factory instead...)
1909
 
        self.orig_report_activity = self._transport._report_activity
1910
 
        self._transport._report_activity = report_activity
1911
 
 
1912
 
    def tearDown(self):
1913
 
        self._transport._report_activity = self.orig_report_activity
1914
 
        self.server.tearDown()
1915
 
        tests.TestCase.tearDown(self)
1916
 
 
1917
 
    def get_transport(self):
1918
 
        return self._transport(self.server.get_url())
1919
 
 
1920
 
    def assertActivitiesMatch(self):
1921
 
        self.assertEqual(self.server.bytes_read,
1922
 
                         self.activities.get('write', 0), 'written bytes')
1923
 
        self.assertEqual(self.server.bytes_written,
1924
 
                         self.activities.get('read', 0), 'read bytes')
1925
 
 
1926
 
    def test_get(self):
1927
 
        self.server.canned_response = '''HTTP/1.1 200 OK\r
1928
 
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
1929
 
Server: Apache/2.0.54 (Fedora)\r
1930
 
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
1931
 
ETag: "56691-23-38e9ae00"\r
1932
 
Accept-Ranges: bytes\r
1933
 
Content-Length: 35\r
1934
 
Connection: close\r
1935
 
Content-Type: text/plain; charset=UTF-8\r
1936
 
\r
1937
 
Bazaar-NG meta directory, format 1
1938
 
'''
1939
 
        t = self.get_transport()
1940
 
        self.assertEqual('Bazaar-NG meta directory, format 1\n',
1941
 
                         t.get('foo/bar').read())
1942
 
        self.assertActivitiesMatch()
1943
 
 
1944
 
    def test_has(self):
1945
 
        self.server.canned_response = '''HTTP/1.1 200 OK\r
1946
 
Server: SimpleHTTP/0.6 Python/2.5.2\r
1947
 
Date: Thu, 29 Jan 2009 20:21:47 GMT\r
1948
 
Content-type: application/octet-stream\r
1949
 
Content-Length: 20\r
1950
 
Last-Modified: Thu, 29 Jan 2009 20:21:47 GMT\r
1951
 
\r
1952
 
'''
1953
 
        t = self.get_transport()
1954
 
        self.assertTrue(t.has('foo/bar'))
1955
 
        self.assertActivitiesMatch()
1956
 
 
1957
 
    def test_readv(self):
1958
 
        self.server.canned_response = '''HTTP/1.1 206 Partial Content\r
1959
 
Date: Tue, 11 Jul 2006 04:49:48 GMT\r
1960
 
Server: Apache/2.0.54 (Fedora)\r
1961
 
Last-Modified: Thu, 06 Jul 2006 20:22:05 GMT\r
1962
 
ETag: "238a3c-16ec2-805c5540"\r
1963
 
Accept-Ranges: bytes\r
1964
 
Content-Length: 1534\r
1965
 
Connection: close\r
1966
 
Content-Type: multipart/byteranges; boundary=418470f848b63279b\r
1967
 
\r
1968
 
\r
1969
 
--418470f848b63279b\r
1970
 
Content-type: text/plain; charset=UTF-8\r
1971
 
Content-range: bytes 0-254/93890\r
1972
 
\r
1973
 
mbp@sourcefrog.net-20050309040815-13242001617e4a06
1974
 
mbp@sourcefrog.net-20050309040929-eee0eb3e6d1e7627
1975
 
mbp@sourcefrog.net-20050309040957-6cad07f466bb0bb8
1976
 
mbp@sourcefrog.net-20050309041501-c840e09071de3b67
1977
 
mbp@sourcefrog.net-20050309044615-c24a3250be83220a
1978
 
\r
1979
 
--418470f848b63279b\r
1980
 
Content-type: text/plain; charset=UTF-8\r
1981
 
Content-range: bytes 1000-2049/93890\r
1982
 
\r
1983
 
40-fd4ec249b6b139ab
1984
 
mbp@sourcefrog.net-20050311063625-07858525021f270b
1985
 
mbp@sourcefrog.net-20050311231934-aa3776aff5200bb9
1986
 
mbp@sourcefrog.net-20050311231953-73aeb3a131c3699a
1987
 
mbp@sourcefrog.net-20050311232353-f5e33da490872c6a
1988
 
mbp@sourcefrog.net-20050312071639-0a8f59a34a024ff0
1989
 
mbp@sourcefrog.net-20050312073432-b2c16a55e0d6e9fb
1990
 
mbp@sourcefrog.net-20050312073831-a47c3335ece1920f
1991
 
mbp@sourcefrog.net-20050312085412-13373aa129ccbad3
1992
 
mbp@sourcefrog.net-20050313052251-2bf004cb96b39933
1993
 
mbp@sourcefrog.net-20050313052856-3edd84094687cb11
1994
 
mbp@sourcefrog.net-20050313053233-e30a4f28aef48f9d
1995
 
mbp@sourcefrog.net-20050313053853-7c64085594ff3072
1996
 
mbp@sourcefrog.net-20050313054757-a86c3f5871069e22
1997
 
mbp@sourcefrog.net-20050313061422-418f1f73b94879b9
1998
 
mbp@sourcefrog.net-20050313120651-497bd231b19df600
1999
 
mbp@sourcefrog.net-20050314024931-eae0170ef25a5d1a
2000
 
mbp@sourcefrog.net-20050314025438-d52099f915fe65fc
2001
 
mbp@sourcefrog.net-20050314025539-637a636692c055cf
2002
 
mbp@sourcefrog.net-20050314025737-55eb441f430ab4ba
2003
 
mbp@sourcefrog.net-20050314025901-d74aa93bb7ee8f62
2004
 
mbp@source\r
2005
 
--418470f848b63279b--\r
2006
 
'''
2007
 
        t = self.get_transport()
2008
 
        # Remember that the request is ignored and that the ranges below
2009
 
        # doesn't have to match the canned response.
2010
 
        l = list(t.readv('/foo/bar', ((0, 255), (1000, 1050))))
2011
 
        self.assertEqual(2, len(l))
2012
 
        self.assertActivitiesMatch()
2013
 
 
2014
 
    def test_post(self):
2015
 
        self.server.canned_response = '''HTTP/1.1 200 OK\r
2016
 
Date: Tue, 11 Jul 2006 04:32:56 GMT\r
2017
 
Server: Apache/2.0.54 (Fedora)\r
2018
 
Last-Modified: Sun, 23 Apr 2006 19:35:20 GMT\r
2019
 
ETag: "56691-23-38e9ae00"\r
2020
 
Accept-Ranges: bytes\r
2021
 
Content-Length: 35\r
2022
 
Connection: close\r
2023
 
Content-Type: text/plain; charset=UTF-8\r
2024
 
\r
2025
 
lalala whatever as long as itsssss
2026
 
'''
2027
 
        t = self.get_transport()
2028
 
        # We must send a single line of body bytes, see
2029
 
        # PredefinedRequestHandler.handle_one_request
2030
 
        code, f = t._post('abc def end-of-body\n')
2031
 
        self.assertEqual('lalala whatever as long as itsssss\n', f.read())
2032
 
        self.assertActivitiesMatch()
 
1750
                          t.send_http_smart_request, 'whatever')
 
1751