1
# Copyright (C) 2007 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
"""Tests for bzrlib.pack."""
20
from cStringIO import StringIO
22
from bzrlib import pack, errors, tests
25
class TestContainerWriter(tests.TestCase):
27
def test_construct(self):
28
"""Test constructing a ContainerWriter.
30
This uses None as the output stream to show that the constructor doesn't
31
try to use the output stream.
33
writer = pack.ContainerWriter(None)
36
"""The begin() method writes the container format marker line."""
38
writer = pack.ContainerWriter(output.write)
40
self.assertEqual('Bazaar pack format 1 (introduced in 0.18)\n',
44
"""The end() method writes an End Marker record."""
46
writer = pack.ContainerWriter(output.write)
49
self.assertEqual('Bazaar pack format 1 (introduced in 0.18)\nE',
52
def test_add_bytes_record_no_name(self):
53
"""Add a bytes record with no name."""
55
writer = pack.ContainerWriter(output.write)
57
offset, length = writer.add_bytes_record('abc', names=[])
58
self.assertEqual((42, 7), (offset, length))
59
self.assertEqual('Bazaar pack format 1 (introduced in 0.18)\nB3\n\nabc',
62
def test_add_bytes_record_one_name(self):
63
"""Add a bytes record with one name."""
65
writer = pack.ContainerWriter(output.write)
67
offset, length = writer.add_bytes_record('abc', names=[('name1', )])
68
self.assertEqual((42, 13), (offset, length))
70
'Bazaar pack format 1 (introduced in 0.18)\n'
74
def test_add_bytes_record_two_names(self):
75
"""Add a bytes record with two names."""
77
writer = pack.ContainerWriter(output.write)
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'
123
def test_add_bytes_record_invalid_name(self):
124
"""Adding a Bytes record with a name with whitespace in it raises
128
writer = pack.ContainerWriter(output.write)
131
errors.InvalidRecordError,
132
writer.add_bytes_record, 'abc', names=[('bad name', )])
135
class TestContainerReader(tests.TestCase):
137
def get_reader_for(self, bytes):
138
stream = StringIO(bytes)
139
reader = pack.ContainerReader(stream)
142
def test_construct(self):
143
"""Test constructing a ContainerReader.
145
This uses None as the output stream to show that the constructor doesn't
146
try to use the input stream.
148
reader = pack.ContainerReader(None)
150
def test_empty_container(self):
151
"""Read an empty container."""
152
reader = self.get_reader_for(
153
"Bazaar pack format 1 (introduced in 0.18)\nE")
154
self.assertEqual([], list(reader.iter_records()))
156
def test_unknown_format(self):
157
"""Unrecognised container formats raise UnknownContainerFormatError."""
158
reader = self.get_reader_for("unknown format\n")
160
errors.UnknownContainerFormatError, reader.iter_records)
162
def test_unexpected_end_of_container(self):
163
"""Containers that don't end with an End Marker record should cause
164
UnexpectedEndOfContainerError to be raised.
166
reader = self.get_reader_for(
167
"Bazaar pack format 1 (introduced in 0.18)\n")
168
iterator = reader.iter_records()
170
errors.UnexpectedEndOfContainerError, iterator.next)
172
def test_unknown_record_type(self):
173
"""Unknown record types cause UnknownRecordTypeError to be raised."""
174
reader = self.get_reader_for(
175
"Bazaar pack format 1 (introduced in 0.18)\nX")
176
iterator = reader.iter_records()
178
errors.UnknownRecordTypeError, iterator.next)
180
def test_container_with_one_unnamed_record(self):
181
"""Read a container with one Bytes record.
183
Parsing Bytes records is more thoroughly exercised by
184
TestBytesRecordReader. This test is here to ensure that
185
ContainerReader's integration with BytesRecordReader is working.
187
reader = self.get_reader_for(
188
"Bazaar pack format 1 (introduced in 0.18)\nB5\n\naaaaaE")
189
expected_records = [([], 'aaaaa')]
192
[(names, read_bytes(None))
193
for (names, read_bytes) in reader.iter_records()])
195
def test_validate_empty_container(self):
196
"""validate does not raise an error for a container with no records."""
197
reader = self.get_reader_for("Bazaar pack format 1 (introduced in 0.18)\nE")
198
# No exception raised
201
def test_validate_non_empty_valid_container(self):
202
"""validate does not raise an error for a container with a valid record.
204
reader = self.get_reader_for(
205
"Bazaar pack format 1 (introduced in 0.18)\nB3\nname\n\nabcE")
206
# No exception raised
209
def test_validate_bad_format(self):
210
"""validate raises an error for unrecognised format strings.
212
It may raise either UnexpectedEndOfContainerError or
213
UnknownContainerFormatError, depending on exactly what the string is.
215
inputs = ["", "x", "Bazaar pack format 1 (introduced in 0.18)", "bad\n"]
217
reader = self.get_reader_for(input)
219
(errors.UnexpectedEndOfContainerError,
220
errors.UnknownContainerFormatError),
223
def test_validate_bad_record_marker(self):
224
"""validate raises UnknownRecordTypeError for unrecognised record
227
reader = self.get_reader_for(
228
"Bazaar pack format 1 (introduced in 0.18)\nX")
229
self.assertRaises(errors.UnknownRecordTypeError, reader.validate)
231
def test_validate_data_after_end_marker(self):
232
"""validate raises ContainerHasExcessDataError if there are any bytes
233
after the end of the container.
235
reader = self.get_reader_for(
236
"Bazaar pack format 1 (introduced in 0.18)\nEcrud")
238
errors.ContainerHasExcessDataError, reader.validate)
240
def test_validate_no_end_marker(self):
241
"""validate raises UnexpectedEndOfContainerError if there's no end of
242
container marker, even if the container up to this point has been valid.
244
reader = self.get_reader_for(
245
"Bazaar pack format 1 (introduced in 0.18)\n")
247
errors.UnexpectedEndOfContainerError, reader.validate)
249
def test_validate_duplicate_name(self):
250
"""validate raises DuplicateRecordNameError if the same name occurs
251
multiple times in the container.
253
reader = self.get_reader_for(
254
"Bazaar pack format 1 (introduced in 0.18)\n"
258
self.assertRaises(errors.DuplicateRecordNameError, reader.validate)
260
def test_validate_undecodeable_name(self):
261
"""Names that aren't valid UTF-8 cause validate to fail."""
262
reader = self.get_reader_for(
263
"Bazaar pack format 1 (introduced in 0.18)\nB0\n\xcc\n\nE")
264
self.assertRaises(errors.InvalidRecordError, reader.validate)
267
class TestBytesRecordReader(tests.TestCase):
268
"""Tests for reading and validating Bytes records with BytesRecordReader."""
270
def get_reader_for(self, bytes):
271
stream = StringIO(bytes)
272
reader = pack.BytesRecordReader(stream)
275
def test_record_with_no_name(self):
276
"""Reading a Bytes record with no name returns an empty list of
279
reader = self.get_reader_for("5\n\naaaaa")
280
names, get_bytes = reader.read()
281
self.assertEqual([], names)
282
self.assertEqual('aaaaa', get_bytes(None))
284
def test_record_with_one_name(self):
285
"""Reading a Bytes record with one name returns a list of just that
288
reader = self.get_reader_for("5\nname1\n\naaaaa")
289
names, get_bytes = reader.read()
290
self.assertEqual([('name1', )], names)
291
self.assertEqual('aaaaa', get_bytes(None))
293
def test_record_with_two_names(self):
294
"""Reading a Bytes record with two names returns a list of both names.
296
reader = self.get_reader_for("5\nname1\nname2\n\naaaaa")
297
names, get_bytes = reader.read()
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)
306
self.assertEqual('aaaaa', get_bytes(None))
308
def test_invalid_length(self):
309
"""If the length-prefix is not a number, parsing raises
312
reader = self.get_reader_for("not a number\n")
313
self.assertRaises(errors.InvalidRecordError, reader.read)
315
def test_early_eof(self):
316
"""Tests for premature EOF occuring during parsing Bytes records with
319
A incomplete container might be interrupted at any point. The
320
BytesRecordReader needs to cope with the input stream running out no
321
matter where it is in the parsing process.
323
In all cases, UnexpectedEndOfContainerError should be raised.
325
complete_record = "6\nname\n\nabcdef"
326
for count in range(0, len(complete_record)):
327
incomplete_record = complete_record[:count]
328
reader = self.get_reader_for(incomplete_record)
329
# We don't use assertRaises to make diagnosing failures easier
330
# (assertRaises doesn't allow a custom failure message).
332
names, read_bytes = reader.read()
334
except errors.UnexpectedEndOfContainerError:
338
"UnexpectedEndOfContainerError not raised when parsing %r"
339
% (incomplete_record,))
341
def test_initial_eof(self):
342
"""EOF before any bytes read at all."""
343
reader = self.get_reader_for("")
344
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
346
def test_eof_after_length(self):
347
"""EOF after reading the length and before reading name(s)."""
348
reader = self.get_reader_for("123\n")
349
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
351
def test_eof_during_name(self):
352
"""EOF during reading a name."""
353
reader = self.get_reader_for("123\nname")
354
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
356
def test_read_invalid_name_whitespace(self):
357
"""Names must have no whitespace."""
358
# A name with a space.
359
reader = self.get_reader_for("0\nbad name\n\n")
360
self.assertRaises(errors.InvalidRecordError, reader.read)
363
reader = self.get_reader_for("0\nbad\tname\n\n")
364
self.assertRaises(errors.InvalidRecordError, reader.read)
366
# A name with a vertical tab.
367
reader = self.get_reader_for("0\nbad\vname\n\n")
368
self.assertRaises(errors.InvalidRecordError, reader.read)
370
def test_validate_whitespace_in_name(self):
371
"""Names must have no whitespace."""
372
reader = self.get_reader_for("0\nbad name\n\n")
373
self.assertRaises(errors.InvalidRecordError, reader.validate)
375
def test_validate_interrupted_prelude(self):
376
"""EOF during reading a record's prelude causes validate to fail."""
377
reader = self.get_reader_for("")
379
errors.UnexpectedEndOfContainerError, reader.validate)
381
def test_validate_interrupted_body(self):
382
"""EOF during reading a record's body causes validate to fail."""
383
reader = self.get_reader_for("1\n\n")
385
errors.UnexpectedEndOfContainerError, reader.validate)
387
def test_validate_unparseable_length(self):
388
"""An unparseable record length causes validate to fail."""
389
reader = self.get_reader_for("\n\n")
391
errors.InvalidRecordError, reader.validate)
393
def test_validate_undecodeable_name(self):
394
"""Names that aren't valid UTF-8 cause validate to fail."""
395
reader = self.get_reader_for("0\n\xcc\n\n")
396
self.assertRaises(errors.InvalidRecordError, reader.validate)
398
def test_read_max_length(self):
399
"""If the max_length passed to the callable returned by read is not
400
None, then no more than that many bytes will be read.
402
reader = self.get_reader_for("6\n\nabcdef")
403
names, get_bytes = reader.read()
404
self.assertEqual('abc', get_bytes(3))
406
def test_read_no_max_length(self):
407
"""If the max_length passed to the callable returned by read is None,
408
then all the bytes in the record will be read.
410
reader = self.get_reader_for("6\n\nabcdef")
411
names, get_bytes = reader.read()
412
self.assertEqual('abcdef', get_bytes(None))
414
def test_repeated_read_calls(self):
415
"""Repeated calls to the callable returned from BytesRecordReader.read
416
will not read beyond the end of the record.
418
reader = self.get_reader_for("6\n\nabcdefB3\nnext-record\nXXX")
419
names, get_bytes = reader.read()
420
self.assertEqual('abcdef', get_bytes(None))
421
self.assertEqual('', get_bytes(None))
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)
added
removed
bzrlib/tests/test_pack.py
