/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: Martin
  • Date: 2017-06-04 18:09:30 UTC
  • mto: This revision was merged to the branch mainline in revision 6653.
  • Revision ID: gzlist@googlemail.com-20170604180930-zpcenvzu13lilaax
Apply 2to3 xrange fix and fix up with sixish range

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_pack.py
1
 
# Copyright (C) 2007 Canonical Ltd
 
1
# Copyright (C) 2007, 2009, 2011, 2012, 2016 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
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
 
"""Tests for bzrlib.pack."""
18
 
 
19
 
 
20
 
from cStringIO import StringIO
21
 
 
22
 
from bzrlib import pack, errors, tests
 
17
"""Tests for breezy.pack."""
 
18
 
 
19
from .. import pack, errors, tests
 
20
from ..sixish import (
 
21
    BytesIO,
 
22
    )
23
23
 
24
24
 
25
25
class TestContainerSerialiser(tests.TestCase):
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
 
67
72
 
68
73
class TestContainerWriter(tests.TestCase):
69
74
 
70
75
    def setUp(self):
71
 
        tests.TestCase.setUp(self)
72
 
        self.output = StringIO()
 
76
        super(TestContainerWriter, self).setUp()
 
77
        self.output = BytesIO()
73
78
        self.writer = pack.ContainerWriter(self.output.write)
74
79
 
75
80
    def assertOutput(self, expected_output):
126
131
    def test_add_bytes_record_one_name(self):
127
132
        """Add a bytes record with one name."""
128
133
        self.writer.begin()
 
134
 
129
135
        offset, length = self.writer.add_bytes_record(
130
136
            'abc', names=[('name1', )])
131
137
        self.assertEqual((42, 13), (offset, length))
133
139
            'Bazaar pack format 1 (introduced in 0.18)\n'
134
140
            'B3\nname1\n\nabc')
135
141
 
 
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
 
136
169
    def test_add_bytes_record_two_names(self):
137
170
        """Add a bytes record with two names."""
138
171
        self.writer.begin()
200
233
    """
201
234
 
202
235
    def get_reader_for(self, bytes):
203
 
        stream = StringIO(bytes)
 
236
        stream = BytesIO(bytes)
204
237
        reader = pack.ContainerReader(stream)
205
238
        return reader
206
239
 
207
240
    def test_construct(self):
208
241
        """Test constructing a ContainerReader.
209
242
 
210
 
        This uses None as the output stream to show that the constructor doesn't
211
 
        try to use the input stream.
 
243
        This uses None as the output stream to show that the constructor
 
244
        doesn't try to use the input stream.
212
245
        """
213
246
        reader = pack.ContainerReader(None)
214
247
 
232
265
            "Bazaar pack format 1 (introduced in 0.18)\n")
233
266
        iterator = reader.iter_records()
234
267
        self.assertRaises(
235
 
            errors.UnexpectedEndOfContainerError, iterator.next)
 
268
            errors.UnexpectedEndOfContainerError, next, iterator)
236
269
 
237
270
    def test_unknown_record_type(self):
238
271
        """Unknown record types cause UnknownRecordTypeError to be raised."""
240
273
            "Bazaar pack format 1 (introduced in 0.18)\nX")
241
274
        iterator = reader.iter_records()
242
275
        self.assertRaises(
243
 
            errors.UnknownRecordTypeError, iterator.next)
 
276
            errors.UnknownRecordTypeError, next, iterator)
244
277
 
245
278
    def test_container_with_one_unnamed_record(self):
246
279
        """Read a container with one Bytes record.
259
292
 
260
293
    def test_validate_empty_container(self):
261
294
        """validate does not raise an error for a container with no records."""
262
 
        reader = self.get_reader_for("Bazaar pack format 1 (introduced in 0.18)\nE")
 
295
        reader = self.get_reader_for(
 
296
            "Bazaar pack format 1 (introduced in 0.18)\nE")
263
297
        # No exception raised
264
298
        reader.validate()
265
299
 
339
373
    """
340
374
 
341
375
    def get_reader_for(self, bytes):
342
 
        stream = StringIO(bytes)
 
376
        stream = BytesIO(bytes)
343
377
        reader = pack.BytesRecordReader(stream)
344
378
        return reader
345
379
 
496
530
class TestMakeReadvReader(tests.TestCaseWithTransport):
497
531
 
498
532
    def test_read_skipping_records(self):
499
 
        pack_data = StringIO()
 
533
        pack_data = BytesIO()
500
534
        writer = pack.ContainerWriter(pack_data.write)
501
535
        writer.begin()
502
536
        memos = []
705
739
        parser.accept_bytes("6\n\nabcdef")
706
740
        self.assertEqual([([], 'abcdef')], parser.read_pending_records())
707
741
        self.assertEqual([], parser.read_pending_records())
708
 
 
709