1
# Copyright (C) 2006-2010 Canonical Ltd
1
# Copyright (C) 2006-2010, 2012, 2013, 2016 Canonical Ltd
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
103
103
# Now, get the response
104
104
resp = conn.getresponse()
105
105
# Read part of the response
106
self.assertEquals('0123456789\n', resp.read(11))
106
self.assertEqual('0123456789\n', resp.read(11))
107
107
# Override the thresold to force the warning emission
108
108
conn._range_warning_thresold = 6 # There are 7 bytes pending
109
109
conn.cleanup_pipe()
122
122
def test_can_read_at_first_access(self):
123
123
"""Test that the just created file can be read."""
124
self.assertEquals(self.alpha, self._file.read())
124
self.assertEqual(self.alpha, self._file.read())
126
126
def test_seek_read(self):
127
127
"""Test seek/read inside the range."""
129
129
start = self.first_range_start
130
130
# Before any use, tell() should be at the range start
131
self.assertEquals(start, f.tell())
131
self.assertEqual(start, f.tell())
132
132
cur = start # For an overall offset assertion
133
133
f.seek(start + 3)
135
self.assertEquals('def', f.read(3))
135
self.assertEqual('def', f.read(3))
136
136
cur += len('def')
139
self.assertEquals('klmn', f.read(4))
139
self.assertEqual('klmn', f.read(4))
140
140
cur += len('klmn')
141
141
# read(0) in the middle of a range
142
self.assertEquals('', f.read(0))
142
self.assertEqual('', f.read(0))
146
self.assertEquals(here, f.tell())
147
self.assertEquals(cur, f.tell())
146
self.assertEqual(here, f.tell())
147
self.assertEqual(cur, f.tell())
149
149
def test_read_zero(self):
151
self.assertEquals('', f.read(0))
151
self.assertEqual('', f.read(0))
153
self.assertEquals('', f.read(0))
153
self.assertEqual('', f.read(0))
155
155
def test_seek_at_range_end(self):
159
159
def test_read_at_range_end(self):
160
160
"""Test read behaviour at range end."""
162
self.assertEquals(self.alpha, f.read())
163
self.assertEquals('', f.read(0))
162
self.assertEqual(self.alpha, f.read())
163
self.assertEqual('', f.read(0))
164
164
self.assertRaises(errors.InvalidRange, f.read, 1)
166
166
def test_unbounded_read_after_seek(self):
169
169
# Should not cross ranges
170
self.assertEquals('yz', f.read())
170
self.assertEqual('yz', f.read())
172
172
def test_seek_backwards(self):
210
self.assertEquals('yz', f.read())
210
self.assertEqual('yz', f.read())
213
213
class TestRangeFileSizeUnknown(tests.TestCase, TestRangeFileMixin):
230
230
def test_read_at_range_end(self):
231
231
"""Test read behaviour at range end."""
233
self.assertEquals(self.alpha, f.read())
234
self.assertEquals('', f.read(0))
235
self.assertEquals('', f.read(1))
233
self.assertEqual(self.alpha, f.read())
234
self.assertEqual('', f.read(0))
235
self.assertEqual('', f.read(1))
238
238
class TestRangeFileSizeKnown(tests.TestCase, TestRangeFileMixin):
348
348
def test_read_all_ranges(self):
350
self.assertEquals(self.alpha, f.read()) # Read first range
350
self.assertEqual(self.alpha, f.read()) # Read first range
351
351
f.seek(100) # Trigger the second range recognition
352
self.assertEquals(self.alpha, f.read()) # Read second range
353
self.assertEquals(126, f.tell())
352
self.assertEqual(self.alpha, f.read()) # Read second range
353
self.assertEqual(126, f.tell())
354
354
f.seek(126) # Start of third range which is also the current pos !
355
self.assertEquals('A', f.read(1))
355
self.assertEqual('A', f.read(1))
357
self.assertEquals('LMN', f.read(3))
357
self.assertEqual('LMN', f.read(3))
359
359
def test_seek_from_end(self):
360
360
"""See TestRangeFileMixin.test_seek_from_end."""
367
self.assertEquals('yz', f.read())
367
self.assertEqual('yz', f.read())
368
368
self.assertRaises(errors.InvalidRange, f.seek, -2, 2)
370
370
def test_seek_into_void(self):
382
382
def test_seek_across_ranges(self):
384
384
f.seek(126) # skip the two first ranges
385
self.assertEquals('AB', f.read(2))
385
self.assertEqual('AB', f.read(2))
387
387
def test_checked_read_dont_overflow_buffers(self):
389
389
# We force a very low value to exercise all code paths in _checked_read
390
390
f._discarded_buf_size = 8
391
391
f.seek(126) # skip the two first ranges
392
self.assertEquals('AB', f.read(2))
392
self.assertEqual('AB', f.read(2))
394
394
def test_seek_twice_between_ranges(self):
408
408
def test_read_at_range_end(self):
410
self.assertEquals(self.alpha, f.read())
411
self.assertEquals(self.alpha, f.read())
412
self.assertEquals(self.alpha.upper(), f.read())
410
self.assertEqual(self.alpha, f.read())
411
self.assertEqual(self.alpha, f.read())
412
self.assertEqual(self.alpha.upper(), f.read())
413
413
self.assertRaises(errors.InvalidHttpResponse, f.read, 1)
458
458
def ok(expected, header_value):
459
459
f.set_range_from_header(header_value)
460
460
# Slightly peek under the covers to get the size
461
self.assertEquals(expected, (f.tell(), f._size))
461
self.assertEqual(expected, (f.tell(), f._size))
463
463
ok((1, 10), 'bytes 1-10/11')
464
464
ok((1, 10), 'bytes 1-10/*')