1
# Copyright (C) 2009 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Tests for Annotators."""
29
def load_tests(standard_tests, module, loader):
30
"""Parameterize tests for all versions of groupcompress."""
32
('python', {'module': _annotator_py}),
34
suite = loader.suiteClass()
35
if compiled_annotator_feature.available():
36
scenarios.append(('C', {'module': compiled_annotator_feature.module}))
38
# the compiled module isn't available, so we add a failing test
39
class FailWithoutFeature(tests.TestCase):
41
self.requireFeature(compiled_annotator_feature)
42
suite.addTest(loader.loadTestsFromTestCase(FailWithoutFeature))
43
result = tests.multiply_tests(standard_tests, scenarios, suite)
47
compiled_annotator_feature = tests.ModuleAvailableFeature(
48
'bzrlib._annotator_pyx')
51
class TestAnnotator(tests.TestCaseWithMemoryTransport):
53
module = None # Set by load_tests
55
fa_key = ('f-id', 'a-id')
56
fb_key = ('f-id', 'b-id')
57
fc_key = ('f-id', 'c-id')
58
fd_key = ('f-id', 'd-id')
59
fe_key = ('f-id', 'e-id')
60
ff_key = ('f-id', 'f-id')
62
def make_no_graph_texts(self):
63
factory = knit.make_pack_factory(False, False, 2)
64
self.vf = factory(self.get_transport())
65
self.ann = self.module.Annotator(self.vf)
66
self.vf.add_lines(self.fa_key, (), ['simple\n', 'content\n'])
67
self.vf.add_lines(self.fb_key, (), ['simple\n', 'new content\n'])
69
def make_simple_text(self):
70
# TODO: all we really need is a VersionedFile instance, we'd like to
71
# avoid creating all the intermediate stuff
72
factory = knit.make_pack_factory(True, True, 2)
73
self.vf = factory(self.get_transport())
74
# This assumes nothing special happens during __init__, which may be
76
self.ann = self.module.Annotator(self.vf)
79
# B 'simple|new content|'
80
self.vf.add_lines(self.fa_key, [], ['simple\n', 'content\n'])
81
self.vf.add_lines(self.fb_key, [self.fa_key],
82
['simple\n', 'new content\n'])
84
def make_merge_text(self):
85
self.make_simple_text()
88
# B | 'simple|new content|'
90
# | C 'simple|from c|content|'
92
# D 'simple|from c|new content|introduced in merge|'
93
self.vf.add_lines(self.fc_key, [self.fa_key],
94
['simple\n', 'from c\n', 'content\n'])
95
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
96
['simple\n', 'from c\n', 'new content\n',
97
'introduced in merge\n'])
99
def make_common_merge_text(self):
100
"""Both sides of the merge will have introduced a line."""
101
self.make_simple_text()
102
# A 'simple|content|'
104
# B | 'simple|new content|'
106
# | C 'simple|new content|'
108
# D 'simple|new content|'
109
self.vf.add_lines(self.fc_key, [self.fa_key],
110
['simple\n', 'new content\n'])
111
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
112
['simple\n', 'new content\n'])
114
def make_many_way_common_merge_text(self):
115
self.make_simple_text()
116
# A-. 'simple|content|'
118
# B | | 'simple|new content|'
120
# | C | 'simple|new content|'
122
# D | 'simple|new content|'
124
# | E 'simple|new content|'
126
# F-' 'simple|new content|'
127
self.vf.add_lines(self.fc_key, [self.fa_key],
128
['simple\n', 'new content\n'])
129
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
130
['simple\n', 'new content\n'])
131
self.vf.add_lines(self.fe_key, [self.fa_key],
132
['simple\n', 'new content\n'])
133
self.vf.add_lines(self.ff_key, [self.fd_key, self.fe_key],
134
['simple\n', 'new content\n'])
136
def make_merge_and_restored_text(self):
137
self.make_simple_text()
138
# A 'simple|content|'
140
# B | 'simple|new content|'
142
# C | 'simple|content|' # reverted to A
144
# D 'simple|content|'
145
# c reverts back to 'a' for the new content line
146
self.vf.add_lines(self.fc_key, [self.fb_key],
147
['simple\n', 'content\n'])
148
# d merges 'a' and 'c', to find both claim last modified
149
self.vf.add_lines(self.fd_key, [self.fa_key, self.fc_key],
150
['simple\n', 'content\n'])
152
def assertAnnotateEqual(self, expected_annotation, key, exp_text=None):
153
annotation, lines = self.ann.annotate(key)
154
self.assertEqual(expected_annotation, annotation)
156
record = self.vf.get_record_stream([key], 'unordered', True).next()
157
exp_text = record.get_bytes_as('fulltext')
158
self.assertEqualDiff(exp_text, ''.join(lines))
160
def test_annotate_missing(self):
161
self.make_simple_text()
162
self.assertRaises(errors.RevisionNotPresent,
163
self.ann.annotate, ('not', 'present'))
165
def test_annotate_simple(self):
166
self.make_simple_text()
167
self.assertAnnotateEqual([(self.fa_key,)]*2, self.fa_key)
168
self.assertAnnotateEqual([(self.fa_key,), (self.fb_key,)], self.fb_key)
170
def test_annotate_merge_text(self):
171
self.make_merge_text()
172
self.assertAnnotateEqual([(self.fa_key,), (self.fc_key,),
173
(self.fb_key,), (self.fd_key,)],
176
def test_annotate_common_merge_text(self):
177
self.make_common_merge_text()
178
self.assertAnnotateEqual([(self.fa_key,), (self.fb_key, self.fc_key)],
181
def test_annotate_many_way_common_merge_text(self):
182
self.make_many_way_common_merge_text()
183
self.assertAnnotateEqual([(self.fa_key,),
184
(self.fb_key, self.fc_key, self.fe_key)],
187
def test_annotate_merge_and_restored(self):
188
self.make_merge_and_restored_text()
189
self.assertAnnotateEqual([(self.fa_key,), (self.fa_key, self.fc_key)],
192
def test_annotate_flat_simple(self):
193
self.make_simple_text()
194
self.assertEqual([(self.fa_key, 'simple\n'),
195
(self.fa_key, 'content\n'),
196
], self.ann.annotate_flat(self.fa_key))
197
self.assertEqual([(self.fa_key, 'simple\n'),
198
(self.fb_key, 'new content\n'),
199
], self.ann.annotate_flat(self.fb_key))
201
def test_annotate_flat_merge_and_restored_text(self):
202
self.make_merge_and_restored_text()
203
# fc is a simple dominator of fa
204
self.assertEqual([(self.fa_key, 'simple\n'),
205
(self.fc_key, 'content\n'),
206
], self.ann.annotate_flat(self.fd_key))
208
def test_annotate_common_merge_text(self):
209
self.make_common_merge_text()
210
# there is no common point, so we just pick the lexicographical lowest
211
# and 'b-id' comes before 'c-id'
212
self.assertEqual([(self.fa_key, 'simple\n'),
213
(self.fb_key, 'new content\n'),
214
], self.ann.annotate_flat(self.fd_key))
216
def test_annotate_many_way_common_merge_text(self):
217
self.make_many_way_common_merge_text()
218
self.assertEqual([(self.fa_key, 'simple\n'),
219
(self.fb_key, 'new content\n')],
220
self.ann.annotate_flat(self.ff_key))
222
def test_annotate_flat_respects_break_ann_tie(self):
223
tiebreaker = annotate._break_annotation_tie
226
def custom_tiebreaker(annotated_lines):
227
self.assertEqual(2, len(annotated_lines))
228
left = annotated_lines[0]
229
self.assertEqual(2, len(left))
230
self.assertEqual('new content\n', left[1])
231
right = annotated_lines[1]
232
self.assertEqual(2, len(right))
233
self.assertEqual('new content\n', right[1])
234
calls.append((left[0], right[0]))
235
# Our custom tiebreaker takes the *largest* value, rather than
236
# the *smallest* value
237
if left[0] < right[0]:
241
annotate._break_annotation_tie = custom_tiebreaker
242
self.make_many_way_common_merge_text()
243
self.assertEqual([(self.fa_key, 'simple\n'),
244
(self.fe_key, 'new content\n')],
245
self.ann.annotate_flat(self.ff_key))
246
self.assertEqual([(self.fe_key, self.fc_key),
247
(self.fe_key, self.fb_key)], calls)
249
annotate._break_annotation_tie = tiebreaker
252
def test_needed_keys_simple(self):
253
self.make_simple_text()
254
keys, ann_keys = self.ann._get_needed_keys(self.fb_key)
255
self.assertEqual([self.fa_key, self.fb_key], sorted(keys))
256
self.assertEqual({self.fa_key: 1, self.fb_key: 1},
257
self.ann._num_needed_children)
258
self.assertEqual(set(), ann_keys)
260
def test_needed_keys_many(self):
261
self.make_many_way_common_merge_text()
262
keys, ann_keys = self.ann._get_needed_keys(self.ff_key)
263
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
264
self.fd_key, self.fe_key, self.ff_key,
266
self.assertEqual({self.fa_key: 3,
272
}, self.ann._num_needed_children)
273
self.assertEqual(set(), ann_keys)
275
def test_needed_keys_with_special_text(self):
276
self.make_many_way_common_merge_text()
277
spec_key = ('f-id', revision.CURRENT_REVISION)
278
spec_text = 'simple\nnew content\nlocally modified\n'
279
self.ann.add_special_text(spec_key, [self.fd_key, self.fe_key],
281
keys, ann_keys = self.ann._get_needed_keys(spec_key)
282
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
283
self.fd_key, self.fe_key,
285
self.assertEqual([spec_key], sorted(ann_keys))
287
def test_needed_keys_with_parent_texts(self):
288
self.make_many_way_common_merge_text()
289
# If 'D' and 'E' are already annotated, we don't need to extract all
291
# D | 'simple|new content|'
293
# | E 'simple|new content|'
295
# F-' 'simple|new content|'
296
self.ann._parent_map[self.fd_key] = (self.fb_key, self.fc_key)
297
self.ann._text_cache[self.fd_key] = ['simple\n', 'new content\n']
298
self.ann._annotations_cache[self.fd_key] = [
300
(self.fb_key, self.fc_key),
302
self.ann._parent_map[self.fe_key] = (self.fa_key,)
303
self.ann._text_cache[self.fe_key] = ['simple\n', 'new content\n']
304
self.ann._annotations_cache[self.fe_key] = [
308
keys, ann_keys = self.ann._get_needed_keys(self.ff_key)
309
self.assertEqual([self.ff_key], sorted(keys))
310
self.assertEqual({self.fd_key: 1,
313
}, self.ann._num_needed_children)
314
self.assertEqual([], sorted(ann_keys))
316
def test_record_annotation_removes_texts(self):
317
self.make_many_way_common_merge_text()
318
# Populate the caches
319
for x in self.ann._get_needed_texts(self.ff_key):
321
self.assertEqual({self.fa_key: 3,
327
}, self.ann._num_needed_children)
328
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
329
self.fd_key, self.fe_key, self.ff_key,
330
], sorted(self.ann._text_cache.keys()))
331
self.ann._record_annotation(self.fa_key, [], [])
332
self.ann._record_annotation(self.fb_key, [self.fa_key], [])
333
self.assertEqual({self.fa_key: 2,
339
}, self.ann._num_needed_children)
340
self.assertTrue(self.fa_key in self.ann._text_cache)
341
self.assertTrue(self.fa_key in self.ann._annotations_cache)
342
self.ann._record_annotation(self.fc_key, [self.fa_key], [])
343
self.ann._record_annotation(self.fd_key, [self.fb_key, self.fc_key], [])
344
self.assertEqual({self.fa_key: 1,
350
}, self.ann._num_needed_children)
351
self.assertTrue(self.fa_key in self.ann._text_cache)
352
self.assertTrue(self.fa_key in self.ann._annotations_cache)
353
self.assertFalse(self.fb_key in self.ann._text_cache)
354
self.assertFalse(self.fb_key in self.ann._annotations_cache)
355
self.assertFalse(self.fc_key in self.ann._text_cache)
356
self.assertFalse(self.fc_key in self.ann._annotations_cache)
358
def test_annotate_special_text(self):
359
# Things like WT and PreviewTree want to annotate an arbitrary text
360
# ('current:') so we need a way to add that to the group of files to be
362
self.make_many_way_common_merge_text()
363
# A-. 'simple|content|'
365
# B | | 'simple|new content|'
367
# | C | 'simple|new content|'
369
# D | 'simple|new content|'
371
# | E 'simple|new content|'
373
# SPEC 'simple|new content|locally modified|'
374
spec_key = ('f-id', revision.CURRENT_REVISION)
375
spec_text = 'simple\nnew content\nlocally modified\n'
376
self.ann.add_special_text(spec_key, [self.fd_key, self.fe_key],
378
self.assertAnnotateEqual([(self.fa_key,),
379
(self.fb_key, self.fc_key, self.fe_key),
384
def test_no_graph(self):
385
self.make_no_graph_texts()
386
self.assertAnnotateEqual([(self.fa_key,),
389
self.assertAnnotateEqual([(self.fb_key,),