1
# Copyright (C) 2007, 2009, 2011, 2012, 2016 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 breezy.pack."""
19
from .. import errors, tests
23
from ..sixish import (
28
class TestContainerSerialiser(tests.TestCase):
29
"""Tests for the ContainerSerialiser class."""
31
def test_construct(self):
32
"""Test constructing a ContainerSerialiser."""
33
pack.ContainerSerialiser()
36
serialiser = pack.ContainerSerialiser()
37
self.assertEqual('Bazaar pack format 1 (introduced in 0.18)\n',
41
serialiser = pack.ContainerSerialiser()
42
self.assertEqual('E', serialiser.end())
44
def test_bytes_record_no_name(self):
45
serialiser = pack.ContainerSerialiser()
46
record = serialiser.bytes_record('bytes', [])
47
self.assertEqual('B5\n\nbytes', record)
49
def test_bytes_record_one_name_with_one_part(self):
50
serialiser = pack.ContainerSerialiser()
51
record = serialiser.bytes_record('bytes', [('name',)])
52
self.assertEqual('B5\nname\n\nbytes', record)
54
def test_bytes_record_one_name_with_two_parts(self):
55
serialiser = pack.ContainerSerialiser()
56
record = serialiser.bytes_record('bytes', [('part1', 'part2')])
57
self.assertEqual('B5\npart1\x00part2\n\nbytes', record)
59
def test_bytes_record_two_names(self):
60
serialiser = pack.ContainerSerialiser()
61
record = serialiser.bytes_record('bytes', [('name1',), ('name2',)])
62
self.assertEqual('B5\nname1\nname2\n\nbytes', record)
64
def test_bytes_record_whitespace_in_name_part(self):
65
serialiser = pack.ContainerSerialiser()
67
errors.InvalidRecordError,
68
serialiser.bytes_record, 'bytes', [('bad name',)])
70
def test_bytes_record_header(self):
71
serialiser = pack.ContainerSerialiser()
72
record = serialiser.bytes_header(32, [('name1',), ('name2',)])
73
self.assertEqual('B32\nname1\nname2\n\n', record)
76
class TestContainerWriter(tests.TestCase):
79
super(TestContainerWriter, self).setUp()
80
self.output = BytesIO()
81
self.writer = pack.ContainerWriter(self.output.write)
83
def assertOutput(self, expected_output):
84
"""Assert that the output of self.writer ContainerWriter is equal to
87
self.assertEqual(expected_output, self.output.getvalue())
89
def test_construct(self):
90
"""Test constructing a ContainerWriter.
92
This uses None as the output stream to show that the constructor
93
doesn't try to use the output stream.
95
writer = pack.ContainerWriter(None)
98
"""The begin() method writes the container format marker line."""
100
self.assertOutput('Bazaar pack format 1 (introduced in 0.18)\n')
102
def test_zero_records_written_after_begin(self):
103
"""After begin is written, 0 records have been written."""
105
self.assertEqual(0, self.writer.records_written)
108
"""The end() method writes an End Marker record."""
111
self.assertOutput('Bazaar pack format 1 (introduced in 0.18)\nE')
113
def test_empty_end_does_not_add_a_record_to_records_written(self):
114
"""The end() method does not count towards the records written."""
117
self.assertEqual(0, self.writer.records_written)
119
def test_non_empty_end_does_not_add_a_record_to_records_written(self):
120
"""The end() method does not count towards the records written."""
122
self.writer.add_bytes_record('foo', names=[])
124
self.assertEqual(1, self.writer.records_written)
126
def test_add_bytes_record_no_name(self):
127
"""Add a bytes record with no name."""
129
offset, length = self.writer.add_bytes_record('abc', names=[])
130
self.assertEqual((42, 7), (offset, length))
132
'Bazaar pack format 1 (introduced in 0.18)\nB3\n\nabc')
134
def test_add_bytes_record_one_name(self):
135
"""Add a bytes record with one name."""
138
offset, length = self.writer.add_bytes_record(
139
'abc', names=[('name1', )])
140
self.assertEqual((42, 13), (offset, length))
142
'Bazaar pack format 1 (introduced in 0.18)\n'
145
def test_add_bytes_record_split_writes(self):
146
"""Write a large record which does multiple IOs"""
149
real_write = self.writer.write_func
151
def record_writes(bytes):
153
return real_write(bytes)
155
self.writer.write_func = record_writes
156
self.writer._JOIN_WRITES_THRESHOLD = 2
159
offset, length = self.writer.add_bytes_record(
160
'abcabc', names=[('name1', )])
161
self.assertEqual((42, 16), (offset, length))
163
'Bazaar pack format 1 (introduced in 0.18)\n'
164
'B6\nname1\n\nabcabc')
167
'Bazaar pack format 1 (introduced in 0.18)\n',
172
def test_add_bytes_record_two_names(self):
173
"""Add a bytes record with two names."""
175
offset, length = self.writer.add_bytes_record(
176
'abc', names=[('name1', ), ('name2', )])
177
self.assertEqual((42, 19), (offset, length))
179
'Bazaar pack format 1 (introduced in 0.18)\n'
180
'B3\nname1\nname2\n\nabc')
182
def test_add_bytes_record_two_names(self):
183
"""Add a bytes record with two names."""
185
offset, length = self.writer.add_bytes_record(
186
'abc', names=[('name1', ), ('name2', )])
187
self.assertEqual((42, 19), (offset, length))
189
'Bazaar pack format 1 (introduced in 0.18)\n'
190
'B3\nname1\nname2\n\nabc')
192
def test_add_bytes_record_two_element_name(self):
193
"""Add a bytes record with a two-element name."""
195
offset, length = self.writer.add_bytes_record(
196
'abc', names=[('name1', 'name2')])
197
self.assertEqual((42, 19), (offset, length))
199
'Bazaar pack format 1 (introduced in 0.18)\n'
200
'B3\nname1\x00name2\n\nabc')
202
def test_add_second_bytes_record_gets_higher_offset(self):
204
self.writer.add_bytes_record('abc', names=[])
205
offset, length = self.writer.add_bytes_record('abc', names=[])
206
self.assertEqual((49, 7), (offset, length))
208
'Bazaar pack format 1 (introduced in 0.18)\n'
212
def test_add_bytes_record_invalid_name(self):
213
"""Adding a Bytes record with a name with whitespace in it raises
218
errors.InvalidRecordError,
219
self.writer.add_bytes_record, 'abc', names=[('bad name', )])
221
def test_add_bytes_records_add_to_records_written(self):
222
"""Adding a Bytes record increments the records_written counter."""
224
self.writer.add_bytes_record('foo', names=[])
225
self.assertEqual(1, self.writer.records_written)
226
self.writer.add_bytes_record('foo', names=[])
227
self.assertEqual(2, self.writer.records_written)
230
class TestContainerReader(tests.TestCase):
231
"""Tests for the ContainerReader.
233
The ContainerReader reads format 1 containers, so these tests explicitly
234
test how it reacts to format 1 data. If a new version of the format is
235
added, then separate tests for that format should be added.
238
def get_reader_for(self, bytes):
239
stream = BytesIO(bytes)
240
reader = pack.ContainerReader(stream)
243
def test_construct(self):
244
"""Test constructing a ContainerReader.
246
This uses None as the output stream to show that the constructor
247
doesn't try to use the input stream.
249
reader = pack.ContainerReader(None)
251
def test_empty_container(self):
252
"""Read an empty container."""
253
reader = self.get_reader_for(
254
"Bazaar pack format 1 (introduced in 0.18)\nE")
255
self.assertEqual([], list(reader.iter_records()))
257
def test_unknown_format(self):
258
"""Unrecognised container formats raise UnknownContainerFormatError."""
259
reader = self.get_reader_for("unknown format\n")
261
errors.UnknownContainerFormatError, reader.iter_records)
263
def test_unexpected_end_of_container(self):
264
"""Containers that don't end with an End Marker record should cause
265
UnexpectedEndOfContainerError to be raised.
267
reader = self.get_reader_for(
268
"Bazaar pack format 1 (introduced in 0.18)\n")
269
iterator = reader.iter_records()
271
errors.UnexpectedEndOfContainerError, next, iterator)
273
def test_unknown_record_type(self):
274
"""Unknown record types cause UnknownRecordTypeError to be raised."""
275
reader = self.get_reader_for(
276
"Bazaar pack format 1 (introduced in 0.18)\nX")
277
iterator = reader.iter_records()
279
errors.UnknownRecordTypeError, next, iterator)
281
def test_container_with_one_unnamed_record(self):
282
"""Read a container with one Bytes record.
284
Parsing Bytes records is more thoroughly exercised by
285
TestBytesRecordReader. This test is here to ensure that
286
ContainerReader's integration with BytesRecordReader is working.
288
reader = self.get_reader_for(
289
"Bazaar pack format 1 (introduced in 0.18)\nB5\n\naaaaaE")
290
expected_records = [([], 'aaaaa')]
293
[(names, read_bytes(None))
294
for (names, read_bytes) in reader.iter_records()])
296
def test_validate_empty_container(self):
297
"""validate does not raise an error for a container with no records."""
298
reader = self.get_reader_for(
299
"Bazaar pack format 1 (introduced in 0.18)\nE")
300
# No exception raised
303
def test_validate_non_empty_valid_container(self):
304
"""validate does not raise an error for a container with a valid record.
306
reader = self.get_reader_for(
307
"Bazaar pack format 1 (introduced in 0.18)\nB3\nname\n\nabcE")
308
# No exception raised
311
def test_validate_bad_format(self):
312
"""validate raises an error for unrecognised format strings.
314
It may raise either UnexpectedEndOfContainerError or
315
UnknownContainerFormatError, depending on exactly what the string is.
317
inputs = ["", "x", "Bazaar pack format 1 (introduced in 0.18)", "bad\n"]
319
reader = self.get_reader_for(input)
321
(errors.UnexpectedEndOfContainerError,
322
errors.UnknownContainerFormatError),
325
def test_validate_bad_record_marker(self):
326
"""validate raises UnknownRecordTypeError for unrecognised record
329
reader = self.get_reader_for(
330
"Bazaar pack format 1 (introduced in 0.18)\nX")
331
self.assertRaises(errors.UnknownRecordTypeError, reader.validate)
333
def test_validate_data_after_end_marker(self):
334
"""validate raises ContainerHasExcessDataError if there are any bytes
335
after the end of the container.
337
reader = self.get_reader_for(
338
"Bazaar pack format 1 (introduced in 0.18)\nEcrud")
340
errors.ContainerHasExcessDataError, reader.validate)
342
def test_validate_no_end_marker(self):
343
"""validate raises UnexpectedEndOfContainerError if there's no end of
344
container marker, even if the container up to this point has been valid.
346
reader = self.get_reader_for(
347
"Bazaar pack format 1 (introduced in 0.18)\n")
349
errors.UnexpectedEndOfContainerError, reader.validate)
351
def test_validate_duplicate_name(self):
352
"""validate raises DuplicateRecordNameError if the same name occurs
353
multiple times in the container.
355
reader = self.get_reader_for(
356
"Bazaar pack format 1 (introduced in 0.18)\n"
360
self.assertRaises(errors.DuplicateRecordNameError, 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(
365
"Bazaar pack format 1 (introduced in 0.18)\nB0\n\xcc\n\nE")
366
self.assertRaises(errors.InvalidRecordError, reader.validate)
369
class TestBytesRecordReader(tests.TestCase):
370
"""Tests for reading and validating Bytes records with
373
Like TestContainerReader, this explicitly tests the reading of format 1
374
data. If a new version of the format is added, then a separate set of
375
tests for reading that format should be added.
378
def get_reader_for(self, bytes):
379
stream = BytesIO(bytes)
380
reader = pack.BytesRecordReader(stream)
383
def test_record_with_no_name(self):
384
"""Reading a Bytes record with no name returns an empty list of
387
reader = self.get_reader_for("5\n\naaaaa")
388
names, get_bytes = reader.read()
389
self.assertEqual([], names)
390
self.assertEqual('aaaaa', get_bytes(None))
392
def test_record_with_one_name(self):
393
"""Reading a Bytes record with one name returns a list of just that
396
reader = self.get_reader_for("5\nname1\n\naaaaa")
397
names, get_bytes = reader.read()
398
self.assertEqual([('name1', )], names)
399
self.assertEqual('aaaaa', get_bytes(None))
401
def test_record_with_two_names(self):
402
"""Reading a Bytes record with two names returns a list of both names.
404
reader = self.get_reader_for("5\nname1\nname2\n\naaaaa")
405
names, get_bytes = reader.read()
406
self.assertEqual([('name1', ), ('name2', )], names)
407
self.assertEqual('aaaaa', get_bytes(None))
409
def test_record_with_two_part_names(self):
410
"""Reading a Bytes record with a two_part name reads both."""
411
reader = self.get_reader_for("5\nname1\x00name2\n\naaaaa")
412
names, get_bytes = reader.read()
413
self.assertEqual([('name1', 'name2', )], names)
414
self.assertEqual('aaaaa', get_bytes(None))
416
def test_invalid_length(self):
417
"""If the length-prefix is not a number, parsing raises
420
reader = self.get_reader_for("not a number\n")
421
self.assertRaises(errors.InvalidRecordError, reader.read)
423
def test_early_eof(self):
424
"""Tests for premature EOF occuring during parsing Bytes records with
427
A incomplete container might be interrupted at any point. The
428
BytesRecordReader needs to cope with the input stream running out no
429
matter where it is in the parsing process.
431
In all cases, UnexpectedEndOfContainerError should be raised.
433
complete_record = "6\nname\n\nabcdef"
434
for count in range(0, len(complete_record)):
435
incomplete_record = complete_record[:count]
436
reader = self.get_reader_for(incomplete_record)
437
# We don't use assertRaises to make diagnosing failures easier
438
# (assertRaises doesn't allow a custom failure message).
440
names, read_bytes = reader.read()
442
except errors.UnexpectedEndOfContainerError:
446
"UnexpectedEndOfContainerError not raised when parsing %r"
447
% (incomplete_record,))
449
def test_initial_eof(self):
450
"""EOF before any bytes read at all."""
451
reader = self.get_reader_for("")
452
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
454
def test_eof_after_length(self):
455
"""EOF after reading the length and before reading name(s)."""
456
reader = self.get_reader_for("123\n")
457
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
459
def test_eof_during_name(self):
460
"""EOF during reading a name."""
461
reader = self.get_reader_for("123\nname")
462
self.assertRaises(errors.UnexpectedEndOfContainerError, reader.read)
464
def test_read_invalid_name_whitespace(self):
465
"""Names must have no whitespace."""
466
# A name with a space.
467
reader = self.get_reader_for("0\nbad name\n\n")
468
self.assertRaises(errors.InvalidRecordError, reader.read)
471
reader = self.get_reader_for("0\nbad\tname\n\n")
472
self.assertRaises(errors.InvalidRecordError, reader.read)
474
# A name with a vertical tab.
475
reader = self.get_reader_for("0\nbad\vname\n\n")
476
self.assertRaises(errors.InvalidRecordError, reader.read)
478
def test_validate_whitespace_in_name(self):
479
"""Names must have no whitespace."""
480
reader = self.get_reader_for("0\nbad name\n\n")
481
self.assertRaises(errors.InvalidRecordError, reader.validate)
483
def test_validate_interrupted_prelude(self):
484
"""EOF during reading a record's prelude causes validate to fail."""
485
reader = self.get_reader_for("")
487
errors.UnexpectedEndOfContainerError, reader.validate)
489
def test_validate_interrupted_body(self):
490
"""EOF during reading a record's body causes validate to fail."""
491
reader = self.get_reader_for("1\n\n")
493
errors.UnexpectedEndOfContainerError, reader.validate)
495
def test_validate_unparseable_length(self):
496
"""An unparseable record length causes validate to fail."""
497
reader = self.get_reader_for("\n\n")
499
errors.InvalidRecordError, reader.validate)
501
def test_validate_undecodeable_name(self):
502
"""Names that aren't valid UTF-8 cause validate to fail."""
503
reader = self.get_reader_for("0\n\xcc\n\n")
504
self.assertRaises(errors.InvalidRecordError, reader.validate)
506
def test_read_max_length(self):
507
"""If the max_length passed to the callable returned by read is not
508
None, then no more than that many bytes will be read.
510
reader = self.get_reader_for("6\n\nabcdef")
511
names, get_bytes = reader.read()
512
self.assertEqual('abc', get_bytes(3))
514
def test_read_no_max_length(self):
515
"""If the max_length passed to the callable returned by read is None,
516
then all the bytes in the record will be read.
518
reader = self.get_reader_for("6\n\nabcdef")
519
names, get_bytes = reader.read()
520
self.assertEqual('abcdef', get_bytes(None))
522
def test_repeated_read_calls(self):
523
"""Repeated calls to the callable returned from BytesRecordReader.read
524
will not read beyond the end of the record.
526
reader = self.get_reader_for("6\n\nabcdefB3\nnext-record\nXXX")
527
names, get_bytes = reader.read()
528
self.assertEqual('abcdef', get_bytes(None))
529
self.assertEqual('', get_bytes(None))
530
self.assertEqual('', get_bytes(99))
533
class TestMakeReadvReader(tests.TestCaseWithTransport):
535
def test_read_skipping_records(self):
536
pack_data = BytesIO()
537
writer = pack.ContainerWriter(pack_data.write)
540
memos.append(writer.add_bytes_record('abc', names=[]))
541
memos.append(writer.add_bytes_record('def', names=[('name1', )]))
542
memos.append(writer.add_bytes_record('ghi', names=[('name2', )]))
543
memos.append(writer.add_bytes_record('jkl', names=[]))
545
transport = self.get_transport()
546
transport.put_bytes('mypack', pack_data.getvalue())
547
requested_records = [memos[0], memos[2]]
548
reader = pack.make_readv_reader(transport, 'mypack', requested_records)
550
for names, reader_func in reader.iter_records():
551
result.append((names, reader_func(None)))
552
self.assertEqual([([], 'abc'), ([('name2', )], 'ghi')], result)
555
class TestReadvFile(tests.TestCaseWithTransport):
556
"""Tests of the ReadVFile class.
558
Error cases are deliberately undefined: this code adapts the underlying
559
transport interface to a single 'streaming read' interface as
560
ContainerReader needs.
563
def test_read_bytes(self):
564
"""Test reading of both single bytes and all bytes in a hunk."""
565
transport = self.get_transport()
566
transport.put_bytes('sample', '0123456789')
567
f = pack.ReadVFile(transport.readv('sample', [(0,1), (1,2), (4,1), (6,2)]))
569
results.append(f.read(1))
570
results.append(f.read(2))
571
results.append(f.read(1))
572
results.append(f.read(1))
573
results.append(f.read(1))
574
self.assertEqual(['0', '12', '4', '6', '7'], results)
576
def test_readline(self):
577
"""Test using readline() as ContainerReader does.
579
This is always within a readv hunk, never across it.
581
transport = self.get_transport()
582
transport.put_bytes('sample', '0\n2\n4\n')
583
f = pack.ReadVFile(transport.readv('sample', [(0,2), (2,4)]))
585
results.append(f.readline())
586
results.append(f.readline())
587
results.append(f.readline())
588
self.assertEqual(['0\n', '2\n', '4\n'], results)
590
def test_readline_and_read(self):
591
"""Test exercising one byte reads, readline, and then read again."""
592
transport = self.get_transport()
593
transport.put_bytes('sample', '0\n2\n4\n')
594
f = pack.ReadVFile(transport.readv('sample', [(0,6)]))
596
results.append(f.read(1))
597
results.append(f.readline())
598
results.append(f.read(4))
599
self.assertEqual(['0', '\n', '2\n4\n'], results)
602
class PushParserTestCase(tests.TestCase):
603
"""Base class for TestCases involving ContainerPushParser."""
605
def make_parser_expecting_record_type(self):
606
parser = pack.ContainerPushParser()
607
parser.accept_bytes("Bazaar pack format 1 (introduced in 0.18)\n")
610
def make_parser_expecting_bytes_record(self):
611
parser = pack.ContainerPushParser()
612
parser.accept_bytes("Bazaar pack format 1 (introduced in 0.18)\nB")
615
def assertRecordParsing(self, expected_record, bytes):
616
"""Assert that 'bytes' is parsed as a given bytes record.
618
:param expected_record: A tuple of (names, bytes).
620
parser = self.make_parser_expecting_bytes_record()
621
parser.accept_bytes(bytes)
622
parsed_records = parser.read_pending_records()
623
self.assertEqual([expected_record], parsed_records)
626
class TestContainerPushParser(PushParserTestCase):
627
"""Tests for ContainerPushParser.
629
The ContainerPushParser reads format 1 containers, so these tests
630
explicitly test how it reacts to format 1 data. If a new version of the
631
format is added, then separate tests for that format should be added.
634
def test_construct(self):
635
"""ContainerPushParser can be constructed."""
636
pack.ContainerPushParser()
638
def test_multiple_records_at_once(self):
639
"""If multiple records worth of data are fed to the parser in one
640
string, the parser will correctly parse all the records.
642
(A naive implementation might stop after parsing the first record.)
644
parser = self.make_parser_expecting_record_type()
645
parser.accept_bytes("B5\nname1\n\nbody1B5\nname2\n\nbody2")
647
[([('name1',)], 'body1'), ([('name2',)], 'body2')],
648
parser.read_pending_records())
650
def test_multiple_empty_records_at_once(self):
651
"""If multiple empty records worth of data are fed to the parser in one
652
string, the parser will correctly parse all the records.
654
(A naive implementation might stop after parsing the first empty
655
record, because the buffer size had not changed.)
657
parser = self.make_parser_expecting_record_type()
658
parser.accept_bytes("B0\nname1\n\nB0\nname2\n\n")
660
[([('name1',)], ''), ([('name2',)], '')],
661
parser.read_pending_records())
664
class TestContainerPushParserBytesParsing(PushParserTestCase):
665
"""Tests for reading Bytes records with ContainerPushParser.
667
The ContainerPushParser reads format 1 containers, so these tests
668
explicitly test how it reacts to format 1 data. If a new version of the
669
format is added, then separate tests for that format should be added.
672
def test_record_with_no_name(self):
673
"""Reading a Bytes record with no name returns an empty list of
676
self.assertRecordParsing(([], 'aaaaa'), "5\n\naaaaa")
678
def test_record_with_one_name(self):
679
"""Reading a Bytes record with one name returns a list of just that
682
self.assertRecordParsing(
683
([('name1', )], 'aaaaa'),
686
def test_record_with_two_names(self):
687
"""Reading a Bytes record with two names returns a list of both names.
689
self.assertRecordParsing(
690
([('name1', ), ('name2', )], 'aaaaa'),
691
"5\nname1\nname2\n\naaaaa")
693
def test_record_with_two_part_names(self):
694
"""Reading a Bytes record with a two_part name reads both."""
695
self.assertRecordParsing(
696
([('name1', 'name2')], 'aaaaa'),
697
"5\nname1\x00name2\n\naaaaa")
699
def test_invalid_length(self):
700
"""If the length-prefix is not a number, parsing raises
703
parser = self.make_parser_expecting_bytes_record()
705
errors.InvalidRecordError, parser.accept_bytes, "not a number\n")
707
def test_incomplete_record(self):
708
"""If the bytes seen so far don't form a complete record, then there
709
will be nothing returned by read_pending_records.
711
parser = self.make_parser_expecting_bytes_record()
712
parser.accept_bytes("5\n\nabcd")
713
self.assertEqual([], parser.read_pending_records())
715
def test_accept_nothing(self):
716
"""The edge case of parsing an empty string causes no error."""
717
parser = self.make_parser_expecting_bytes_record()
718
parser.accept_bytes("")
720
def assertInvalidRecord(self, bytes):
721
"""Assert that parsing the given bytes will raise an
724
parser = self.make_parser_expecting_bytes_record()
726
errors.InvalidRecordError, parser.accept_bytes, bytes)
728
def test_read_invalid_name_whitespace(self):
729
"""Names must have no whitespace."""
730
# A name with a space.
731
self.assertInvalidRecord("0\nbad name\n\n")
734
self.assertInvalidRecord("0\nbad\tname\n\n")
736
# A name with a vertical tab.
737
self.assertInvalidRecord("0\nbad\vname\n\n")
739
def test_repeated_read_pending_records(self):
740
"""read_pending_records will not return the same record twice."""
741
parser = self.make_parser_expecting_bytes_record()
742
parser.accept_bytes("6\n\nabcdef")
743
self.assertEqual([([], 'abcdef')], parser.read_pending_records())
744
self.assertEqual([], parser.read_pending_records())
added
removed
breezy/tests/test_pack.py
