/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to breezy/tests/test_http_response.py

• browse files at revision 7122.6.3
• compare with another revision
• download diff
• download tarball
• view history from revision 7122.6.3

Show diffs side-by-side

added

removed

breezy/tests/test_http_response.py

112

112

        # Read part of the response

113

113

        self.assertEqual(b'0123456789\n', resp.read(11))

114

114

        # Override the thresold to force the warning emission

115

 

        conn._range_warning_thresold = 6 # There are 7 bytes pending

 

115

        conn._range_warning_thresold = 6  # There are 7 bytes pending

116

116

        conn.cleanup_pipe()

117

117

        self.assertContainsRe(self.get_log(), 'Got a 200 response when asking')

118

118

 

136

136

        start = self.first_range_start

137

137

        # Before any use, tell() should be at the range start

138

138

        self.assertEqual(start, f.tell())

139

 

        cur = start # For an overall offset assertion

 

139

        cur = start  # For an overall offset assertion

140

140

        f.seek(start + 3)

141

141

        cur += 3

142

142

        self.assertEqual(b'def', f.read(3))

200

200

        self.assertRaises(errors.InvalidRange, f.read, 10)

201

201

 

202

202

    def test_seek_from_end(self):

203

 

       """Test seeking from the end of the file.

204

 

 

205

 

       The semantic is unclear in case of multiple ranges. Seeking from end

206

 

       exists only for the http transports, cannot be used if the file size is

207

 

       unknown and is not used in breezy itself. This test must be (and is)

208

 

       overridden by daughter classes.

209

 

 

210

 

       Reading from end makes sense only when a range has been requested from

211

 

       the end of the file (see HttpTransportBase._get() when using the

212

 

       'tail_amount' parameter). The HTTP response can only be a whole file or

213

 

       a single range.

214

 

       """

215

 

       f = self._file

216

 

       f.seek(-2, 2)

217

 

       self.assertEqual(b'yz', f.read())

 

203

        """Test seeking from the end of the file.

 

204

 

 

205

        The semantic is unclear in case of multiple ranges. Seeking from end

 

206

        exists only for the http transports, cannot be used if the file size is

 

207

        unknown and is not used in breezy itself. This test must be (and is)

 

208

        overridden by daughter classes.

 

209

 

 

210

        Reading from end makes sense only when a range has been requested from

 

211

        the end of the file (see HttpTransportBase._get() when using the

 

212

        'tail_amount' parameter). The HTTP response can only be a whole file or

 

213

        a single range.

 

214

        """

 

215

        f = self._file

 

216

        f.seek(-2, 2)

 

217

        self.assertEqual(b'yz', f.read())

218

218

 

219

219

 

220

220

class TestRangeFileSizeUnknown(tests.TestCase, TestRangeFileMixin):

225

225

        self._file = response.RangeFile('Whole_file_size_known',

226

226

                                        BytesIO(self.alpha))

227

227

        # We define no range, relying on RangeFile to provide default values

228

 

        self.first_range_start = 0 # It's the whole file

 

228

        self.first_range_start = 0  # It's the whole file

229

229

 

230

230

    def test_seek_from_end(self):

231

231

        """See TestRangeFileMixin.test_seek_from_end.

250

250

        self._file = response.RangeFile('Whole_file_size_known',

251

251

                                        BytesIO(self.alpha))

252

252

        self._file.set_range(0, len(self.alpha))

253

 

        self.first_range_start = 0 # It's the whole file

 

253

        self.first_range_start = 0  # It's the whole file

254

254

 

255

255

 

256

256

class TestRangeFileSingleRange(tests.TestCase, TestRangeFileMixin):

263

263

        self.first_range_start = 15

264

264

        self._file.set_range(self.first_range_start, len(self.alpha))

265

265

 

266

 

 

267

266

    def test_read_before_range(self):

268

267

        # This can't occur under normal circumstances, we have to force it

269

268

        f = self._file

270

 

        f._pos = 0 # Force an invalid pos

 

269

        f._pos = 0  # Force an invalid pos

271

270

        self.assertRaises(errors.InvalidRange, f.read, 2)

272

271

 

273

272

 

297

296

 

298

297

        content = b''

299

298

        self.first_range_start = 25

300

 

        file_size = 200 # big enough to encompass all ranges

 

299

        file_size = 200  # big enough to encompass all ranges

301

300

        for (start, part) in [(self.first_range_start, self.alpha),

302

301

                              # Two contiguous ranges

303

302

                              (100, self.alpha),

347

346

        if isinstance(file_size, int):

348

347

            file_size = b'%d' % file_size

349

348

        range += b'Content-Range: bytes %d-%d/%s\r\n' % (offset,

350

 

                                                         offset+len(data)-1,

 

349

                                                         offset +

 

350

                                                         len(data) - 1,

351

351

                                                         file_size)

352

352

        range += b'\r\n'

353

353

        # Finally the raw bytes

356

356

 

357

357

    def test_read_all_ranges(self):

358

358

        f = self._file

359

 

        self.assertEqual(self.alpha, f.read()) # Read first range

360

 

        f.seek(100) # Trigger the second range recognition

361

 

        self.assertEqual(self.alpha, f.read()) # Read second range

 

359

        self.assertEqual(self.alpha, f.read())  # Read first range

 

360

        f.seek(100)  # Trigger the second range recognition

 

361

        self.assertEqual(self.alpha, f.read())  # Read second range

362

362

        self.assertEqual(126, f.tell())

363

 

        f.seek(126) # Start of third range which is also the current pos !

 

363

        f.seek(126)  # Start of third range which is also the current pos !

364

364

        self.assertEqual(b'A', f.read(1))

365

365

        f.seek(10, 1)

366

366

        self.assertEqual(b'LMN', f.read(3))

390

390

 

391

391

    def test_seek_across_ranges(self):

392

392

        f = self._file

393

 

        f.seek(126) # skip the two first ranges

 

393

        f.seek(126)  # skip the two first ranges

394

394

        self.assertEqual(b'AB', f.read(2))

395

395

 

396

396

    def test_checked_read_dont_overflow_buffers(self):

397

397

        f = self._file

398

398

        # We force a very low value to exercise all code paths in _checked_read

399

399

        f._discarded_buf_size = 8

400

 

        f.seek(126) # skip the two first ranges

 

400

        f.seek(126)  # skip the two first ranges

401

401

        self.assertEqual(b'AB', f.read(2))

402

402

 

403

403

    def test_seek_twice_between_ranges(self):

404

404

        f = self._file

405

405

        start = self.first_range_start

406

 

        f.seek(start + 40) # Past the first range but before the second

 

406

        f.seek(start + 40)  # Past the first range but before the second

407

407

        # Now the file is positioned at the second range start (100)

408

408

        self.assertRaises(errors.InvalidRange, f.seek, start + 41)

409

409

 

474

474

        ok((12, 2), '\tbytes 12-13/*')

475

475

        ok((28, 1), '  bytes 28-28/*')

476

476

        ok((2123, 2120), 'bytes  2123-4242/12310')

477

 

        ok((1, 10), 'bytes 1-10/ttt') # We don't check total (ttt)

 

477

        ok((1, 10), 'bytes 1-10/ttt')  # We don't check total (ttt)

478

478

 

479

479

        def nok(header_value):

480

480

            self.assertRaises(errors.InvalidHttpRange,

594

594

Proxy-Connection: keep-alive\r

595

595

\r

596

596

""",

597

 

b"""\r

 

597

                                   b"""\r

598

598

--squid/2.5.STABLE12:C99323425AD4FE26F726261FA6C24196\r

599

599

Content-Type: text/plain\r

600

600

Content-Range: bytes 0-99/18672\r

689

689

Content-Length: 598\r

690

690

\r

691

691

""",

692

 

b"""\r

 

692

                               b"""\r

693

693

--THIS_SEPARATES\r

694

694

Content-Type: text/plain\r

695

695

\r

703

703

Content-Length: 598\r

704

704

\r

705

705

""",

706

 

b"""\r

 

706

                          b"""\r

707

707

--THIS_SEPARATES\r

708

708

Content-Type: text/plain\r

709

709

Content-Range: bytes 0-18/18672\r

821

821

        # create a test datablock larger than _max_read_size.

822

822

        chunk_size = response.RangeFile._max_read_size

823

823

        test_pattern = b'0123456789ABCDEF'

824

 

        self.test_data =  test_pattern * (3 * chunk_size // len(test_pattern))

 

824

        self.test_data = test_pattern * (3 * chunk_size // len(test_pattern))

825

825

        self.test_data_len = len(self.test_data)

826

826

 

827

827

    def test_max_read_size(self):

844

844

        if response_data != self.test_data:

845

845

            message = "Data not equal.  Expected %d bytes, received %d."

846

846

            self.fail(message % (len(response_data), self.test_data_len))

847

 

 

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0