31
31
def load_tests(loader, standard_tests, pattern):
32
32
"""Parameterize tests for all versions of groupcompress."""
33
33
suite, _ = tests.permute_tests_for_extension(standard_tests, loader,
34
'breezy._annotator_py', 'breezy._annotator_pyx')
34
'breezy._annotator_py', 'breezy._annotator_pyx')
38
38
class TestAnnotator(tests.TestCaseWithMemoryTransport):
40
module = None # Set by load_tests
40
module = None # Set by load_tests
42
fa_key = (b'f-id', b'a-id')
43
fb_key = (b'f-id', b'b-id')
44
fc_key = (b'f-id', b'c-id')
45
fd_key = (b'f-id', b'd-id')
46
fe_key = (b'f-id', b'e-id')
47
ff_key = (b'f-id', b'f-id')
42
fa_key = ('f-id', 'a-id')
43
fb_key = ('f-id', 'b-id')
44
fc_key = ('f-id', 'c-id')
45
fd_key = ('f-id', 'd-id')
46
fe_key = ('f-id', 'e-id')
47
ff_key = ('f-id', 'f-id')
49
49
def make_no_graph_texts(self):
50
50
factory = knit.make_pack_factory(False, False, 2)
51
51
self.vf = factory(self.get_transport())
52
52
self.ann = self.module.Annotator(self.vf)
53
self.vf.add_lines(self.fa_key, (), [b'simple\n', b'content\n'])
54
self.vf.add_lines(self.fb_key, (), [b'simple\n', b'new content\n'])
53
self.vf.add_lines(self.fa_key, (), ['simple\n', 'content\n'])
54
self.vf.add_lines(self.fb_key, (), ['simple\n', 'new content\n'])
56
56
def make_simple_text(self):
57
57
# TODO: all we really need is a VersionedFile instance, we'd like to
113
113
# F-' 'simple|new content|'
114
114
self.vf.add_lines(self.fc_key, [self.fa_key],
115
[b'simple\n', b'new content\n'])
115
['simple\n', 'new content\n'])
116
116
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
117
[b'simple\n', b'new content\n'])
117
['simple\n', 'new content\n'])
118
118
self.vf.add_lines(self.fe_key, [self.fa_key],
119
[b'simple\n', b'new content\n'])
119
['simple\n', 'new content\n'])
120
120
self.vf.add_lines(self.ff_key, [self.fd_key, self.fe_key],
121
[b'simple\n', b'new content\n'])
121
['simple\n', 'new content\n'])
123
123
def make_merge_and_restored_text(self):
124
124
self.make_simple_text()
142
142
if exp_text is None:
143
143
record = next(self.vf.get_record_stream([key], 'unordered', True))
144
144
exp_text = record.get_bytes_as('fulltext')
145
self.assertEqualDiff(exp_text, b''.join(lines))
145
self.assertEqualDiff(exp_text, ''.join(lines))
147
147
def test_annotate_missing(self):
148
148
self.make_simple_text()
149
149
self.assertRaises(errors.RevisionNotPresent,
150
self.ann.annotate, (b'not', b'present'))
150
self.ann.annotate, ('not', 'present'))
152
152
def test_annotate_simple(self):
153
153
self.make_simple_text()
154
self.assertAnnotateEqual([(self.fa_key,)] * 2, self.fa_key)
154
self.assertAnnotateEqual([(self.fa_key,)]*2, self.fa_key)
155
155
self.assertAnnotateEqual([(self.fa_key,), (self.fb_key,)], self.fb_key)
157
157
def test_annotate_merge_text(self):
179
179
def test_annotate_flat_simple(self):
180
180
self.make_simple_text()
181
self.assertEqual([(self.fa_key, b'simple\n'),
182
(self.fa_key, b'content\n'),
183
], self.ann.annotate_flat(self.fa_key))
184
self.assertEqual([(self.fa_key, b'simple\n'),
185
(self.fb_key, b'new content\n'),
186
], self.ann.annotate_flat(self.fb_key))
181
self.assertEqual([(self.fa_key, 'simple\n'),
182
(self.fa_key, 'content\n'),
183
], self.ann.annotate_flat(self.fa_key))
184
self.assertEqual([(self.fa_key, 'simple\n'),
185
(self.fb_key, 'new content\n'),
186
], self.ann.annotate_flat(self.fb_key))
188
188
def test_annotate_flat_merge_and_restored_text(self):
189
189
self.make_merge_and_restored_text()
190
190
# fc is a simple dominator of fa
191
self.assertEqual([(self.fa_key, b'simple\n'),
192
(self.fc_key, b'content\n'),
193
], self.ann.annotate_flat(self.fd_key))
191
self.assertEqual([(self.fa_key, 'simple\n'),
192
(self.fc_key, 'content\n'),
193
], self.ann.annotate_flat(self.fd_key))
195
195
def test_annotate_common_merge_text(self):
196
196
self.make_common_merge_text()
197
197
# there is no common point, so we just pick the lexicographical lowest
198
# and b'b-id' comes before b'c-id'
199
self.assertEqual([(self.fa_key, b'simple\n'),
200
(self.fb_key, b'new content\n'),
201
], self.ann.annotate_flat(self.fd_key))
198
# and 'b-id' comes before 'c-id'
199
self.assertEqual([(self.fa_key, 'simple\n'),
200
(self.fb_key, 'new content\n'),
201
], self.ann.annotate_flat(self.fd_key))
203
203
def test_annotate_many_way_common_merge_text(self):
204
204
self.make_many_way_common_merge_text()
205
self.assertEqual([(self.fa_key, b'simple\n'),
206
(self.fb_key, b'new content\n')],
205
self.assertEqual([(self.fa_key, 'simple\n'),
206
(self.fb_key, 'new content\n')],
207
207
self.ann.annotate_flat(self.ff_key))
209
209
def test_annotate_flat_respects_break_ann_tie(self):
210
tiebreaker = annotate._break_annotation_tie
213
def custom_tiebreaker(annotated_lines):
214
self.assertEqual(2, len(annotated_lines))
215
left = annotated_lines[0]
216
self.assertEqual(2, len(left))
217
self.assertEqual('new content\n', left[1])
218
right = annotated_lines[1]
219
self.assertEqual(2, len(right))
220
self.assertEqual('new content\n', right[1])
221
calls.append((left[0], right[0]))
222
# Our custom tiebreaker takes the *largest* value, rather than
223
# the *smallest* value
224
if left[0] < right[0]:
228
annotate._break_annotation_tie = custom_tiebreaker
229
self.make_many_way_common_merge_text()
230
self.assertEqual([(self.fa_key, 'simple\n'),
231
(self.fe_key, 'new content\n')],
232
self.ann.annotate_flat(self.ff_key))
233
self.assertEqual([(self.fe_key, self.fc_key),
234
(self.fe_key, self.fb_key)], calls)
236
annotate._break_annotation_tie = tiebreaker
212
def custom_tiebreaker(annotated_lines):
213
self.assertEqual(2, len(annotated_lines))
214
left = annotated_lines[0]
215
self.assertEqual(2, len(left))
216
self.assertEqual(b'new content\n', left[1])
217
right = annotated_lines[1]
218
self.assertEqual(2, len(right))
219
self.assertEqual(b'new content\n', right[1])
220
seen.update([left[0], right[0]])
221
# Our custom tiebreaker takes the *largest* value, rather than
222
# the *smallest* value
223
if left[0] < right[0]:
227
self.overrideAttr(annotate, '_break_annotation_tie', custom_tiebreaker)
228
self.make_many_way_common_merge_text()
229
self.assertEqual([(self.fa_key, b'simple\n'),
230
(self.fe_key, b'new content\n')],
231
self.ann.annotate_flat(self.ff_key))
232
# Calls happen in set iteration order but should keys should be seen
233
self.assertEqual({self.fb_key, self.fc_key, self.fe_key}, seen)
235
239
def test_needed_keys_simple(self):
236
240
self.make_simple_text()
245
249
keys, ann_keys = self.ann._get_needed_keys(self.ff_key)
246
250
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
247
251
self.fd_key, self.fe_key, self.ff_key,
249
253
self.assertEqual({self.fa_key: 3,
255
}, self.ann._num_needed_children)
259
}, self.ann._num_needed_children)
256
260
self.assertEqual(set(), ann_keys)
258
262
def test_needed_keys_with_special_text(self):
259
263
self.make_many_way_common_merge_text()
260
spec_key = (b'f-id', revision.CURRENT_REVISION)
261
spec_text = b'simple\nnew content\nlocally modified\n'
264
spec_key = ('f-id', revision.CURRENT_REVISION)
265
spec_text = 'simple\nnew content\nlocally modified\n'
262
266
self.ann.add_special_text(spec_key, [self.fd_key, self.fe_key],
264
268
keys, ann_keys = self.ann._get_needed_keys(spec_key)
265
269
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
266
270
self.fd_key, self.fe_key,
268
272
self.assertEqual([spec_key], sorted(ann_keys))
270
274
def test_needed_keys_with_parent_texts(self):
322
}, self.ann._num_needed_children)
326
}, self.ann._num_needed_children)
323
327
self.assertTrue(self.fa_key in self.ann._text_cache)
324
328
self.assertTrue(self.fa_key in self.ann._annotations_cache)
325
329
self.ann._record_annotation(self.fc_key, [self.fa_key], [])
326
self.ann._record_annotation(
327
self.fd_key, [self.fb_key, self.fc_key], [])
330
self.ann._record_annotation(self.fd_key, [self.fb_key, self.fc_key], [])
328
331
self.assertEqual({self.fa_key: 1,
334
}, self.ann._num_needed_children)
337
}, self.ann._num_needed_children)
335
338
self.assertTrue(self.fa_key in self.ann._text_cache)
336
339
self.assertTrue(self.fa_key in self.ann._annotations_cache)
337
340
self.assertFalse(self.fb_key in self.ann._text_cache)