/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_http_response.py

  • Committer: Jelmer Vernooij
  • Date: 2018-02-18 21:42:57 UTC
  • mto: This revision was merged to the branch mainline in revision 6859.
  • Revision ID: jelmer@jelmer.uk-20180218214257-jpevutp1wa30tz3v
Update TODO to reference Breezy, not Bazaar.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_http_response.py
1
 
# Copyright (C) 2006-2010 Canonical Ltd
 
1
# Copyright (C) 2006-2010, 2012, 2013, 2016 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
37
37
  InvalidHttpResponse.
38
38
"""
39
39
 
40
 
from cStringIO import StringIO
41
 
import httplib
 
40
try:
 
41
    import http.client as http_client
 
42
except ImportError:  # python < 3
 
43
    import httplib as http_client
42
44
 
43
 
from bzrlib import (
 
45
from .. import (
44
46
    errors,
45
47
    tests,
46
48
    )
47
 
from bzrlib.transport.http import (
 
49
from ..sixish import (
 
50
    BytesIO,
 
51
    )
 
52
from ..transport.http import (
48
53
    response,
49
54
    _urllib2_wrappers,
50
55
    )
51
 
from bzrlib.tests.file_utils import (
 
56
from .file_utils import (
52
57
    FakeReadFile,
53
58
    )
54
59
 
57
62
    """A socket-like object that can be given a predefined content."""
58
63
 
59
64
    def __init__(self, data):
60
 
        self.readfile = StringIO(data)
 
65
        self.readfile = BytesIO(data)
61
66
 
62
67
    def makefile(self, mode='r', bufsize=None):
63
68
        return self.readfile
75
80
        pass
76
81
 
77
82
 
 
83
class TestResponseFileIter(tests.TestCase):
 
84
 
 
85
    def test_iter_empty(self):
 
86
        f = response.ResponseFile('empty', BytesIO())
 
87
        self.assertEqual([], list(f))
 
88
 
 
89
    def test_iter_many(self):
 
90
        f = response.ResponseFile('many', BytesIO(b'0\n1\nboo!\n'))
 
91
        self.assertEqual(['0\n', '1\n', 'boo!\n'], list(f))
 
92
 
 
93
 
78
94
class TestHTTPConnection(tests.TestCase):
79
95
 
80
96
    def test_cleanup_pipe(self):
92
108
        # Now, get the response
93
109
        resp = conn.getresponse()
94
110
        # Read part of the response
95
 
        self.assertEquals('0123456789\n', resp.read(11))
 
111
        self.assertEqual('0123456789\n', resp.read(11))
96
112
        # Override the thresold to force the warning emission
97
113
        conn._range_warning_thresold = 6 # There are 7 bytes pending
98
114
        conn.cleanup_pipe()
110
126
 
111
127
    def test_can_read_at_first_access(self):
112
128
        """Test that the just created file can be read."""
113
 
        self.assertEquals(self.alpha, self._file.read())
 
129
        self.assertEqual(self.alpha, self._file.read())
114
130
 
115
131
    def test_seek_read(self):
116
132
        """Test seek/read inside the range."""
117
133
        f = self._file
118
134
        start = self.first_range_start
119
135
        # Before any use, tell() should be at the range start
120
 
        self.assertEquals(start, f.tell())
 
136
        self.assertEqual(start, f.tell())
121
137
        cur = start # For an overall offset assertion
122
138
        f.seek(start + 3)
123
139
        cur += 3
124
 
        self.assertEquals('def', f.read(3))
 
140
        self.assertEqual('def', f.read(3))
125
141
        cur += len('def')
126
142
        f.seek(4, 1)
127
143
        cur += 4
128
 
        self.assertEquals('klmn', f.read(4))
 
144
        self.assertEqual('klmn', f.read(4))
129
145
        cur += len('klmn')
130
146
        # read(0) in the middle of a range
131
 
        self.assertEquals('', f.read(0))
 
147
        self.assertEqual('', f.read(0))
132
148
        # seek in place
133
149
        here = f.tell()
134
150
        f.seek(0, 1)
135
 
        self.assertEquals(here, f.tell())
136
 
        self.assertEquals(cur, f.tell())
 
151
        self.assertEqual(here, f.tell())
 
152
        self.assertEqual(cur, f.tell())
137
153
 
138
154
    def test_read_zero(self):
139
155
        f = self._file
140
 
        start = self.first_range_start
141
 
        self.assertEquals('', f.read(0))
 
156
        self.assertEqual('', f.read(0))
142
157
        f.seek(10, 1)
143
 
        self.assertEquals('', f.read(0))
 
158
        self.assertEqual('', f.read(0))
144
159
 
145
160
    def test_seek_at_range_end(self):
146
161
        f = self._file
149
164
    def test_read_at_range_end(self):
150
165
        """Test read behaviour at range end."""
151
166
        f = self._file
152
 
        self.assertEquals(self.alpha, f.read())
153
 
        self.assertEquals('', f.read(0))
 
167
        self.assertEqual(self.alpha, f.read())
 
168
        self.assertEqual('', f.read(0))
154
169
        self.assertRaises(errors.InvalidRange, f.read, 1)
155
170
 
156
171
    def test_unbounded_read_after_seek(self):
157
172
        f = self._file
158
173
        f.seek(24, 1)
159
174
        # Should not cross ranges
160
 
        self.assertEquals('yz', f.read())
 
175
        self.assertEqual('yz', f.read())
161
176
 
162
177
    def test_seek_backwards(self):
163
178
        f = self._file
187
202
 
188
203
       The semantic is unclear in case of multiple ranges. Seeking from end
189
204
       exists only for the http transports, cannot be used if the file size is
190
 
       unknown and is not used in bzrlib itself. This test must be (and is)
 
205
       unknown and is not used in breezy itself. This test must be (and is)
191
206
       overridden by daughter classes.
192
207
 
193
208
       Reading from end makes sense only when a range has been requested from
197
212
       """
198
213
       f = self._file
199
214
       f.seek(-2, 2)
200
 
       self.assertEquals('yz', f.read())
 
215
       self.assertEqual('yz', f.read())
201
216
 
202
217
 
203
218
class TestRangeFileSizeUnknown(tests.TestCase, TestRangeFileMixin):
206
221
    def setUp(self):
207
222
        super(TestRangeFileSizeUnknown, self).setUp()
208
223
        self._file = response.RangeFile('Whole_file_size_known',
209
 
                                        StringIO(self.alpha))
 
224
                                        BytesIO(self.alpha))
210
225
        # We define no range, relying on RangeFile to provide default values
211
226
        self.first_range_start = 0 # It's the whole file
212
227
 
220
235
    def test_read_at_range_end(self):
221
236
        """Test read behaviour at range end."""
222
237
        f = self._file
223
 
        self.assertEquals(self.alpha, f.read())
224
 
        self.assertEquals('', f.read(0))
225
 
        self.assertEquals('', f.read(1))
 
238
        self.assertEqual(self.alpha, f.read())
 
239
        self.assertEqual('', f.read(0))
 
240
        self.assertEqual('', f.read(1))
226
241
 
227
242
 
228
243
class TestRangeFileSizeKnown(tests.TestCase, TestRangeFileMixin):
231
246
    def setUp(self):
232
247
        super(TestRangeFileSizeKnown, self).setUp()
233
248
        self._file = response.RangeFile('Whole_file_size_known',
234
 
                                        StringIO(self.alpha))
 
249
                                        BytesIO(self.alpha))
235
250
        self._file.set_range(0, len(self.alpha))
236
251
        self.first_range_start = 0 # It's the whole file
237
252
 
242
257
    def setUp(self):
243
258
        super(TestRangeFileSingleRange, self).setUp()
244
259
        self._file = response.RangeFile('Single_range_file',
245
 
                                        StringIO(self.alpha))
 
260
                                        BytesIO(self.alpha))
246
261
        self.first_range_start = 15
247
262
        self._file.set_range(self.first_range_start, len(self.alpha))
248
263
 
291
306
        content += self._boundary_line()
292
307
 
293
308
        self._file = response.RangeFile('Multiple_ranges_file',
294
 
                                        StringIO(content))
 
309
                                        BytesIO(content))
295
310
        self.set_file_boundary()
296
311
 
297
312
    def _boundary_line(self):
337
352
 
338
353
    def test_read_all_ranges(self):
339
354
        f = self._file
340
 
        self.assertEquals(self.alpha, f.read()) # Read first range
 
355
        self.assertEqual(self.alpha, f.read()) # Read first range
341
356
        f.seek(100) # Trigger the second range recognition
342
 
        self.assertEquals(self.alpha, f.read()) # Read second range
343
 
        self.assertEquals(126, f.tell())
 
357
        self.assertEqual(self.alpha, f.read()) # Read second range
 
358
        self.assertEqual(126, f.tell())
344
359
        f.seek(126) # Start of third range which is also the current pos !
345
 
        self.assertEquals('A', f.read(1))
 
360
        self.assertEqual('A', f.read(1))
346
361
        f.seek(10, 1)
347
 
        self.assertEquals('LMN', f.read(3))
 
362
        self.assertEqual('LMN', f.read(3))
348
363
 
349
364
    def test_seek_from_end(self):
350
365
        """See TestRangeFileMixin.test_seek_from_end."""
354
369
        # behaviour.
355
370
        f = self._file
356
371
        f.seek(-2, 2)
357
 
        self.assertEquals('yz', f.read())
 
372
        self.assertEqual('yz', f.read())
358
373
        self.assertRaises(errors.InvalidRange, f.seek, -2, 2)
359
374
 
360
375
    def test_seek_into_void(self):
371
386
 
372
387
    def test_seek_across_ranges(self):
373
388
        f = self._file
374
 
        start = self.first_range_start
375
389
        f.seek(126) # skip the two first ranges
376
 
        self.assertEquals('AB', f.read(2))
 
390
        self.assertEqual('AB', f.read(2))
377
391
 
378
392
    def test_checked_read_dont_overflow_buffers(self):
379
393
        f = self._file
380
 
        start = self.first_range_start
381
394
        # We force a very low value to exercise all code paths in _checked_read
382
395
        f._discarded_buf_size = 8
383
396
        f.seek(126) # skip the two first ranges
384
 
        self.assertEquals('AB', f.read(2))
 
397
        self.assertEqual('AB', f.read(2))
385
398
 
386
399
    def test_seek_twice_between_ranges(self):
387
400
        f = self._file
399
412
 
400
413
    def test_read_at_range_end(self):
401
414
        f = self._file
402
 
        self.assertEquals(self.alpha, f.read())
403
 
        self.assertEquals(self.alpha, f.read())
404
 
        self.assertEquals(self.alpha.upper(), f.read())
 
415
        self.assertEqual(self.alpha, f.read())
 
416
        self.assertEqual(self.alpha, f.read())
 
417
        self.assertEqual(self.alpha.upper(), f.read())
405
418
        self.assertRaises(errors.InvalidHttpResponse, f.read, 1)
406
419
 
407
420
 
435
448
 
436
449
    def test_seek_whence(self):
437
450
        """Test the seek whence parameter values."""
438
 
        f = response.RangeFile('foo', StringIO('abc'))
 
451
        f = response.RangeFile('foo', BytesIO(b'abc'))
439
452
        f.set_range(0, 3)
440
453
        f.seek(0)
441
454
        f.seek(1, 1)
445
458
    def test_range_syntax(self):
446
459
        """Test the Content-Range scanning."""
447
460
 
448
 
        f = response.RangeFile('foo', StringIO())
 
461
        f = response.RangeFile('foo', BytesIO())
449
462
 
450
463
        def ok(expected, header_value):
451
464
            f.set_range_from_header(header_value)
452
465
            # Slightly peek under the covers to get the size
453
 
            self.assertEquals(expected, (f.tell(), f._size))
 
466
            self.assertEqual(expected, (f.tell(), f._size))
454
467
 
455
468
        ok((1, 10), 'bytes 1-10/11')
456
469
        ok((1, 10), 'bytes 1-10/*')
701
714
class TestHandleResponse(tests.TestCase):
702
715
 
703
716
    def _build_HTTPMessage(self, raw_headers):
704
 
        status_and_headers = StringIO(raw_headers)
 
717
        status_and_headers = BytesIO(raw_headers)
705
718
        # Get rid of the status line
706
719
        status_and_headers.readline()
707
 
        msg = httplib.HTTPMessage(status_and_headers)
 
720
        msg = http_client.HTTPMessage(status_and_headers)
708
721
        return msg
709
722
 
710
723
    def get_response(self, a_response):
712
725
        code, raw_headers, body = a_response
713
726
        msg = self._build_HTTPMessage(raw_headers)
714
727
        return response.handle_response('http://foo', code, msg,
715
 
                                        StringIO(a_response[2]))
 
728
                                        BytesIO(a_response[2]))
716
729
 
717
730
    def test_full_text(self):
718
731
        out = self.get_response(_full_text_response)
719
 
        # It is a StringIO from the original data
 
732
        # It is a BytesIO from the original data
720
733
        self.assertEqual(_full_text_response[2], out.read())
721
734
 
722
735
    def test_single_range(self):
764
777
        # We should not require Content-Type for a full response
765
778
        code, raw_headers, body = _full_text_response_no_content_type
766
779
        msg = self._build_HTTPMessage(raw_headers)
767
 
        out = response.handle_response('http://foo', code, msg, StringIO(body))
 
780
        out = response.handle_response('http://foo', code, msg, BytesIO(body))
768
781
        self.assertEqual(body, out.read())
769
782
 
770
783
    def test_full_text_no_content_length(self):
771
784
        code, raw_headers, body = _full_text_response_no_content_length
772
785
        msg = self._build_HTTPMessage(raw_headers)
773
 
        out = response.handle_response('http://foo', code, msg, StringIO(body))
 
786
        out = response.handle_response('http://foo', code, msg, BytesIO(body))
774
787
        self.assertEqual(body, out.read())
775
788
 
776
789
    def test_missing_content_range(self):
778
791
        msg = self._build_HTTPMessage(raw_headers)
779
792
        self.assertRaises(errors.InvalidHttpResponse,
780
793
                          response.handle_response,
781
 
                          'http://bogus', code, msg, StringIO(body))
 
794
                          'http://bogus', code, msg, BytesIO(body))
782
795
 
783
796
    def test_multipart_no_content_range(self):
784
797
        code, raw_headers, body = _multipart_no_content_range
785
798
        msg = self._build_HTTPMessage(raw_headers)
786
799
        self.assertRaises(errors.InvalidHttpResponse,
787
800
                          response.handle_response,
788
 
                          'http://bogus', code, msg, StringIO(body))
 
801
                          'http://bogus', code, msg, BytesIO(body))
789
802
 
790
803
    def test_multipart_no_boundary(self):
791
804
        out = self.get_response(_multipart_no_boundary)
800
813
    """
801
814
 
802
815
    def setUp(self):
803
 
        tests.TestCase.setUp(self)
 
816
        super(TestRangeFileSizeReadLimited, self).setUp()
804
817
        # create a test datablock larger than _max_read_size.
805
818
        chunk_size = response.RangeFile._max_read_size
806
819
        test_pattern = '0123456789ABCDEF'