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