← 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: Canonical.com Patch Queue Manager
  • Date: 2010-09-29 22:03:03 UTC
  • mfrom: (5416.2.6 jam-integration)
  • Revision ID: pqm@pqm.ubuntu.com-20100929220303-cr95h8iwtggco721
(mbp) Add 'break-lock --force'

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_pack.py
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()
 
71
        tests.TestCase.setUp(self)
77
72
        self.output = StringIO()
78
73
        self.writer = pack.ContainerWriter(self.output.write)
79
74
 
131
126
    def test_add_bytes_record_one_name(self):
132
127
        """Add a bytes record with one name."""
133
128
        self.writer.begin()
134
 
 
135
129
        offset, length = self.writer.add_bytes_record(
136
130
            'abc', names=[('name1', )])
137
131
        self.assertEqual((42, 13), (offset, length))
139
133
            'Bazaar pack format 1 (introduced in 0.18)\n'
140
134
            'B3\nname1\n\nabc')
141
135
 
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.assertEquals([
164
 
            'Bazaar pack format 1 (introduced in 0.18)\n',
165
 
            'B6\nname1\n\n',
166
 
            'abcabc'],
167
 
            writes)
168
 
 
169
136
    def test_add_bytes_record_two_names(self):
170
137
        """Add a bytes record with two names."""
171
138
        self.writer.begin()
240
207
    def test_construct(self):
241
208
        """Test constructing a ContainerReader.
242
209
 
243
 
        This uses None as the output stream to show that the constructor
244
 
        doesn't try to use the input stream.
 
210
        This uses None as the output stream to show that the constructor doesn't
 
211
        try to use the input stream.
245
212
        """
246
213
        reader = pack.ContainerReader(None)
247
214
 
292
259
 
293
260
    def test_validate_empty_container(self):
294
261
        """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")
 
262
        reader = self.get_reader_for("Bazaar pack format 1 (introduced in 0.18)\nE")
297
263
        # No exception raised
298
264
        reader.validate()
299
265
 
739
705
        parser.accept_bytes("6\n\nabcdef")
740
706
        self.assertEqual([([], 'abcdef')], parser.read_pending_records())
741
707
        self.assertEqual([], parser.read_pending_records())
 
708
 
 
709