/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to breezy/tests/test_pack.py

  • Committer: Jelmer Vernooij
  • Date: 2020-05-06 02:13:25 UTC
  • mfrom: (7490.7.21 work)
  • mto: This revision was merged to the branch mainline in revision 7501.
  • Revision ID: jelmer@jelmer.uk-20200506021325-awbmmqu1zyorz7sj
Merge 3.1 branch.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_pack.py
16
16
 
17
17
"""Tests for breezy.pack."""
18
18
 
 
19
from io import BytesIO
 
20
 
19
21
from .. import errors, tests
20
22
from ..bzr import (
21
23
    pack,
22
24
    )
23
 
from ..sixish import (
24
 
    BytesIO,
25
 
    )
26
25
 
27
26
 
28
27
class TestContainerSerialiser(tests.TestCase):
92
91
        This uses None as the output stream to show that the constructor
93
92
        doesn't try to use the output stream.
94
93
        """
95
 
        writer = pack.ContainerWriter(None)
 
94
        pack.ContainerWriter(None)
96
95
 
97
96
    def test_begin(self):
98
97
        """The begin() method writes the container format marker line."""
119
118
    def test_non_empty_end_does_not_add_a_record_to_records_written(self):
120
119
        """The end() method does not count towards the records written."""
121
120
        self.writer.begin()
122
 
        self.writer.add_bytes_record(b'foo', names=[])
 
121
        self.writer.add_bytes_record([b'foo'], len(b'foo'), names=[])
123
122
        self.writer.end()
124
123
        self.assertEqual(1, self.writer.records_written)
125
124
 
126
125
    def test_add_bytes_record_no_name(self):
127
126
        """Add a bytes record with no name."""
128
127
        self.writer.begin()
129
 
        offset, length = self.writer.add_bytes_record(b'abc', names=[])
 
128
        offset, length = self.writer.add_bytes_record([b'abc'], len(b'abc'), names=[])
130
129
        self.assertEqual((42, 7), (offset, length))
131
130
        self.assertOutput(
132
131
            b'Bazaar pack format 1 (introduced in 0.18)\nB3\n\nabc')
136
135
        self.writer.begin()
137
136
 
138
137
        offset, length = self.writer.add_bytes_record(
139
 
            b'abc', names=[(b'name1', )])
 
138
            [b'abc'], len(b'abc'), names=[(b'name1', )])
140
139
        self.assertEqual((42, 13), (offset, length))
141
140
        self.assertOutput(
142
141
            b'Bazaar pack format 1 (introduced in 0.18)\n'
157
156
 
158
157
        self.writer.begin()
159
158
        offset, length = self.writer.add_bytes_record(
160
 
            b'abcabc', names=[(b'name1', )])
 
159
            [b'abcabc'], len(b'abcabc'), names=[(b'name1', )])
161
160
        self.assertEqual((42, 16), (offset, length))
162
161
        self.assertOutput(
163
162
            b'Bazaar pack format 1 (introduced in 0.18)\n'
173
172
        """Add a bytes record with two names."""
174
173
        self.writer.begin()
175
174
        offset, length = self.writer.add_bytes_record(
176
 
            b'abc', names=[(b'name1', ), (b'name2', )])
 
175
            [b'abc'], len(b'abc'), names=[(b'name1', ), (b'name2', )])
177
176
        self.assertEqual((42, 19), (offset, length))
178
177
        self.assertOutput(
179
178
            b'Bazaar pack format 1 (introduced in 0.18)\n'
183
182
        """Add a bytes record with two names."""
184
183
        self.writer.begin()
185
184
        offset, length = self.writer.add_bytes_record(
186
 
            b'abc', names=[(b'name1', ), (b'name2', )])
 
185
            [b'abc'], len(b'abc'), names=[(b'name1', ), (b'name2', )])
187
186
        self.assertEqual((42, 19), (offset, length))
188
187
        self.assertOutput(
189
188
            b'Bazaar pack format 1 (introduced in 0.18)\n'
193
192
        """Add a bytes record with a two-element name."""
194
193
        self.writer.begin()
195
194
        offset, length = self.writer.add_bytes_record(
196
 
            b'abc', names=[(b'name1', b'name2')])
 
195
            [b'abc'], len(b'abc'), names=[(b'name1', b'name2')])
197
196
        self.assertEqual((42, 19), (offset, length))
198
197
        self.assertOutput(
199
198
            b'Bazaar pack format 1 (introduced in 0.18)\n'
201
200
 
202
201
    def test_add_second_bytes_record_gets_higher_offset(self):
203
202
        self.writer.begin()
204
 
        self.writer.add_bytes_record(b'abc', names=[])
205
 
        offset, length = self.writer.add_bytes_record(b'abc', names=[])
 
203
        self.writer.add_bytes_record([b'a', b'bc'], len(b'abc'), names=[])
 
204
        offset, length = self.writer.add_bytes_record([b'abc'], len(b'abc'), names=[])
206
205
        self.assertEqual((49, 7), (offset, length))
207
206
        self.assertOutput(
208
207
            b'Bazaar pack format 1 (introduced in 0.18)\n'
216
215
        self.writer.begin()
217
216
        self.assertRaises(
218
217
            errors.InvalidRecordError,
219
 
            self.writer.add_bytes_record, b'abc', names=[(b'bad name', )])
 
218
            self.writer.add_bytes_record, [b'abc'], len(b'abc'), names=[(b'bad name', )])
220
219
 
221
220
    def test_add_bytes_records_add_to_records_written(self):
222
221
        """Adding a Bytes record increments the records_written counter."""
223
222
        self.writer.begin()
224
 
        self.writer.add_bytes_record(b'foo', names=[])
 
223
        self.writer.add_bytes_record([b'foo'], len(b'foo'), names=[])
225
224
        self.assertEqual(1, self.writer.records_written)
226
 
        self.writer.add_bytes_record(b'foo', names=[])
 
225
        self.writer.add_bytes_record([b'foo'], len(b'foo'), names=[])
227
226
        self.assertEqual(2, self.writer.records_written)
228
227
 
229
228
 
246
245
        This uses None as the output stream to show that the constructor
247
246
        doesn't try to use the input stream.
248
247
        """
249
 
        reader = pack.ContainerReader(None)
 
248
        pack.ContainerReader(None)
250
249
 
251
250
    def test_empty_container(self):
252
251
        """Read an empty container."""
342
341
 
343
342
    def test_validate_no_end_marker(self):
344
343
        """validate raises UnexpectedEndOfContainerError if there's no end of
345
 
        container marker, even if the container up to this point has been valid.
 
344
        container marker, even if the container up to this point has been
 
345
        valid.
346
346
        """
347
347
        reader = self.get_reader_for(
348
348
            b"Bazaar pack format 1 (introduced in 0.18)\n")
538
538
        writer = pack.ContainerWriter(pack_data.write)
539
539
        writer.begin()
540
540
        memos = []
541
 
        memos.append(writer.add_bytes_record(b'abc', names=[]))
542
 
        memos.append(writer.add_bytes_record(b'def', names=[(b'name1', )]))
543
 
        memos.append(writer.add_bytes_record(b'ghi', names=[(b'name2', )]))
544
 
        memos.append(writer.add_bytes_record(b'jkl', names=[]))
 
541
        memos.append(writer.add_bytes_record([b'abc'], 3, names=[]))
 
542
        memos.append(writer.add_bytes_record([b'def'], 3, names=[(b'name1', )]))
 
543
        memos.append(writer.add_bytes_record([b'ghi'], 3, names=[(b'name2', )]))
 
544
        memos.append(writer.add_bytes_record([b'jkl'], 3, names=[]))
545
545
        writer.end()
546
546
        transport = self.get_transport()
547
 
        transport.put_bytes(b'mypack', pack_data.getvalue())
 
547
        transport.put_bytes('mypack', pack_data.getvalue())
548
548
        requested_records = [memos[0], memos[2]]
549
 
        reader = pack.make_readv_reader(transport, b'mypack', requested_records)
 
549
        reader = pack.make_readv_reader(transport, 'mypack', requested_records)
550
550
        result = []
551
551
        for names, reader_func in reader.iter_records():
552
552
            result.append((names, reader_func(None)))
564
564
    def test_read_bytes(self):
565
565
        """Test reading of both single bytes and all bytes in a hunk."""
566
566
        transport = self.get_transport()
567
 
        transport.put_bytes(b'sample', b'0123456789')
568
 
        f = pack.ReadVFile(transport.readv(b'sample', [(0, 1), (1, 2), (4, 1), (6, 2)]))
 
567
        transport.put_bytes('sample', b'0123456789')
 
568
        f = pack.ReadVFile(transport.readv(
 
569
            'sample', [(0, 1), (1, 2), (4, 1), (6, 2)]))
569
570
        results = []
570
571
        results.append(f.read(1))
571
572
        results.append(f.read(2))
580
581
        This is always within a readv hunk, never across it.
581
582
        """
582
583
        transport = self.get_transport()
583
 
        transport.put_bytes(b'sample', b'0\n2\n4\n')
584
 
        f = pack.ReadVFile(transport.readv(b'sample', [(0, 2), (2, 4)]))
 
584
        transport.put_bytes('sample', b'0\n2\n4\n')
 
585
        f = pack.ReadVFile(transport.readv('sample', [(0, 2), (2, 4)]))
585
586
        results = []
586
587
        results.append(f.readline())
587
588
        results.append(f.readline())
591
592
    def test_readline_and_read(self):
592
593
        """Test exercising one byte reads, readline, and then read again."""
593
594
        transport = self.get_transport()
594
 
        transport.put_bytes(b'sample', b'0\n2\n4\n')
595
 
        f = pack.ReadVFile(transport.readv(b'sample', [(0, 6)]))
 
595
        transport.put_bytes('sample', b'0\n2\n4\n')
 
596
        f = pack.ReadVFile(transport.readv('sample', [(0, 6)]))
596
597
        results = []
597
598
        results.append(f.read(1))
598
599
        results.append(f.readline())