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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Tests for bzrlib.pack."""
20
from cStringIO import StringIO
22
from bzrlib import pack, errors, tests
25
class TestContainerSerialiser(tests.TestCase):
26
"""Tests for the ContainerSerialiser class."""
28
def test_construct(self):
29
"""Test constructing a ContainerSerialiser."""
30
pack.ContainerSerialiser()
33
serialiser = pack.ContainerSerialiser()
34
self.assertEqual('Bazaar pack format 1 (introduced in 0.18)\n',
38
serialiser = pack.ContainerSerialiser()
39
self.assertEqual('E', serialiser.end())
41
def test_bytes_record_no_name(self):
42
serialiser = pack.ContainerSerialiser()
43
record = serialiser.bytes_record('bytes', [])
44
self.assertEqual('B5\n\nbytes', record)
46
def test_bytes_record_one_name_with_one_part(self):
47
serialiser = pack.ContainerSerialiser()
48
record = serialiser.bytes_record('bytes', [('name',)])
49
self.assertEqual('B5\nname\n\nbytes', record)
51
def test_bytes_record_one_name_with_two_parts(self):
52
serialiser = pack.ContainerSerialiser()
53
record = serialiser.bytes_record('bytes', [('part1', 'part2')])
54
self.assertEqual('B5\npart1\x00part2\n\nbytes', record)
56
def test_bytes_record_two_names(self):
57
serialiser = pack.ContainerSerialiser()
58
record = serialiser.bytes_record('bytes', [('name1',), ('name2',)])
59
self.assertEqual('B5\nname1\nname2\n\nbytes', record)
61
def test_bytes_record_whitespace_in_name_part(self):
62
serialiser = pack.ContainerSerialiser()
64
errors.InvalidRecordError,
65
serialiser.bytes_record, 'bytes', [('bad name',)])
67
def test_bytes_record_header(self):
68
serialiser = pack.ContainerSerialiser()
69
record = serialiser.bytes_header(32, [('name1',), ('name2',)])
70
self.assertEqual('B32\nname1\nname2\n\n', record)
73
class TestContainerWriter(tests.TestCase):
76
super(TestContainerWriter, self).setUp()
77
self.output = StringIO()
78
self.writer = pack.ContainerWriter(self.output.write)
80
def assertOutput(self, expected_output):
81
"""Assert that the output of self.writer ContainerWriter is equal to
84
self.assertEqual(expected_output, self.output.getvalue())
86
def test_construct(self):
87
"""Test constructing a ContainerWriter.
89
This uses None as the output stream to show that the constructor
90
doesn't try to use the output stream.
92
writer = pack.ContainerWriter(None)
95
"""The begin() method writes the container format marker line."""
97
self.assertOutput('Bazaar pack format 1 (introduced in 0.18)\n')
99
def test_zero_records_written_after_begin(self):
100
"""After begin is written, 0 records have been written."""
102
self.assertEqual(0, self.writer.records_written)
105
"""The end() method writes an End Marker record."""
108
self.assertOutput('Bazaar pack format 1 (introduced in 0.18)\nE')
110
def test_empty_end_does_not_add_a_record_to_records_written(self):
111
"""The end() method does not count towards the records written."""
114
self.assertEqual(0, self.writer.records_written)
116
def test_non_empty_end_does_not_add_a_record_to_records_written(self):
117
"""The end() method does not count towards the records written."""
119
self.writer.add_bytes_record('foo', names=[])
121
self.assertEqual(1, self.writer.records_written)
123
def test_add_bytes_record_no_name(self):
124
"""Add a bytes record with no name."""
126
offset, length = self.writer.add_bytes_record('abc', names=[])
127
self.assertEqual((42, 7), (offset, length))
129
'Bazaar pack format 1 (introduced in 0.18)\nB3\n\nabc')
131
def test_add_bytes_record_one_name(self):
132
"""Add a bytes record with one name."""
135
offset, length = self.writer.add_bytes_record(
136
'abc', names=[('name1', )])
137
self.assertEqual((42, 13), (offset, length))
139
'Bazaar pack format 1 (introduced in 0.18)\n'
142
def test_add_bytes_record_split_writes(self):
143
"""Write a large record which does multiple IOs"""
146
real_write = self.writer.write_func
148
def record_writes(bytes):
150
return real_write(bytes)
152
self.writer.write_func = record_writes
153
self.writer._JOIN_WRITES_THRESHOLD = 2
156
offset, length = self.writer.add_bytes_record(
157
'abcabc', names=[('name1', )])
158
self.assertEqual((42, 16), (offset, length))
160
'Bazaar pack format 1 (introduced in 0.18)\n'
161
'B6\nname1\n\nabcabc')
164
'Bazaar pack format 1 (introduced in 0.18)\n',
169
def test_add_bytes_record_two_names(self):
170
"""Add a bytes record with two names."""
172
offset, length = self.writer.add_bytes_record(
173
'abc', names=[('name1', ), ('name2', )])
174
self.assertEqual((42, 19), (offset, length))
176
'Bazaar pack format 1 (introduced in 0.18)\n'
177
'B3\nname1\nname2\n\nabc')
179
def test_add_bytes_record_two_names(self):
180
"""Add a bytes record with two names."""
182
offset, length = self.writer.add_bytes_record(
183
'abc', names=[('name1', ), ('name2', )])
184
self.assertEqual((42, 19), (offset, length))
186
'Bazaar pack format 1 (introduced in 0.18)\n'
187
'B3\nname1\nname2\n\nabc')
189
def test_add_bytes_record_two_element_name(self):
190
"""Add a bytes record with a two-element name."""
192
offset, length = self.writer.add_bytes_record(
193
'abc', names=[('name1', 'name2')])
194
self.assertEqual((42, 19), (offset, length))
196
'Bazaar pack format 1 (introduced in 0.18)\n'
197
'B3\nname1\x00name2\n\nabc')
199
def test_add_second_bytes_record_gets_higher_offset(self):
201
self.writer.add_bytes_record('abc', names=[])
202
offset, length = self.writer.add_bytes_record('abc', names=[])
203
self.assertEqual((49, 7), (offset, length))
205
'Bazaar pack format 1 (introduced in 0.18)\n'
209
def test_add_bytes_record_invalid_name(self):
210
"""Adding a Bytes record with a name with whitespace in it raises
215
errors.InvalidRecordError,
216
self.writer.add_bytes_record, 'abc', names=[('bad name', )])
218
def test_add_bytes_records_add_to_records_written(self):
219
"""Adding a Bytes record increments the records_written counter."""
221
self.writer.add_bytes_record('foo', names=[])
222
self.assertEqual(1, self.writer.records_written)
223
self.writer.add_bytes_record('foo', names=[])
224
self.assertEqual(2, self.writer.records_written)
227
class TestContainerReader(tests.TestCase):
228
"""Tests for the ContainerReader.
230
The ContainerReader reads format 1 containers, so these tests explicitly
231
test how it reacts to format 1 data. If a new version of the format is
232
added, then separate tests for that format should be added.
235
def get_reader_for(self, bytes):
236
stream = StringIO(bytes)
237
reader = pack.ContainerReader(stream)
240
def test_construct(self):
241
"""Test constructing a ContainerReader.
243
This uses None as the output stream to show that the constructor
244
doesn't try to use the input stream.
246
reader = pack.ContainerReader(None)
248
def test_empty_container(self):
249
"""Read an empty container."""
250
reader = self.get_reader_for(
251
"Bazaar pack format 1 (introduced in 0.18)\nE")
252
self.assertEqual([], list(reader.iter_records()))
254
def test_unknown_format(self):
255
"""Unrecognised container formats raise UnknownContainerFormatError."""
256
reader = self.get_reader_for("unknown format\n")
258
errors.UnknownContainerFormatError, reader.iter_records)
260
def test_unexpected_end_of_container(self):
261
"""Containers that don't end with an End Marker record should cause
262
UnexpectedEndOfContainerError to be raised.
264
reader = self.get_reader_for(
265
"Bazaar pack format 1 (introduced in 0.18)\n")
266
iterator = reader.iter_records()
268
errors.UnexpectedEndOfContainerError, iterator.next)
270
def test_unknown_record_type(self):
271
"""Unknown record types cause UnknownRecordTypeError to be raised."""
272
reader = self.get_reader_for(
273
"Bazaar pack format 1 (introduced in 0.18)\nX")
274
iterator = reader.iter_records()
276
errors.UnknownRecordTypeError, iterator.next)
278
def test_container_with_one_unnamed_record(self):
279
"""Read a container with one Bytes record.
281
Parsing Bytes records is more thoroughly exercised by
282
TestBytesRecordReader. This test is here to ensure that
283
ContainerReader's integration with BytesRecordReader is working.
285
reader = self.get_reader_for(
286
"Bazaar pack format 1 (introduced in 0.18)\nB5\n\naaaaaE")
287
expected_records = [([], 'aaaaa')]
290
[(names, read_bytes(None))
291
for (names, read_bytes) in reader.iter_records()])
293
def test_validate_empty_container(self):
294
"""validate does not raise an error for a container with no records."""
295
reader = self.get_reader_for(
296
"Bazaar pack format 1 (introduced in 0.18)\nE")
297
# No exception raised
300
def test_validate_non_empty_valid_container(self):
301
"""validate does not raise an error for a container with a valid record.
303
reader = self.get_reader_for(
304
"Bazaar pack format 1 (introduced in 0.18)\nB3\nname\n\nabcE")
305
# No exception raised
308
def test_validate_bad_format(self):
309
"""validate raises an error for unrecognised format strings.
311
It may raise either UnexpectedEndOfContainerError or
312
UnknownContainerFormatError, depending on exactly what the string is.
314
inputs = ["", "x", "Bazaar pack format 1 (introduced in 0.18)", "bad\n"]
316
reader = self.get_reader_for(input)
318
(errors.UnexpectedEndOfContainerError,
319
errors.UnknownContainerFormatError),
322
def test_validate_bad_record_marker(self):
323
"""validate raises UnknownRecordTypeError for unrecognised record
326
reader = self.get_reader_for(
327
"Bazaar pack format 1 (introduced in 0.18)\nX")
328
self.assertRaises(errors.UnknownRecordTypeError, reader.validate)
330
def test_validate_data_after_end_marker(self):
331
"""validate raises ContainerHasExcessDataError if there are any bytes
332
after the end of the container.
334
reader = self.get_reader_for(
335
"Bazaar pack format 1 (introduced in 0.18)\nEcrud")
337
errors.ContainerHasExcessDataError, reader.validate)
339
def test_validate_no_end_marker(self):
340
"""validate raises UnexpectedEndOfContainerError if there's no end of
341
container marker, even if the container up to this point has been valid.
343
reader = self.get_reader_for(
344
"Bazaar pack format 1 (introduced in 0.18)\n")
346
errors.UnexpectedEndOfContainerError, reader.validate)
348
def test_validate_duplicate_name(self):
349
"""validate raises DuplicateRecordNameError if the same name occurs
350
multiple times in the container.
352
reader = self.get_reader_for(
353
"Bazaar pack format 1 (introduced in 0.18)\n"
357
self.assertRaises(errors.DuplicateRecordNameError, reader.validate)
359
def test_validate_undecodeable_name(self):
360
"""Names that aren't valid UTF-8 cause validate to fail."""
361
reader = self.get_reader_for(
362
"Bazaar pack format 1 (introduced in 0.18)\nB0\n\xcc\n\nE")
363
self.assertRaises(errors.InvalidRecordError, reader.validate)
366
class TestBytesRecordReader(tests.TestCase):
367
"""Tests for reading and validating Bytes records with
370
Like TestContainerReader, this explicitly tests the reading of format 1
371
data. If a new version of the format is added, then a separate set of
372
tests for reading that format should be added.
375
def get_reader_for(self, bytes):
376
stream = StringIO(bytes)
377
reader = pack.BytesRecordReader(stream)
380
def test_record_with_no_name(self):
381
"""Reading a Bytes record with no name returns an empty list of
384
reader = self.get_reader_for("5\n\naaaaa")
385
names, get_bytes = reader.read()
386
self.assertEqual([], names)
387
self.assertEqual('aaaaa', get_bytes(None))
389
def test_record_with_one_name(self):
390
"""Reading a Bytes record with one name returns a list of just that
393
reader = self.get_reader_for("5\nname1\n\naaaaa")
394
names, get_bytes = reader.read()
395
self.assertEqual([('name1', )], names)
396
self.assertEqual('aaaaa', get_bytes(None))
398
def test_record_with_two_names(self):
399
"""Reading a Bytes record with two names returns a list of both names.
401
reader = self.get_reader_for("5\nname1\nname2\n\naaaaa")
402
names, get_bytes = reader.read()
403
self.assertEqual([('name1', ), ('name2', )], names)
404
self.assertEqual('aaaaa', get_bytes(None))
406
def test_record_with_two_part_names(self):
407
"""Reading a Bytes record with a two_part name reads both."""
408
reader = self.get_reader_for("5\nname1\x00name2\n\naaaaa")
409
names, get_bytes = reader.read()
410
self.assertEqual([('name1', 'name2', )], names)
411
self.assertEqual('aaaaa', get_bytes(None))
413
def test_invalid_length(self):
414
"""If the length-prefix is not a number, parsing raises
417
reader = self.get_reader_for("not a number\n")
418
self.assertRaises(errors.InvalidRecordError, reader.read)
420
def test_early_eof(self):
421
"""Tests for premature EOF occuring during parsing Bytes records with
424
A incomplete container might be interrupted at any point. The
425
BytesRecordReader needs to cope with the input stream running out no
426
matter where it is in the parsing process.
428
In all cases, UnexpectedEndOfContainerError should be raised.
430
complete_record = "6\nname\n\nabcdef"
431
for count in range(0, len(complete_record)):
432
incomplete_record = complete_record[:count]
433
reader = self.get_reader_for(incomplete_record)
434
# We don't use assertRaises to make diagnosing failures easier
435
# (assertRaises doesn't allow a custom failure message).
437
names, read_bytes = reader.read()
439
except errors.UnexpectedEndOfContainerError:
443
"UnexpectedEndOfContainerError not raised when parsing %r"
444
% (incomplete_record,))
446
def test_initial_eof(self):
447
"""EOF before any bytes read at all."""
448
reader = self.get_reader_for("")
449
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
451
def test_eof_after_length(self):
452
"""EOF after reading the length and before reading name(s)."""
453
reader = self.get_reader_for("123\n")
454
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
456
def test_eof_during_name(self):
457
"""EOF during reading a name."""
458
reader = self.get_reader_for("123\nname")
459
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
461
def test_read_invalid_name_whitespace(self):
462
"""Names must have no whitespace."""
463
# A name with a space.
464
reader = self.get_reader_for("0\nbad name\n\n")
465
self.assertRaises(errors.InvalidRecordError, reader.read)
468
reader = self.get_reader_for("0\nbad\tname\n\n")
469
self.assertRaises(errors.InvalidRecordError, reader.read)
471
# A name with a vertical tab.
472
reader = self.get_reader_for("0\nbad\vname\n\n")
473
self.assertRaises(errors.InvalidRecordError, reader.read)
475
def test_validate_whitespace_in_name(self):
476
"""Names must have no whitespace."""
477
reader = self.get_reader_for("0\nbad name\n\n")
478
self.assertRaises(errors.InvalidRecordError, reader.validate)
480
def test_validate_interrupted_prelude(self):
481
"""EOF during reading a record's prelude causes validate to fail."""
482
reader = self.get_reader_for("")
484
errors.UnexpectedEndOfContainerError, reader.validate)
486
def test_validate_interrupted_body(self):
487
"""EOF during reading a record's body causes validate to fail."""
488
reader = self.get_reader_for("1\n\n")
490
errors.UnexpectedEndOfContainerError, reader.validate)
492
def test_validate_unparseable_length(self):
493
"""An unparseable record length causes validate to fail."""
494
reader = self.get_reader_for("\n\n")
496
errors.InvalidRecordError, reader.validate)
498
def test_validate_undecodeable_name(self):
499
"""Names that aren't valid UTF-8 cause validate to fail."""
500
reader = self.get_reader_for("0\n\xcc\n\n")
501
self.assertRaises(errors.InvalidRecordError, reader.validate)
503
def test_read_max_length(self):
504
"""If the max_length passed to the callable returned by read is not
505
None, then no more than that many bytes will be read.
507
reader = self.get_reader_for("6\n\nabcdef")
508
names, get_bytes = reader.read()
509
self.assertEqual('abc', get_bytes(3))
511
def test_read_no_max_length(self):
512
"""If the max_length passed to the callable returned by read is None,
513
then all the bytes in the record will be read.
515
reader = self.get_reader_for("6\n\nabcdef")
516
names, get_bytes = reader.read()
517
self.assertEqual('abcdef', get_bytes(None))
519
def test_repeated_read_calls(self):
520
"""Repeated calls to the callable returned from BytesRecordReader.read
521
will not read beyond the end of the record.
523
reader = self.get_reader_for("6\n\nabcdefB3\nnext-record\nXXX")
524
names, get_bytes = reader.read()
525
self.assertEqual('abcdef', get_bytes(None))
526
self.assertEqual('', get_bytes(None))
527
self.assertEqual('', get_bytes(99))
530
class TestMakeReadvReader(tests.TestCaseWithTransport):
532
def test_read_skipping_records(self):
533
pack_data = StringIO()
534
writer = pack.ContainerWriter(pack_data.write)
537
memos.append(writer.add_bytes_record('abc', names=[]))
538
memos.append(writer.add_bytes_record('def', names=[('name1', )]))
539
memos.append(writer.add_bytes_record('ghi', names=[('name2', )]))
540
memos.append(writer.add_bytes_record('jkl', names=[]))
542
transport = self.get_transport()
543
transport.put_bytes('mypack', pack_data.getvalue())
544
requested_records = [memos[0], memos[2]]
545
reader = pack.make_readv_reader(transport, 'mypack', requested_records)
547
for names, reader_func in reader.iter_records():
548
result.append((names, reader_func(None)))
549
self.assertEqual([([], 'abc'), ([('name2', )], 'ghi')], result)
552
class TestReadvFile(tests.TestCaseWithTransport):
553
"""Tests of the ReadVFile class.
555
Error cases are deliberately undefined: this code adapts the underlying
556
transport interface to a single 'streaming read' interface as
557
ContainerReader needs.
560
def test_read_bytes(self):
561
"""Test reading of both single bytes and all bytes in a hunk."""
562
transport = self.get_transport()
563
transport.put_bytes('sample', '0123456789')
564
f = pack.ReadVFile(transport.readv('sample', [(0,1), (1,2), (4,1), (6,2)]))
566
results.append(f.read(1))
567
results.append(f.read(2))
568
results.append(f.read(1))
569
results.append(f.read(1))
570
results.append(f.read(1))
571
self.assertEqual(['0', '12', '4', '6', '7'], results)
573
def test_readline(self):
574
"""Test using readline() as ContainerReader does.
576
This is always within a readv hunk, never across it.
578
transport = self.get_transport()
579
transport.put_bytes('sample', '0\n2\n4\n')
580
f = pack.ReadVFile(transport.readv('sample', [(0,2), (2,4)]))
582
results.append(f.readline())
583
results.append(f.readline())
584
results.append(f.readline())
585
self.assertEqual(['0\n', '2\n', '4\n'], results)
587
def test_readline_and_read(self):
588
"""Test exercising one byte reads, readline, and then read again."""
589
transport = self.get_transport()
590
transport.put_bytes('sample', '0\n2\n4\n')
591
f = pack.ReadVFile(transport.readv('sample', [(0,6)]))
593
results.append(f.read(1))
594
results.append(f.readline())
595
results.append(f.read(4))
596
self.assertEqual(['0', '\n', '2\n4\n'], results)
599
class PushParserTestCase(tests.TestCase):
600
"""Base class for TestCases involving ContainerPushParser."""
602
def make_parser_expecting_record_type(self):
603
parser = pack.ContainerPushParser()
604
parser.accept_bytes("Bazaar pack format 1 (introduced in 0.18)\n")
607
def make_parser_expecting_bytes_record(self):
608
parser = pack.ContainerPushParser()
609
parser.accept_bytes("Bazaar pack format 1 (introduced in 0.18)\nB")
612
def assertRecordParsing(self, expected_record, bytes):
613
"""Assert that 'bytes' is parsed as a given bytes record.
615
:param expected_record: A tuple of (names, bytes).
617
parser = self.make_parser_expecting_bytes_record()
618
parser.accept_bytes(bytes)
619
parsed_records = parser.read_pending_records()
620
self.assertEqual([expected_record], parsed_records)
623
class TestContainerPushParser(PushParserTestCase):
624
"""Tests for ContainerPushParser.
626
The ContainerPushParser reads format 1 containers, so these tests
627
explicitly test how it reacts to format 1 data. If a new version of the
628
format is added, then separate tests for that format should be added.
631
def test_construct(self):
632
"""ContainerPushParser can be constructed."""
633
pack.ContainerPushParser()
635
def test_multiple_records_at_once(self):
636
"""If multiple records worth of data are fed to the parser in one
637
string, the parser will correctly parse all the records.
639
(A naive implementation might stop after parsing the first record.)
641
parser = self.make_parser_expecting_record_type()
642
parser.accept_bytes("B5\nname1\n\nbody1B5\nname2\n\nbody2")
644
[([('name1',)], 'body1'), ([('name2',)], 'body2')],
645
parser.read_pending_records())
647
def test_multiple_empty_records_at_once(self):
648
"""If multiple empty records worth of data are fed to the parser in one
649
string, the parser will correctly parse all the records.
651
(A naive implementation might stop after parsing the first empty
652
record, because the buffer size had not changed.)
654
parser = self.make_parser_expecting_record_type()
655
parser.accept_bytes("B0\nname1\n\nB0\nname2\n\n")
657
[([('name1',)], ''), ([('name2',)], '')],
658
parser.read_pending_records())
661
class TestContainerPushParserBytesParsing(PushParserTestCase):
662
"""Tests for reading Bytes records with ContainerPushParser.
664
The ContainerPushParser reads format 1 containers, so these tests
665
explicitly test how it reacts to format 1 data. If a new version of the
666
format is added, then separate tests for that format should be added.
669
def test_record_with_no_name(self):
670
"""Reading a Bytes record with no name returns an empty list of
673
self.assertRecordParsing(([], 'aaaaa'), "5\n\naaaaa")
675
def test_record_with_one_name(self):
676
"""Reading a Bytes record with one name returns a list of just that
679
self.assertRecordParsing(
680
([('name1', )], 'aaaaa'),
683
def test_record_with_two_names(self):
684
"""Reading a Bytes record with two names returns a list of both names.
686
self.assertRecordParsing(
687
([('name1', ), ('name2', )], 'aaaaa'),
688
"5\nname1\nname2\n\naaaaa")
690
def test_record_with_two_part_names(self):
691
"""Reading a Bytes record with a two_part name reads both."""
692
self.assertRecordParsing(
693
([('name1', 'name2')], 'aaaaa'),
694
"5\nname1\x00name2\n\naaaaa")
696
def test_invalid_length(self):
697
"""If the length-prefix is not a number, parsing raises
700
parser = self.make_parser_expecting_bytes_record()
702
errors.InvalidRecordError, parser.accept_bytes, "not a number\n")
704
def test_incomplete_record(self):
705
"""If the bytes seen so far don't form a complete record, then there
706
will be nothing returned by read_pending_records.
708
parser = self.make_parser_expecting_bytes_record()
709
parser.accept_bytes("5\n\nabcd")
710
self.assertEqual([], parser.read_pending_records())
712
def test_accept_nothing(self):
713
"""The edge case of parsing an empty string causes no error."""
714
parser = self.make_parser_expecting_bytes_record()
715
parser.accept_bytes("")
717
def assertInvalidRecord(self, bytes):
718
"""Assert that parsing the given bytes will raise an
721
parser = self.make_parser_expecting_bytes_record()
723
errors.InvalidRecordError, parser.accept_bytes, bytes)
725
def test_read_invalid_name_whitespace(self):
726
"""Names must have no whitespace."""
727
# A name with a space.
728
self.assertInvalidRecord("0\nbad name\n\n")
731
self.assertInvalidRecord("0\nbad\tname\n\n")
733
# A name with a vertical tab.
734
self.assertInvalidRecord("0\nbad\vname\n\n")
736
def test_repeated_read_pending_records(self):
737
"""read_pending_records will not return the same record twice."""
738
parser = self.make_parser_expecting_bytes_record()
739
parser.accept_bytes("6\n\nabcdef")
740
self.assertEqual([([], 'abcdef')], parser.read_pending_records())
741
self.assertEqual([], parser.read_pending_records())
added
removed
bzrlib/tests/test_pack.py
