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_second_bytes_record_gets_higher_offset(self):
88
writer = pack.ContainerWriter(output.write)
90
writer.add_bytes_record('abc', names=[])
91
offset, length = writer.add_bytes_record('abc', names=[])
92
self.assertEqual((49, 7), (offset, length))
94
'Bazaar pack format 1 (introduced in 0.18)\n'
99
def test_add_bytes_record_invalid_name(self):
100
"""Adding a Bytes record with a name with whitespace in it raises
104
writer = pack.ContainerWriter(output.write)
107
errors.InvalidRecordError,
108
writer.add_bytes_record, 'abc', names=['bad name'])
111
class TestContainerReader(tests.TestCase):
113
def get_reader_for(self, bytes):
114
stream = StringIO(bytes)
115
reader = pack.ContainerReader(stream)
118
def test_construct(self):
119
"""Test constructing a ContainerReader.
121
This uses None as the output stream to show that the constructor doesn't
122
try to use the input stream.
124
reader = pack.ContainerReader(None)
126
def test_empty_container(self):
127
"""Read an empty container."""
128
reader = self.get_reader_for(
129
"Bazaar pack format 1 (introduced in 0.18)\nE")
130
self.assertEqual([], list(reader.iter_records()))
132
def test_unknown_format(self):
133
"""Unrecognised container formats raise UnknownContainerFormatError."""
134
reader = self.get_reader_for("unknown format\n")
136
errors.UnknownContainerFormatError, reader.iter_records)
138
def test_unexpected_end_of_container(self):
139
"""Containers that don't end with an End Marker record should cause
140
UnexpectedEndOfContainerError to be raised.
142
reader = self.get_reader_for(
143
"Bazaar pack format 1 (introduced in 0.18)\n")
144
iterator = reader.iter_records()
146
errors.UnexpectedEndOfContainerError, iterator.next)
148
def test_unknown_record_type(self):
149
"""Unknown record types cause UnknownRecordTypeError to be raised."""
150
reader = self.get_reader_for(
151
"Bazaar pack format 1 (introduced in 0.18)\nX")
152
iterator = reader.iter_records()
154
errors.UnknownRecordTypeError, iterator.next)
156
def test_container_with_one_unnamed_record(self):
157
"""Read a container with one Bytes record.
159
Parsing Bytes records is more thoroughly exercised by
160
TestBytesRecordReader. This test is here to ensure that
161
ContainerReader's integration with BytesRecordReader is working.
163
reader = self.get_reader_for(
164
"Bazaar pack format 1 (introduced in 0.18)\nB5\n\naaaaaE")
165
expected_records = [([], 'aaaaa')]
168
[(names, read_bytes(None))
169
for (names, read_bytes) in reader.iter_records()])
171
def test_validate_empty_container(self):
172
"""validate does not raise an error for a container with no records."""
173
reader = self.get_reader_for("Bazaar pack format 1 (introduced in 0.18)\nE")
174
# No exception raised
177
def test_validate_non_empty_valid_container(self):
178
"""validate does not raise an error for a container with a valid record.
180
reader = self.get_reader_for(
181
"Bazaar pack format 1 (introduced in 0.18)\nB3\nname\n\nabcE")
182
# No exception raised
185
def test_validate_bad_format(self):
186
"""validate raises an error for unrecognised format strings.
188
It may raise either UnexpectedEndOfContainerError or
189
UnknownContainerFormatError, depending on exactly what the string is.
191
inputs = ["", "x", "Bazaar pack format 1 (introduced in 0.18)", "bad\n"]
193
reader = self.get_reader_for(input)
195
(errors.UnexpectedEndOfContainerError,
196
errors.UnknownContainerFormatError),
199
def test_validate_bad_record_marker(self):
200
"""validate raises UnknownRecordTypeError for unrecognised record
203
reader = self.get_reader_for(
204
"Bazaar pack format 1 (introduced in 0.18)\nX")
205
self.assertRaises(errors.UnknownRecordTypeError, reader.validate)
207
def test_validate_data_after_end_marker(self):
208
"""validate raises ContainerHasExcessDataError if there are any bytes
209
after the end of the container.
211
reader = self.get_reader_for(
212
"Bazaar pack format 1 (introduced in 0.18)\nEcrud")
214
errors.ContainerHasExcessDataError, reader.validate)
216
def test_validate_no_end_marker(self):
217
"""validate raises UnexpectedEndOfContainerError if there's no end of
218
container marker, even if the container up to this point has been valid.
220
reader = self.get_reader_for(
221
"Bazaar pack format 1 (introduced in 0.18)\n")
223
errors.UnexpectedEndOfContainerError, reader.validate)
225
def test_validate_duplicate_name(self):
226
"""validate raises DuplicateRecordNameError if the same name occurs
227
multiple times in the container.
229
reader = self.get_reader_for(
230
"Bazaar pack format 1 (introduced in 0.18)\n"
234
self.assertRaises(errors.DuplicateRecordNameError, reader.validate)
236
def test_validate_undecodeable_name(self):
237
"""Names that aren't valid UTF-8 cause validate to fail."""
238
reader = self.get_reader_for(
239
"Bazaar pack format 1 (introduced in 0.18)\nB0\n\xcc\n\nE")
240
self.assertRaises(errors.InvalidRecordError, reader.validate)
243
class TestBytesRecordReader(tests.TestCase):
244
"""Tests for reading and validating Bytes records with BytesRecordReader."""
246
def get_reader_for(self, bytes):
247
stream = StringIO(bytes)
248
reader = pack.BytesRecordReader(stream)
251
def test_record_with_no_name(self):
252
"""Reading a Bytes record with no name returns an empty list of
255
reader = self.get_reader_for("5\n\naaaaa")
256
names, get_bytes = reader.read()
257
self.assertEqual([], names)
258
self.assertEqual('aaaaa', get_bytes(None))
260
def test_record_with_one_name(self):
261
"""Reading a Bytes record with one name returns a list of just that
264
reader = self.get_reader_for("5\nname1\n\naaaaa")
265
names, get_bytes = reader.read()
266
self.assertEqual(['name1'], names)
267
self.assertEqual('aaaaa', get_bytes(None))
269
def test_record_with_two_names(self):
270
"""Reading a Bytes record with two names returns a list of both names.
272
reader = self.get_reader_for("5\nname1\nname2\n\naaaaa")
273
names, get_bytes = reader.read()
274
self.assertEqual(['name1', 'name2'], names)
275
self.assertEqual('aaaaa', get_bytes(None))
277
def test_invalid_length(self):
278
"""If the length-prefix is not a number, parsing raises
281
reader = self.get_reader_for("not a number\n")
282
self.assertRaises(errors.InvalidRecordError, reader.read)
284
def test_early_eof(self):
285
"""Tests for premature EOF occuring during parsing Bytes records with
288
A incomplete container might be interrupted at any point. The
289
BytesRecordReader needs to cope with the input stream running out no
290
matter where it is in the parsing process.
292
In all cases, UnexpectedEndOfContainerError should be raised.
294
complete_record = "6\nname\n\nabcdef"
295
for count in range(0, len(complete_record)):
296
incomplete_record = complete_record[:count]
297
reader = self.get_reader_for(incomplete_record)
298
# We don't use assertRaises to make diagnosing failures easier
299
# (assertRaises doesn't allow a custom failure message).
301
names, read_bytes = reader.read()
303
except errors.UnexpectedEndOfContainerError:
307
"UnexpectedEndOfContainerError not raised when parsing %r"
308
% (incomplete_record,))
310
def test_initial_eof(self):
311
"""EOF before any bytes read at all."""
312
reader = self.get_reader_for("")
313
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
315
def test_eof_after_length(self):
316
"""EOF after reading the length and before reading name(s)."""
317
reader = self.get_reader_for("123\n")
318
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
320
def test_eof_during_name(self):
321
"""EOF during reading a name."""
322
reader = self.get_reader_for("123\nname")
323
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
325
def test_read_invalid_name_whitespace(self):
326
"""Names must have no whitespace."""
327
# A name with a space.
328
reader = self.get_reader_for("0\nbad name\n\n")
329
self.assertRaises(errors.InvalidRecordError, reader.read)
332
reader = self.get_reader_for("0\nbad\tname\n\n")
333
self.assertRaises(errors.InvalidRecordError, reader.read)
335
# A name with a vertical tab.
336
reader = self.get_reader_for("0\nbad\vname\n\n")
337
self.assertRaises(errors.InvalidRecordError, reader.read)
339
def test_validate_whitespace_in_name(self):
340
"""Names must have no whitespace."""
341
reader = self.get_reader_for("0\nbad name\n\n")
342
self.assertRaises(errors.InvalidRecordError, reader.validate)
344
def test_validate_interrupted_prelude(self):
345
"""EOF during reading a record's prelude causes validate to fail."""
346
reader = self.get_reader_for("")
348
errors.UnexpectedEndOfContainerError, reader.validate)
350
def test_validate_interrupted_body(self):
351
"""EOF during reading a record's body causes validate to fail."""
352
reader = self.get_reader_for("1\n\n")
354
errors.UnexpectedEndOfContainerError, reader.validate)
356
def test_validate_unparseable_length(self):
357
"""An unparseable record length causes validate to fail."""
358
reader = self.get_reader_for("\n\n")
360
errors.InvalidRecordError, reader.validate)
362
def test_validate_undecodeable_name(self):
363
"""Names that aren't valid UTF-8 cause validate to fail."""
364
reader = self.get_reader_for("0\n\xcc\n\n")
365
self.assertRaises(errors.InvalidRecordError, reader.validate)
367
def test_read_max_length(self):
368
"""If the max_length passed to the callable returned by read is not
369
None, then no more than that many bytes will be read.
371
reader = self.get_reader_for("6\n\nabcdef")
372
names, get_bytes = reader.read()
373
self.assertEqual('abc', get_bytes(3))
375
def test_read_no_max_length(self):
376
"""If the max_length passed to the callable returned by read is None,
377
then all the bytes in the record will be read.
379
reader = self.get_reader_for("6\n\nabcdef")
380
names, get_bytes = reader.read()
381
self.assertEqual('abcdef', get_bytes(None))
383
def test_repeated_read_calls(self):
384
"""Repeated calls to the callable returned from BytesRecordReader.read
385
will not read beyond the end of the record.
387
reader = self.get_reader_for("6\n\nabcdefB3\nnext-record\nXXX")
388
names, get_bytes = reader.read()
389
self.assertEqual('abcdef', get_bytes(None))
390
self.assertEqual('', get_bytes(None))
391
self.assertEqual('', get_bytes(99))
394
class TestMakeReadvReader(tests.TestCaseWithTransport):
396
def test_read_skipping_records(self):
397
pack_data = StringIO()
398
writer = pack.ContainerWriter(pack_data.write)
401
memos.append(writer.add_bytes_record('abc', names=[]))
402
memos.append(writer.add_bytes_record('def', names=['name1']))
403
memos.append(writer.add_bytes_record('ghi', names=['name2']))
404
memos.append(writer.add_bytes_record('jkl', names=[]))
406
transport = self.get_transport()
407
transport.put_bytes('mypack', pack_data.getvalue())
408
requested_records = [memos[0], memos[2]]
409
reader = pack.make_readv_reader(transport, 'mypack', requested_records)
411
for names, reader_func in reader.iter_records():
412
result.append((names, reader_func(None)))
413
self.assertEqual([([], 'abc'), (['name2'], 'ghi')], result)
416
class TestReadvFile(tests.TestCaseWithTransport):
417
"""Tests of the ReadVFile class.
419
Error cases are deliberately undefined: this code adapts the underlying
420
transport interface to a single 'streaming read' interface as
421
ContainerReader needs.
424
def test_read_bytes(self):
425
"""Test reading of both single bytes and all bytes in a hunk."""
426
transport = self.get_transport()
427
transport.put_bytes('sample', '0123456789')
428
f = pack.ReadVFile(transport.readv('sample', [(0,1), (1,2), (4,1), (6,2)]))
430
results.append(f.read(1))
431
results.append(f.read(2))
432
results.append(f.read(1))
433
results.append(f.read(1))
434
results.append(f.read(1))
435
self.assertEqual(['0', '12', '4', '6', '7'], results)
437
def test_readline(self):
438
"""Test using readline() as ContainerReader does.
440
This is always within a readv hunk, never across it.
442
transport = self.get_transport()
443
transport.put_bytes('sample', '0\n2\n4\n')
444
f = pack.ReadVFile(transport.readv('sample', [(0,2), (2,4)]))
446
results.append(f.readline())
447
results.append(f.readline())
448
results.append(f.readline())
449
self.assertEqual(['0\n', '2\n', '4\n'], results)
451
def test_readline_and_read(self):
452
"""Test exercising one byte reads, readline, and then read again."""
453
transport = self.get_transport()
454
transport.put_bytes('sample', '0\n2\n4\n')
455
f = pack.ReadVFile(transport.readv('sample', [(0,6)]))
457
results.append(f.read(1))
458
results.append(f.readline())
459
results.append(f.read(4))
460
self.assertEqual(['0', '\n', '2\n4\n'], results)
added
removed
bzrlib/tests/test_pack.py
