← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_pack.py

  • Committer: John Arbash Meinel
  • Author(s): Mark Hammond
  • Date: 2008-09-09 17:02:21 UTC
  • mto: This revision was merged to the branch mainline in revision 3697.
  • Revision ID: john@arbash-meinel.com-20080909170221-svim3jw2mrz0amp3
An updated transparent icon for bzr.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_pack.py
1
 
# Copyright (C) 2007, 2009, 2011, 2012, 2016 Canonical Ltd
 
1
# Copyright (C) 2007 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
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
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Tests for bzrlib.pack."""
18
18
 
42
42
        serialiser = pack.ContainerSerialiser()
43
43
        record = serialiser.bytes_record('bytes', [])
44
44
        self.assertEqual('B5\n\nbytes', record)
45
 
 
 
45
        
46
46
    def test_bytes_record_one_name_with_one_part(self):
47
47
        serialiser = pack.ContainerSerialiser()
48
48
        record = serialiser.bytes_record('bytes', [('name',)])
49
49
        self.assertEqual('B5\nname\n\nbytes', record)
50
 
 
 
50
        
51
51
    def test_bytes_record_one_name_with_two_parts(self):
52
52
        serialiser = pack.ContainerSerialiser()
53
53
        record = serialiser.bytes_record('bytes', [('part1', 'part2')])
54
54
        self.assertEqual('B5\npart1\x00part2\n\nbytes', record)
55
 
 
 
55
        
56
56
    def test_bytes_record_two_names(self):
57
57
        serialiser = pack.ContainerSerialiser()
58
58
        record = serialiser.bytes_record('bytes', [('name1',), ('name2',)])
64
64
            errors.InvalidRecordError,
65
65
            serialiser.bytes_record, 'bytes', [('bad name',)])
66
66
 
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)
71
 
 
72
67
 
73
68
class TestContainerWriter(tests.TestCase):
74
69
 
75
70
    def setUp(self):
76
 
        super(TestContainerWriter, self).setUp()
77
71
        self.output = StringIO()
78
72
        self.writer = pack.ContainerWriter(self.output.write)
79
73
 
85
79
 
86
80
    def test_construct(self):
87
81
        """Test constructing a ContainerWriter.
88
 
 
 
82
        
89
83
        This uses None as the output stream to show that the constructor
90
84
        doesn't try to use the output stream.
91
85
        """
131
125
    def test_add_bytes_record_one_name(self):
132
126
        """Add a bytes record with one name."""
133
127
        self.writer.begin()
134
 
 
135
128
        offset, length = self.writer.add_bytes_record(
136
129
            'abc', names=[('name1', )])
137
130
        self.assertEqual((42, 13), (offset, length))
139
132
            'Bazaar pack format 1 (introduced in 0.18)\n'
140
133
            'B3\nname1\n\nabc')
141
134
 
142
 
    def test_add_bytes_record_split_writes(self):
143
 
        """Write a large record which does multiple IOs"""
144
 
 
145
 
        writes = []
146
 
        real_write = self.writer.write_func
147
 
 
148
 
        def record_writes(bytes):
149
 
            writes.append(bytes)
150
 
            return real_write(bytes)
151
 
 
152
 
        self.writer.write_func = record_writes
153
 
        self.writer._JOIN_WRITES_THRESHOLD = 2
154
 
 
155
 
        self.writer.begin()
156
 
        offset, length = self.writer.add_bytes_record(
157
 
            'abcabc', names=[('name1', )])
158
 
        self.assertEqual((42, 16), (offset, length))
159
 
        self.assertOutput(
160
 
            'Bazaar pack format 1 (introduced in 0.18)\n'
161
 
            'B6\nname1\n\nabcabc')
162
 
 
163
 
        self.assertEqual([
164
 
            'Bazaar pack format 1 (introduced in 0.18)\n',
165
 
            'B6\nname1\n\n',
166
 
            'abcabc'],
167
 
            writes)
168
 
 
169
135
    def test_add_bytes_record_two_names(self):
170
136
        """Add a bytes record with two names."""
171
137
        self.writer.begin()
239
205
 
240
206
    def test_construct(self):
241
207
        """Test constructing a ContainerReader.
242
 
 
243
 
        This uses None as the output stream to show that the constructor
244
 
        doesn't try to use the input stream.
 
208
        
 
209
        This uses None as the output stream to show that the constructor doesn't
 
210
        try to use the input stream.
245
211
        """
246
212
        reader = pack.ContainerReader(None)
247
213
 
277
243
 
278
244
    def test_container_with_one_unnamed_record(self):
279
245
        """Read a container with one Bytes record.
280
 
 
 
246
        
281
247
        Parsing Bytes records is more thoroughly exercised by
282
248
        TestBytesRecordReader.  This test is here to ensure that
283
249
        ContainerReader's integration with BytesRecordReader is working.
292
258
 
293
259
    def test_validate_empty_container(self):
294
260
        """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")
 
261
        reader = self.get_reader_for("Bazaar pack format 1 (introduced in 0.18)\nE")
297
262
        # No exception raised
298
263
        reader.validate()
299
264
 
361
326
        reader = self.get_reader_for(
362
327
            "Bazaar pack format 1 (introduced in 0.18)\nB0\n\xcc\n\nE")
363
328
        self.assertRaises(errors.InvalidRecordError, reader.validate)
364
 
 
 
329
        
365
330
 
366
331
class TestBytesRecordReader(tests.TestCase):
367
332
    """Tests for reading and validating Bytes records with
368
333
    BytesRecordReader.
369
 
 
 
334
    
370
335
    Like TestContainerReader, this explicitly tests the reading of format 1
371
336
    data.  If a new version of the format is added, then a separate set of
372
337
    tests for reading that format should be added.
420
385
    def test_early_eof(self):
421
386
        """Tests for premature EOF occuring during parsing Bytes records with
422
387
        BytesRecordReader.
423
 
 
 
388
        
424
389
        A incomplete container might be interrupted at any point.  The
425
390
        BytesRecordReader needs to cope with the input stream running out no
426
391
        matter where it is in the parsing process.
553
518
    """Tests of the ReadVFile class.
554
519
 
555
520
    Error cases are deliberately undefined: this code adapts the underlying
556
 
    transport interface to a single 'streaming read' interface as
 
521
    transport interface to a single 'streaming read' interface as 
557
522
    ContainerReader needs.
558
523
    """
559
524
 
619
584
        parsed_records = parser.read_pending_records()
620
585
        self.assertEqual([expected_record], parsed_records)
621
586
 
622
 
 
 
587
        
623
588
class TestContainerPushParser(PushParserTestCase):
624
589
    """Tests for ContainerPushParser.
625
 
 
 
590
    
626
591
    The ContainerPushParser reads format 1 containers, so these tests
627
592
    explicitly test how it reacts to format 1 data.  If a new version of the
628
593
    format is added, then separate tests for that format should be added.
644
609
            [([('name1',)], 'body1'), ([('name2',)], 'body2')],
645
610
            parser.read_pending_records())
646
611
 
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.
650
 
 
651
 
        (A naive implementation might stop after parsing the first empty
652
 
        record, because the buffer size had not changed.)
653
 
        """
654
 
        parser = self.make_parser_expecting_record_type()
655
 
        parser.accept_bytes("B0\nname1\n\nB0\nname2\n\n")
656
 
        self.assertEqual(
657
 
            [([('name1',)], ''), ([('name2',)], '')],
658
 
            parser.read_pending_records())
659
 
 
660
612
 
661
613
class TestContainerPushParserBytesParsing(PushParserTestCase):
662
614
    """Tests for reading Bytes records with ContainerPushParser.
663
 
 
 
615
    
664
616
    The ContainerPushParser reads format 1 containers, so these tests
665
617
    explicitly test how it reacts to format 1 data.  If a new version of the
666
618
    format is added, then separate tests for that format should be added.
739
691
        parser.accept_bytes("6\n\nabcdef")
740
692
        self.assertEqual([([], 'abcdef')], parser.read_pending_records())
741
693
        self.assertEqual([], parser.read_pending_records())
 
694
 
 
695