323
319
self.assertRaises(errors.BzrError, osutils.kind_marker, "unknown")
326
class TestPumpFile(TestCase):
327
"""Test pumpfile method."""
329
# create a test datablock
330
self.block_size = 512
331
pattern = '0123456789ABCDEF'
332
self.test_data = pattern * (3 * self.block_size / len(pattern))
333
self.test_data_len = len(self.test_data)
335
def test_bracket_block_size(self):
336
"""Read data in blocks with the requested read size bracketing the
338
# make sure test data is larger than max read size
339
self.assertTrue(self.test_data_len > self.block_size)
341
from_file = FakeReadFile(self.test_data)
344
# read (max / 2) bytes and verify read size wasn't affected
345
num_bytes_to_read = self.block_size / 2
346
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
347
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
348
self.assertEqual(from_file.get_read_count(), 1)
350
# read (max) bytes and verify read size wasn't affected
351
num_bytes_to_read = self.block_size
352
from_file.reset_read_count()
353
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
354
self.assertEqual(from_file.get_max_read_size(), num_bytes_to_read)
355
self.assertEqual(from_file.get_read_count(), 1)
357
# read (max + 1) bytes and verify read size was limited
358
num_bytes_to_read = self.block_size + 1
359
from_file.reset_read_count()
360
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
361
self.assertEqual(from_file.get_max_read_size(), self.block_size)
362
self.assertEqual(from_file.get_read_count(), 2)
364
# finish reading the rest of the data
365
num_bytes_to_read = self.test_data_len - to_file.tell()
366
pumpfile(from_file, to_file, num_bytes_to_read, self.block_size)
368
# report error if the data wasn't equal (we only report the size due
369
# to the length of the data)
370
response_data = to_file.getvalue()
371
if response_data != self.test_data:
372
message = "Data not equal. Expected %d bytes, received %d."
373
self.fail(message % (len(response_data), self.test_data_len))
375
def test_specified_size(self):
376
"""Request a transfer larger than the maximum block size and verify
377
that the maximum read doesn't exceed the block_size."""
378
# make sure test data is larger than max read size
379
self.assertTrue(self.test_data_len > self.block_size)
381
# retrieve data in blocks
382
from_file = FakeReadFile(self.test_data)
384
pumpfile(from_file, to_file, self.test_data_len, self.block_size)
386
# verify read size was equal to the maximum read size
387
self.assertTrue(from_file.get_max_read_size() > 0)
388
self.assertEqual(from_file.get_max_read_size(), self.block_size)
389
self.assertEqual(from_file.get_read_count(), 3)
391
# report error if the data wasn't equal (we only report the size due
392
# to the length of the data)
393
response_data = to_file.getvalue()
394
if response_data != self.test_data:
395
message = "Data not equal. Expected %d bytes, received %d."
396
self.fail(message % (len(response_data), self.test_data_len))
398
def test_to_eof(self):
399
"""Read to end-of-file and verify that the reads are not larger than
400
the maximum read size."""
401
# make sure test data is larger than max read size
402
self.assertTrue(self.test_data_len > self.block_size)
404
# retrieve data to EOF
405
from_file = FakeReadFile(self.test_data)
407
pumpfile(from_file, to_file, -1, self.block_size)
409
# verify read size was equal to the maximum read size
410
self.assertEqual(from_file.get_max_read_size(), self.block_size)
411
self.assertEqual(from_file.get_read_count(), 4)
413
# report error if the data wasn't equal (we only report the size due
414
# to the length of the data)
415
response_data = to_file.getvalue()
416
if response_data != self.test_data:
417
message = "Data not equal. Expected %d bytes, received %d."
418
self.fail(message % (len(response_data), self.test_data_len))
420
def test_defaults(self):
421
"""Verifies that the default arguments will read to EOF -- this
422
test verifies that any existing usages of pumpfile will not be broken
423
with this new version."""
424
# retrieve data using default (old) pumpfile method
425
from_file = FakeReadFile(self.test_data)
427
pumpfile(from_file, to_file)
429
# report error if the data wasn't equal (we only report the size due
430
# to the length of the data)
431
response_data = to_file.getvalue()
432
if response_data != self.test_data:
433
message = "Data not equal. Expected %d bytes, received %d."
434
self.fail(message % (len(response_data), self.test_data_len))
436
322
class TestSafeUnicode(TestCase):
438
324
def test_from_ascii_string(self):
added
removed
bzrlib/tests/test_osutils.py
