74
76
output = StringIO()
75
77
writer = pack.ContainerWriter(output.write)
77
writer.add_bytes_record('abc', names=['name1', 'name2'])
79
'Bazaar pack format 1 (introduced in 0.18)\n'
80
'B3\nname1\nname2\n\nabc',
79
offset, length = writer.add_bytes_record('abc', names=[('name1', ), ('name2', )])
80
self.assertEqual((42, 19), (offset, length))
82
'Bazaar pack format 1 (introduced in 0.18)\n'
83
'B3\nname1\nname2\n\nabc',
86
def test_add_bytes_record_two_names(self):
87
"""Add a bytes record with two names."""
89
writer = pack.ContainerWriter(output.write)
91
offset, length = writer.add_bytes_record('abc', names=[('name1', ), ('name2', )])
92
self.assertEqual((42, 19), (offset, length))
94
'Bazaar pack format 1 (introduced in 0.18)\n'
95
'B3\nname1\nname2\n\nabc',
98
def test_add_bytes_record_two_element_name(self):
99
"""Add a bytes record with a two-element name."""
101
writer = pack.ContainerWriter(output.write)
103
offset, length = writer.add_bytes_record('abc', names=[('name1', 'name2')])
104
self.assertEqual((42, 19), (offset, length))
106
'Bazaar pack format 1 (introduced in 0.18)\n'
107
'B3\nname1\x00name2\n\nabc',
110
def test_add_second_bytes_record_gets_higher_offset(self):
112
writer = pack.ContainerWriter(output.write)
114
writer.add_bytes_record('abc', names=[])
115
offset, length = writer.add_bytes_record('abc', names=[])
116
self.assertEqual((49, 7), (offset, length))
118
'Bazaar pack format 1 (introduced in 0.18)\n'
81
121
output.getvalue())
83
123
def test_add_bytes_record_invalid_name(self):
256
296
reader = self.get_reader_for("5\nname1\nname2\n\naaaaa")
257
297
names, get_bytes = reader.read()
258
self.assertEqual(['name1', 'name2'], names)
298
self.assertEqual([('name1', ), ('name2', )], names)
299
self.assertEqual('aaaaa', get_bytes(None))
301
def test_record_with_two_part_names(self):
302
"""Reading a Bytes record with a two_part name reads both."""
303
reader = self.get_reader_for("5\nname1\x00name2\n\naaaaa")
304
names, get_bytes = reader.read()
305
self.assertEqual([('name1', 'name2', )], names)
259
306
self.assertEqual('aaaaa', get_bytes(None))
261
308
def test_invalid_length(self):
375
422
self.assertEqual('', get_bytes(99))
425
class TestMakeReadvReader(tests.TestCaseWithTransport):
427
def test_read_skipping_records(self):
428
pack_data = StringIO()
429
writer = pack.ContainerWriter(pack_data.write)
432
memos.append(writer.add_bytes_record('abc', names=[]))
433
memos.append(writer.add_bytes_record('def', names=[('name1', )]))
434
memos.append(writer.add_bytes_record('ghi', names=[('name2', )]))
435
memos.append(writer.add_bytes_record('jkl', names=[]))
437
transport = self.get_transport()
438
transport.put_bytes('mypack', pack_data.getvalue())
439
requested_records = [memos[0], memos[2]]
440
reader = pack.make_readv_reader(transport, 'mypack', requested_records)
442
for names, reader_func in reader.iter_records():
443
result.append((names, reader_func(None)))
444
self.assertEqual([([], 'abc'), ([('name2', )], 'ghi')], result)
447
class TestReadvFile(tests.TestCaseWithTransport):
448
"""Tests of the ReadVFile class.
450
Error cases are deliberately undefined: this code adapts the underlying
451
transport interface to a single 'streaming read' interface as
452
ContainerReader needs.
455
def test_read_bytes(self):
456
"""Test reading of both single bytes and all bytes in a hunk."""
457
transport = self.get_transport()
458
transport.put_bytes('sample', '0123456789')
459
f = pack.ReadVFile(transport.readv('sample', [(0,1), (1,2), (4,1), (6,2)]))
461
results.append(f.read(1))
462
results.append(f.read(2))
463
results.append(f.read(1))
464
results.append(f.read(1))
465
results.append(f.read(1))
466
self.assertEqual(['0', '12', '4', '6', '7'], results)
468
def test_readline(self):
469
"""Test using readline() as ContainerReader does.
471
This is always within a readv hunk, never across it.
473
transport = self.get_transport()
474
transport.put_bytes('sample', '0\n2\n4\n')
475
f = pack.ReadVFile(transport.readv('sample', [(0,2), (2,4)]))
477
results.append(f.readline())
478
results.append(f.readline())
479
results.append(f.readline())
480
self.assertEqual(['0\n', '2\n', '4\n'], results)
482
def test_readline_and_read(self):
483
"""Test exercising one byte reads, readline, and then read again."""
484
transport = self.get_transport()
485
transport.put_bytes('sample', '0\n2\n4\n')
486
f = pack.ReadVFile(transport.readv('sample', [(0,6)]))
488
results.append(f.read(1))
489
results.append(f.readline())
490
results.append(f.read(4))
491
self.assertEqual(['0', '\n', '2\n4\n'], results)