17
17
"""Tests for Annotators."""
29
def load_tests(standard_tests, module, loader):
31
def load_tests(loader, standard_tests, pattern):
30
32
"""Parameterize tests for all versions of groupcompress."""
31
33
suite, _ = tests.permute_tests_for_extension(standard_tests, loader,
32
'bzrlib._annotator_py', 'bzrlib._annotator_pyx')
34
'breezy._annotator_py', 'breezy._annotator_pyx')
36
38
class TestAnnotator(tests.TestCaseWithMemoryTransport):
38
module = None # Set by load_tests
40
module = None # Set by load_tests
40
fa_key = ('f-id', 'a-id')
41
fb_key = ('f-id', 'b-id')
42
fc_key = ('f-id', 'c-id')
43
fd_key = ('f-id', 'd-id')
44
fe_key = ('f-id', 'e-id')
45
ff_key = ('f-id', 'f-id')
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')
47
49
def make_no_graph_texts(self):
48
50
factory = knit.make_pack_factory(False, False, 2)
49
51
self.vf = factory(self.get_transport())
50
52
self.ann = self.module.Annotator(self.vf)
51
self.vf.add_lines(self.fa_key, (), ['simple\n', 'content\n'])
52
self.vf.add_lines(self.fb_key, (), ['simple\n', 'new content\n'])
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'])
54
56
def make_simple_text(self):
55
57
# TODO: all we really need is a VersionedFile instance, we'd like to
111
113
# F-' 'simple|new content|'
112
114
self.vf.add_lines(self.fc_key, [self.fa_key],
113
['simple\n', 'new content\n'])
115
[b'simple\n', b'new content\n'])
114
116
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
115
['simple\n', 'new content\n'])
117
[b'simple\n', b'new content\n'])
116
118
self.vf.add_lines(self.fe_key, [self.fa_key],
117
['simple\n', 'new content\n'])
119
[b'simple\n', b'new content\n'])
118
120
self.vf.add_lines(self.ff_key, [self.fd_key, self.fe_key],
119
['simple\n', 'new content\n'])
121
[b'simple\n', b'new content\n'])
121
123
def make_merge_and_restored_text(self):
122
124
self.make_simple_text()
129
131
# D 'simple|content|'
130
132
# c reverts back to 'a' for the new content line
131
133
self.vf.add_lines(self.fc_key, [self.fb_key],
132
['simple\n', 'content\n'])
134
[b'simple\n', b'content\n'])
133
135
# d merges 'a' and 'c', to find both claim last modified
134
136
self.vf.add_lines(self.fd_key, [self.fa_key, self.fc_key],
135
['simple\n', 'content\n'])
137
[b'simple\n', b'content\n'])
137
139
def assertAnnotateEqual(self, expected_annotation, key, exp_text=None):
138
140
annotation, lines = self.ann.annotate(key)
139
141
self.assertEqual(expected_annotation, annotation)
140
142
if exp_text is None:
141
record = self.vf.get_record_stream([key], 'unordered', True).next()
143
record = next(self.vf.get_record_stream([key], 'unordered', True))
142
144
exp_text = record.get_bytes_as('fulltext')
143
self.assertEqualDiff(exp_text, ''.join(lines))
145
self.assertEqualDiff(exp_text, b''.join(lines))
145
147
def test_annotate_missing(self):
146
148
self.make_simple_text()
147
149
self.assertRaises(errors.RevisionNotPresent,
148
self.ann.annotate, ('not', 'present'))
150
self.ann.annotate, (b'not', b'present'))
150
152
def test_annotate_simple(self):
151
153
self.make_simple_text()
152
self.assertAnnotateEqual([(self.fa_key,)]*2, self.fa_key)
154
self.assertAnnotateEqual([(self.fa_key,)] * 2, self.fa_key)
153
155
self.assertAnnotateEqual([(self.fa_key,), (self.fb_key,)], self.fb_key)
155
157
def test_annotate_merge_text(self):
177
179
def test_annotate_flat_simple(self):
178
180
self.make_simple_text()
179
self.assertEqual([(self.fa_key, 'simple\n'),
180
(self.fa_key, 'content\n'),
181
], self.ann.annotate_flat(self.fa_key))
182
self.assertEqual([(self.fa_key, 'simple\n'),
183
(self.fb_key, 'new content\n'),
184
], self.ann.annotate_flat(self.fb_key))
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))
186
188
def test_annotate_flat_merge_and_restored_text(self):
187
189
self.make_merge_and_restored_text()
188
190
# fc is a simple dominator of fa
189
self.assertEqual([(self.fa_key, 'simple\n'),
190
(self.fc_key, 'content\n'),
191
], self.ann.annotate_flat(self.fd_key))
191
self.assertEqual([(self.fa_key, b'simple\n'),
192
(self.fc_key, b'content\n'),
193
], self.ann.annotate_flat(self.fd_key))
193
195
def test_annotate_common_merge_text(self):
194
196
self.make_common_merge_text()
195
197
# there is no common point, so we just pick the lexicographical lowest
196
# and 'b-id' comes before 'c-id'
197
self.assertEqual([(self.fa_key, 'simple\n'),
198
(self.fb_key, 'new content\n'),
199
], self.ann.annotate_flat(self.fd_key))
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))
201
203
def test_annotate_many_way_common_merge_text(self):
202
204
self.make_many_way_common_merge_text()
203
self.assertEqual([(self.fa_key, 'simple\n'),
204
(self.fb_key, 'new content\n')],
205
self.assertEqual([(self.fa_key, b'simple\n'),
206
(self.fb_key, b'new content\n')],
205
207
self.ann.annotate_flat(self.ff_key))
207
209
def test_annotate_flat_respects_break_ann_tie(self):
208
tiebreaker = annotate._break_annotation_tie
211
def custom_tiebreaker(annotated_lines):
212
self.assertEqual(2, len(annotated_lines))
213
left = annotated_lines[0]
214
self.assertEqual(2, len(left))
215
self.assertEqual('new content\n', left[1])
216
right = annotated_lines[1]
217
self.assertEqual(2, len(right))
218
self.assertEqual('new content\n', right[1])
219
calls.append((left[0], right[0]))
220
# Our custom tiebreaker takes the *largest* value, rather than
221
# the *smallest* value
222
if left[0] < right[0]:
226
annotate._break_annotation_tie = custom_tiebreaker
227
self.make_many_way_common_merge_text()
228
self.assertEqual([(self.fa_key, 'simple\n'),
229
(self.fe_key, 'new content\n')],
230
self.ann.annotate_flat(self.ff_key))
231
self.assertEqual([(self.fe_key, self.fc_key),
232
(self.fe_key, self.fb_key)], calls)
234
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)
237
235
def test_needed_keys_simple(self):
238
236
self.make_simple_text()
247
245
keys, ann_keys = self.ann._get_needed_keys(self.ff_key)
248
246
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
249
247
self.fd_key, self.fe_key, self.ff_key,
251
249
self.assertEqual({self.fa_key: 3,
257
}, self.ann._num_needed_children)
255
}, self.ann._num_needed_children)
258
256
self.assertEqual(set(), ann_keys)
260
258
def test_needed_keys_with_special_text(self):
261
259
self.make_many_way_common_merge_text()
262
spec_key = ('f-id', revision.CURRENT_REVISION)
263
spec_text = 'simple\nnew content\nlocally modified\n'
260
spec_key = (b'f-id', revision.CURRENT_REVISION)
261
spec_text = b'simple\nnew content\nlocally modified\n'
264
262
self.ann.add_special_text(spec_key, [self.fd_key, self.fe_key],
266
264
keys, ann_keys = self.ann._get_needed_keys(spec_key)
267
265
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
268
266
self.fd_key, self.fe_key,
270
268
self.assertEqual([spec_key], sorted(ann_keys))
272
270
def test_needed_keys_with_parent_texts(self):
324
}, self.ann._num_needed_children)
322
}, self.ann._num_needed_children)
325
323
self.assertTrue(self.fa_key in self.ann._text_cache)
326
324
self.assertTrue(self.fa_key in self.ann._annotations_cache)
327
325
self.ann._record_annotation(self.fc_key, [self.fa_key], [])
328
self.ann._record_annotation(self.fd_key, [self.fb_key, self.fc_key], [])
326
self.ann._record_annotation(
327
self.fd_key, [self.fb_key, self.fc_key], [])
329
328
self.assertEqual({self.fa_key: 1,
335
}, self.ann._num_needed_children)
334
}, self.ann._num_needed_children)
336
335
self.assertTrue(self.fa_key in self.ann._text_cache)
337
336
self.assertTrue(self.fa_key in self.ann._annotations_cache)
338
337
self.assertFalse(self.fb_key in self.ann._text_cache)