/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__groupcompress.py

  • Committer: Jelmer Vernooij
  • Date: 2017-07-23 22:06:41 UTC
  • mfrom: (6738 trunk)
  • mto: This revision was merged to the branch mainline in revision 6739.
  • Revision ID: jelmer@jelmer.uk-20170723220641-69eczax9bmv8d6kk
Merge trunk, address review comments.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test__groupcompress.py
1
 
# Copyright (C) 2008, 2009, 2010 Canonical Ltd
 
1
# Copyright (C) 2008-2011 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
16
16
 
17
17
"""Tests for the python and pyrex extensions of groupcompress"""
18
18
 
19
 
from bzrlib import (
20
 
    groupcompress,
 
19
from .. import (
 
20
    tests,
 
21
    )
 
22
from ..bzr import (
21
23
    _groupcompress_py,
22
 
    tests,
23
 
    )
24
 
 
25
 
 
26
 
def load_tests(standard_tests, module, loader):
27
 
    """Parameterize tests for all versions of groupcompress."""
28
 
    two_way_scenarios = [
29
 
        ('PP', {'make_delta': _groupcompress_py.make_delta,
30
 
                'apply_delta': _groupcompress_py.apply_delta})
31
 
        ]
 
24
    )
 
25
from .scenarios import (
 
26
    load_tests_apply_scenarios,
 
27
    )
 
28
from . import (
 
29
    features,
 
30
    )
 
31
 
 
32
 
 
33
def module_scenarios():
32
34
    scenarios = [
33
35
        ('python', {'_gc_module': _groupcompress_py}),
34
36
        ]
36
38
        gc_module = compiled_groupcompress_feature.module
37
39
        scenarios.append(('C',
38
40
            {'_gc_module': gc_module}))
39
 
        two_way_scenarios.extend([
 
41
    return scenarios
 
42
 
 
43
 
 
44
def two_way_scenarios():
 
45
    scenarios = [
 
46
        ('PP', {'make_delta': _groupcompress_py.make_delta,
 
47
                'apply_delta': _groupcompress_py.apply_delta})
 
48
        ]
 
49
    if compiled_groupcompress_feature.available():
 
50
        gc_module = compiled_groupcompress_feature.module
 
51
        scenarios.extend([
40
52
            ('CC', {'make_delta': gc_module.make_delta,
41
53
                    'apply_delta': gc_module.apply_delta}),
42
54
            ('PC', {'make_delta': _groupcompress_py.make_delta,
44
56
            ('CP', {'make_delta': gc_module.make_delta,
45
57
                    'apply_delta': _groupcompress_py.apply_delta}),
46
58
            ])
47
 
    to_adapt, result = tests.split_suite_by_condition(
48
 
        standard_tests, tests.condition_isinstance((TestMakeAndApplyDelta,
49
 
                                                    TestBase128Int)))
50
 
    result = tests.multiply_tests(to_adapt, scenarios, result)
51
 
    to_adapt, result = tests.split_suite_by_condition(result,
52
 
        tests.condition_isinstance(TestMakeAndApplyCompatible))
53
 
    result = tests.multiply_tests(to_adapt, two_way_scenarios, result)
54
 
    return result
55
 
 
56
 
 
57
 
compiled_groupcompress_feature = tests.ModuleAvailableFeature(
58
 
                                    'bzrlib._groupcompress_pyx')
 
59
    return scenarios
 
60
 
 
61
 
 
62
load_tests = load_tests_apply_scenarios
 
63
 
 
64
 
 
65
compiled_groupcompress_feature = features.ModuleAvailableFeature(
 
66
    'breezy.bzr._groupcompress_pyx')
59
67
 
60
68
_text1 = """\
61
69
This is a bit
116
124
 
117
125
class TestMakeAndApplyDelta(tests.TestCase):
118
126
 
 
127
    scenarios = module_scenarios()
119
128
    _gc_module = None # Set by load_tests
120
129
 
121
130
    def setUp(self):
226
235
 
227
236
class TestMakeAndApplyCompatible(tests.TestCase):
228
237
 
 
238
    scenarios = two_way_scenarios()
 
239
 
229
240
    make_delta = None # Set by load_tests
230
241
    apply_delta = None # Set by load_tests
231
242
 
258
269
        di = self._gc_module.DeltaIndex('test text\n')
259
270
        self.assertEqual('DeltaIndex(1, 10)', repr(di))
260
271
 
 
272
    def test__dump_no_index(self):
 
273
        di = self._gc_module.DeltaIndex()
 
274
        self.assertEqual(None, di._dump_index())
 
275
 
 
276
    def test__dump_index_simple(self):
 
277
        di = self._gc_module.DeltaIndex()
 
278
        di.add_source(_text1, 0)
 
279
        self.assertFalse(di._has_index())
 
280
        self.assertEqual(None, di._dump_index())
 
281
        _ = di.make_delta(_text1)
 
282
        self.assertTrue(di._has_index())
 
283
        hash_list, entry_list = di._dump_index()
 
284
        self.assertEqual(16, len(hash_list))
 
285
        self.assertEqual(68, len(entry_list))
 
286
        just_entries = [(idx, text_offset, hash_val)
 
287
                        for idx, (text_offset, hash_val)
 
288
                         in enumerate(entry_list)
 
289
                         if text_offset != 0 or hash_val != 0]
 
290
        rabin_hash = self._gc_module._rabin_hash
 
291
        self.assertEqual([(8, 16, rabin_hash(_text1[1:17])),
 
292
                          (25, 48, rabin_hash(_text1[33:49])),
 
293
                          (34, 32, rabin_hash(_text1[17:33])),
 
294
                          (47, 64, rabin_hash(_text1[49:65])),
 
295
                         ], just_entries)
 
296
        # This ensures that the hash map points to the location we expect it to
 
297
        for entry_idx, text_offset, hash_val in just_entries:
 
298
            self.assertEqual(entry_idx, hash_list[hash_val & 0xf])
 
299
 
 
300
    def test__dump_index_two_sources(self):
 
301
        di = self._gc_module.DeltaIndex()
 
302
        di.add_source(_text1, 0)
 
303
        di.add_source(_text2, 2)
 
304
        start2 = len(_text1) + 2
 
305
        self.assertTrue(di._has_index())
 
306
        hash_list, entry_list = di._dump_index()
 
307
        self.assertEqual(16, len(hash_list))
 
308
        self.assertEqual(68, len(entry_list))
 
309
        just_entries = [(idx, text_offset, hash_val)
 
310
                        for idx, (text_offset, hash_val)
 
311
                         in enumerate(entry_list)
 
312
                         if text_offset != 0 or hash_val != 0]
 
313
        rabin_hash = self._gc_module._rabin_hash
 
314
        self.assertEqual([(8, 16, rabin_hash(_text1[1:17])),
 
315
                          (9, start2+16, rabin_hash(_text2[1:17])),
 
316
                          (25, 48, rabin_hash(_text1[33:49])),
 
317
                          (30, start2+64, rabin_hash(_text2[49:65])),
 
318
                          (34, 32, rabin_hash(_text1[17:33])),
 
319
                          (35, start2+32, rabin_hash(_text2[17:33])),
 
320
                          (43, start2+48, rabin_hash(_text2[33:49])),
 
321
                          (47, 64, rabin_hash(_text1[49:65])),
 
322
                         ], just_entries)
 
323
        # Each entry should be in the appropriate hash bucket.
 
324
        for entry_idx, text_offset, hash_val in just_entries:
 
325
            hash_idx = hash_val & 0xf
 
326
            self.assertTrue(
 
327
                hash_list[hash_idx] <= entry_idx < hash_list[hash_idx+1])
 
328
 
261
329
    def test_first_add_source_doesnt_index_until_make_delta(self):
262
330
        di = self._gc_module.DeltaIndex()
263
331
        self.assertFalse(di._has_index())
269
337
        self.assertTrue(di._has_index())
270
338
        self.assertEqual('N\x90/\x1fdiffer from\nagainst other text\n', delta)
271
339
 
 
340
    def test_add_source_max_bytes_to_index(self):
 
341
        di = self._gc_module.DeltaIndex()
 
342
        di._max_bytes_to_index = 3*16
 
343
        di.add_source(_text1, 0) # (77 bytes -1) // 3 = 25 byte stride
 
344
        di.add_source(_text3, 3) # (135 bytes -1) // 3 = 44 byte stride
 
345
        start2 = len(_text1) + 3
 
346
        hash_list, entry_list = di._dump_index()
 
347
        self.assertEqual(16, len(hash_list))
 
348
        self.assertEqual(67, len(entry_list))
 
349
        just_entries = sorted([(text_offset, hash_val)
 
350
                               for text_offset, hash_val in entry_list
 
351
                                if text_offset != 0 or hash_val != 0])
 
352
        rabin_hash = self._gc_module._rabin_hash
 
353
        self.assertEqual([(25, rabin_hash(_text1[10:26])),
 
354
                          (50, rabin_hash(_text1[35:51])),
 
355
                          (75, rabin_hash(_text1[60:76])),
 
356
                          (start2+44, rabin_hash(_text3[29:45])),
 
357
                          (start2+88, rabin_hash(_text3[73:89])),
 
358
                          (start2+132, rabin_hash(_text3[117:133])),
 
359
                         ], just_entries)
 
360
 
272
361
    def test_second_add_source_triggers_make_index(self):
273
362
        di = self._gc_module.DeltaIndex()
274
363
        self.assertFalse(di._has_index())
457
546
 
458
547
class TestBase128Int(tests.TestCase):
459
548
 
 
549
    scenarios = module_scenarios()
 
550
 
460
551
    _gc_module = None # Set by load_tests
461
552
 
462
553
    def assertEqualEncode(self, bytes, val):