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
writer.add_bytes_record('abc', names=[])
58
self.assertEqual('Bazaar pack format 1 (introduced in 0.18)\nB3\n\nabc',
61
def test_add_bytes_record_one_name(self):
62
"""Add a bytes record with one name."""
64
writer = pack.ContainerWriter(output.write)
66
writer.add_bytes_record('abc', names=['name1'])
68
'Bazaar pack format 1 (introduced in 0.18)\n'
72
def test_add_bytes_record_two_names(self):
73
"""Add a bytes record with two names."""
75
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',
83
def test_add_bytes_record_invalid_name(self):
84
"""Adding a Bytes record with a name with whitespace in it raises
88
writer = pack.ContainerWriter(output.write)
91
errors.InvalidRecordError,
92
writer.add_bytes_record, 'abc', names=['bad name'])
95
class TestContainerReader(tests.TestCase):
97
def get_reader_for(self, bytes):
98
stream = StringIO(bytes)
99
reader = pack.ContainerReader(stream)
102
def test_construct(self):
103
"""Test constructing a ContainerReader.
105
This uses None as the output stream to show that the constructor doesn't
106
try to use the input stream.
108
reader = pack.ContainerReader(None)
110
def test_empty_container(self):
111
"""Read an empty container."""
112
reader = self.get_reader_for(
113
"Bazaar pack format 1 (introduced in 0.18)\nE")
114
self.assertEqual([], list(reader.iter_records()))
116
def test_unknown_format(self):
117
"""Unrecognised container formats raise UnknownContainerFormatError."""
118
reader = self.get_reader_for("unknown format\n")
120
errors.UnknownContainerFormatError, reader.iter_records)
122
def test_unexpected_end_of_container(self):
123
"""Containers that don't end with an End Marker record should cause
124
UnexpectedEndOfContainerError to be raised.
126
reader = self.get_reader_for(
127
"Bazaar pack format 1 (introduced in 0.18)\n")
128
iterator = reader.iter_records()
130
errors.UnexpectedEndOfContainerError, iterator.next)
132
def test_unknown_record_type(self):
133
"""Unknown record types cause UnknownRecordTypeError to be raised."""
134
reader = self.get_reader_for(
135
"Bazaar pack format 1 (introduced in 0.18)\nX")
136
iterator = reader.iter_records()
138
errors.UnknownRecordTypeError, iterator.next)
140
def test_container_with_one_unnamed_record(self):
141
"""Read a container with one Bytes record.
143
Parsing Bytes records is more thoroughly exercised by
144
TestBytesRecordReader. This test is here to ensure that
145
ContainerReader's integration with BytesRecordReader is working.
147
reader = self.get_reader_for(
148
"Bazaar pack format 1 (introduced in 0.18)\nB5\n\naaaaaE")
149
expected_records = [([], 'aaaaa')]
152
[(names, read_bytes(None))
153
for (names, read_bytes) in reader.iter_records()])
155
def test_validate_empty_container(self):
156
"""validate does not raise an error for a container with no records."""
157
reader = self.get_reader_for("Bazaar pack format 1 (introduced in 0.18)\nE")
158
# No exception raised
161
def test_validate_non_empty_valid_container(self):
162
"""validate does not raise an error for a container with a valid record.
164
reader = self.get_reader_for(
165
"Bazaar pack format 1 (introduced in 0.18)\nB3\nname\n\nabcE")
166
# No exception raised
169
def test_validate_bad_format(self):
170
"""validate raises an error for unrecognised format strings.
172
It may raise either UnexpectedEndOfContainerError or
173
UnknownContainerFormatError, depending on exactly what the string is.
175
inputs = ["", "x", "Bazaar pack format 1 (introduced in 0.18)", "bad\n"]
177
reader = self.get_reader_for(input)
179
(errors.UnexpectedEndOfContainerError,
180
errors.UnknownContainerFormatError),
183
def test_validate_bad_record_marker(self):
184
"""validate raises UnknownRecordTypeError for unrecognised record
187
reader = self.get_reader_for(
188
"Bazaar pack format 1 (introduced in 0.18)\nX")
189
self.assertRaises(errors.UnknownRecordTypeError, reader.validate)
191
def test_validate_data_after_end_marker(self):
192
"""validate raises ContainerHasExcessDataError if there are any bytes
193
after the end of the container.
195
reader = self.get_reader_for(
196
"Bazaar pack format 1 (introduced in 0.18)\nEcrud")
198
errors.ContainerHasExcessDataError, reader.validate)
200
def test_validate_no_end_marker(self):
201
"""validate raises UnexpectedEndOfContainerError if there's no end of
202
container marker, even if the container up to this point has been valid.
204
reader = self.get_reader_for(
205
"Bazaar pack format 1 (introduced in 0.18)\n")
207
errors.UnexpectedEndOfContainerError, reader.validate)
209
def test_validate_duplicate_name(self):
210
"""validate raises DuplicateRecordNameError if the same name occurs
211
multiple times in the container.
213
reader = self.get_reader_for(
214
"Bazaar pack format 1 (introduced in 0.18)\n"
218
self.assertRaises(errors.DuplicateRecordNameError, reader.validate)
220
def test_validate_undecodeable_name(self):
221
"""Names that aren't valid UTF-8 cause validate to fail."""
222
reader = self.get_reader_for(
223
"Bazaar pack format 1 (introduced in 0.18)\nB0\n\xcc\n\nE")
224
self.assertRaises(errors.InvalidRecordError, reader.validate)
227
class TestBytesRecordReader(tests.TestCase):
228
"""Tests for reading and validating Bytes records with BytesRecordReader."""
230
def get_reader_for(self, bytes):
231
stream = StringIO(bytes)
232
reader = pack.BytesRecordReader(stream)
235
def test_record_with_no_name(self):
236
"""Reading a Bytes record with no name returns an empty list of
239
reader = self.get_reader_for("5\n\naaaaa")
240
names, get_bytes = reader.read()
241
self.assertEqual([], names)
242
self.assertEqual('aaaaa', get_bytes(None))
244
def test_record_with_one_name(self):
245
"""Reading a Bytes record with one name returns a list of just that
248
reader = self.get_reader_for("5\nname1\n\naaaaa")
249
names, get_bytes = reader.read()
250
self.assertEqual(['name1'], names)
251
self.assertEqual('aaaaa', get_bytes(None))
253
def test_record_with_two_names(self):
254
"""Reading a Bytes record with two names returns a list of both names.
256
reader = self.get_reader_for("5\nname1\nname2\n\naaaaa")
257
names, get_bytes = reader.read()
258
self.assertEqual(['name1', 'name2'], names)
259
self.assertEqual('aaaaa', get_bytes(None))
261
def test_invalid_length(self):
262
"""If the length-prefix is not a number, parsing raises
265
reader = self.get_reader_for("not a number\n")
266
self.assertRaises(errors.InvalidRecordError, reader.read)
268
def test_early_eof(self):
269
"""Tests for premature EOF occuring during parsing Bytes records with
272
A incomplete container might be interrupted at any point. The
273
BytesRecordReader needs to cope with the input stream running out no
274
matter where it is in the parsing process.
276
In all cases, UnexpectedEndOfContainerError should be raised.
278
complete_record = "6\nname\n\nabcdef"
279
for count in range(0, len(complete_record)):
280
incomplete_record = complete_record[:count]
281
reader = self.get_reader_for(incomplete_record)
282
# We don't use assertRaises to make diagnosing failures easier
283
# (assertRaises doesn't allow a custom failure message).
285
names, read_bytes = reader.read()
287
except errors.UnexpectedEndOfContainerError:
291
"UnexpectedEndOfContainerError not raised when parsing %r"
292
% (incomplete_record,))
294
def test_initial_eof(self):
295
"""EOF before any bytes read at all."""
296
reader = self.get_reader_for("")
297
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
299
def test_eof_after_length(self):
300
"""EOF after reading the length and before reading name(s)."""
301
reader = self.get_reader_for("123\n")
302
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
304
def test_eof_during_name(self):
305
"""EOF during reading a name."""
306
reader = self.get_reader_for("123\nname")
307
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
309
def test_read_invalid_name_whitespace(self):
310
"""Names must have no whitespace."""
311
# A name with a space.
312
reader = self.get_reader_for("0\nbad name\n\n")
313
self.assertRaises(errors.InvalidRecordError, reader.read)
316
reader = self.get_reader_for("0\nbad\tname\n\n")
317
self.assertRaises(errors.InvalidRecordError, reader.read)
319
# A name with a vertical tab.
320
reader = self.get_reader_for("0\nbad\vname\n\n")
321
self.assertRaises(errors.InvalidRecordError, reader.read)
323
def test_validate_whitespace_in_name(self):
324
"""Names must have no whitespace."""
325
reader = self.get_reader_for("0\nbad name\n\n")
326
self.assertRaises(errors.InvalidRecordError, reader.validate)
328
def test_validate_interrupted_prelude(self):
329
"""EOF during reading a record's prelude causes validate to fail."""
330
reader = self.get_reader_for("")
332
errors.UnexpectedEndOfContainerError, reader.validate)
334
def test_validate_interrupted_body(self):
335
"""EOF during reading a record's body causes validate to fail."""
336
reader = self.get_reader_for("1\n\n")
338
errors.UnexpectedEndOfContainerError, reader.validate)
340
def test_validate_unparseable_length(self):
341
"""An unparseable record length causes validate to fail."""
342
reader = self.get_reader_for("\n\n")
344
errors.InvalidRecordError, reader.validate)
346
def test_validate_undecodeable_name(self):
347
"""Names that aren't valid UTF-8 cause validate to fail."""
348
reader = self.get_reader_for("0\n\xcc\n\n")
349
self.assertRaises(errors.InvalidRecordError, reader.validate)
351
def test_read_max_length(self):
352
"""If the max_length passed to the callable returned by read is not
353
None, then no more than that many bytes will be read.
355
reader = self.get_reader_for("6\n\nabcdef")
356
names, get_bytes = reader.read()
357
self.assertEqual('abc', get_bytes(3))
359
def test_read_no_max_length(self):
360
"""If the max_length passed to the callable returned by read is None,
361
then all the bytes in the record will be read.
363
reader = self.get_reader_for("6\n\nabcdef")
364
names, get_bytes = reader.read()
365
self.assertEqual('abcdef', get_bytes(None))
367
def test_repeated_read_calls(self):
368
"""Repeated calls to the callable returned from BytesRecordReader.read
369
will not read beyond the end of the record.
371
reader = self.get_reader_for("6\n\nabcdefB3\nnext-record\nXXX")
372
names, get_bytes = reader.read()
373
self.assertEqual('abcdef', get_bytes(None))
374
self.assertEqual('', get_bytes(None))
375
self.assertEqual('', get_bytes(99))
added
removed
bzrlib/tests/test_pack.py
