1
# Copyright (C) 2009, 2010, 2011 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."""
28
def load_tests(standard_tests, module, loader):
29
"""Parameterize tests for all versions of groupcompress."""
30
suite, _ = tests.permute_tests_for_extension(standard_tests, loader,
31
'bzrlib._annotator_py', 'bzrlib._annotator_pyx')
35
class TestAnnotator(tests.TestCaseWithMemoryTransport):
37
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')
46
def make_no_graph_texts(self):
47
factory = knit.make_pack_factory(False, False, 2)
48
self.vf = factory(self.get_transport())
49
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
def make_simple_text(self):
54
# TODO: all we really need is a VersionedFile instance, we'd like to
55
# avoid creating all the intermediate stuff
56
factory = knit.make_pack_factory(True, True, 2)
57
self.vf = factory(self.get_transport())
58
# This assumes nothing special happens during __init__, which may be
60
self.ann = self.module.Annotator(self.vf)
63
# B 'simple|new content|'
64
self.vf.add_lines(self.fa_key, [], ['simple\n', 'content\n'])
65
self.vf.add_lines(self.fb_key, [self.fa_key],
66
['simple\n', 'new content\n'])
68
def make_merge_text(self):
69
self.make_simple_text()
72
# B | 'simple|new content|'
74
# | C 'simple|from c|content|'
76
# D 'simple|from c|new content|introduced in merge|'
77
self.vf.add_lines(self.fc_key, [self.fa_key],
78
['simple\n', 'from c\n', 'content\n'])
79
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
80
['simple\n', 'from c\n', 'new content\n',
81
'introduced in merge\n'])
83
def make_common_merge_text(self):
84
"""Both sides of the merge will have introduced a line."""
85
self.make_simple_text()
88
# B | 'simple|new content|'
90
# | C 'simple|new content|'
92
# D 'simple|new content|'
93
self.vf.add_lines(self.fc_key, [self.fa_key],
94
['simple\n', 'new content\n'])
95
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
96
['simple\n', 'new content\n'])
98
def make_many_way_common_merge_text(self):
99
self.make_simple_text()
100
# A-. 'simple|content|'
102
# B | | 'simple|new content|'
104
# | C | 'simple|new content|'
106
# D | 'simple|new content|'
108
# | E 'simple|new content|'
110
# F-' 'simple|new content|'
111
self.vf.add_lines(self.fc_key, [self.fa_key],
112
['simple\n', 'new content\n'])
113
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
114
['simple\n', 'new content\n'])
115
self.vf.add_lines(self.fe_key, [self.fa_key],
116
['simple\n', 'new content\n'])
117
self.vf.add_lines(self.ff_key, [self.fd_key, self.fe_key],
118
['simple\n', 'new content\n'])
120
def make_merge_and_restored_text(self):
121
self.make_simple_text()
122
# A 'simple|content|'
124
# B | 'simple|new content|'
126
# C | 'simple|content|' # reverted to A
128
# D 'simple|content|'
129
# c reverts back to 'a' for the new content line
130
self.vf.add_lines(self.fc_key, [self.fb_key],
131
['simple\n', 'content\n'])
132
# d merges 'a' and 'c', to find both claim last modified
133
self.vf.add_lines(self.fd_key, [self.fa_key, self.fc_key],
134
['simple\n', 'content\n'])
136
def assertAnnotateEqual(self, expected_annotation, key, exp_text=None):
137
annotation, lines = self.ann.annotate(key)
138
self.assertEqual(expected_annotation, annotation)
140
record = self.vf.get_record_stream([key], 'unordered', True).next()
141
exp_text = record.get_bytes_as('fulltext')
142
self.assertEqualDiff(exp_text, ''.join(lines))
144
def test_annotate_missing(self):
145
self.make_simple_text()
146
self.assertRaises(errors.RevisionNotPresent,
147
self.ann.annotate, ('not', 'present'))
149
def test_annotate_simple(self):
150
self.make_simple_text()
151
self.assertAnnotateEqual([(self.fa_key,)]*2, self.fa_key)
152
self.assertAnnotateEqual([(self.fa_key,), (self.fb_key,)], self.fb_key)
154
def test_annotate_merge_text(self):
155
self.make_merge_text()
156
self.assertAnnotateEqual([(self.fa_key,), (self.fc_key,),
157
(self.fb_key,), (self.fd_key,)],
160
def test_annotate_common_merge_text(self):
161
self.make_common_merge_text()
162
self.assertAnnotateEqual([(self.fa_key,), (self.fb_key, self.fc_key)],
165
def test_annotate_many_way_common_merge_text(self):
166
self.make_many_way_common_merge_text()
167
self.assertAnnotateEqual([(self.fa_key,),
168
(self.fb_key, self.fc_key, self.fe_key)],
171
def test_annotate_merge_and_restored(self):
172
self.make_merge_and_restored_text()
173
self.assertAnnotateEqual([(self.fa_key,), (self.fa_key, self.fc_key)],
176
def test_annotate_flat_simple(self):
177
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))
185
def test_annotate_flat_merge_and_restored_text(self):
186
self.make_merge_and_restored_text()
187
# 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))
192
def test_annotate_common_merge_text(self):
193
self.make_common_merge_text()
194
# 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))
200
def test_annotate_many_way_common_merge_text(self):
201
self.make_many_way_common_merge_text()
202
self.assertEqual([(self.fa_key, 'simple\n'),
203
(self.fb_key, 'new content\n')],
204
self.ann.annotate_flat(self.ff_key))
206
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
236
def test_needed_keys_simple(self):
237
self.make_simple_text()
238
keys, ann_keys = self.ann._get_needed_keys(self.fb_key)
239
self.assertEqual([self.fa_key, self.fb_key], sorted(keys))
240
self.assertEqual({self.fa_key: 1, self.fb_key: 1},
241
self.ann._num_needed_children)
242
self.assertEqual(set(), ann_keys)
244
def test_needed_keys_many(self):
245
self.make_many_way_common_merge_text()
246
keys, ann_keys = self.ann._get_needed_keys(self.ff_key)
247
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
248
self.fd_key, self.fe_key, self.ff_key,
250
self.assertEqual({self.fa_key: 3,
256
}, self.ann._num_needed_children)
257
self.assertEqual(set(), ann_keys)
259
def test_needed_keys_with_special_text(self):
260
self.make_many_way_common_merge_text()
261
spec_key = ('f-id', revision.CURRENT_REVISION)
262
spec_text = 'simple\nnew content\nlocally modified\n'
263
self.ann.add_special_text(spec_key, [self.fd_key, self.fe_key],
265
keys, ann_keys = self.ann._get_needed_keys(spec_key)
266
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
267
self.fd_key, self.fe_key,
269
self.assertEqual([spec_key], sorted(ann_keys))
271
def test_needed_keys_with_parent_texts(self):
272
self.make_many_way_common_merge_text()
273
# If 'D' and 'E' are already annotated, we don't need to extract all
275
# D | 'simple|new content|'
277
# | E 'simple|new content|'
279
# F-' 'simple|new content|'
280
self.ann._parent_map[self.fd_key] = (self.fb_key, self.fc_key)
281
self.ann._text_cache[self.fd_key] = ['simple\n', 'new content\n']
282
self.ann._annotations_cache[self.fd_key] = [
284
(self.fb_key, self.fc_key),
286
self.ann._parent_map[self.fe_key] = (self.fa_key,)
287
self.ann._text_cache[self.fe_key] = ['simple\n', 'new content\n']
288
self.ann._annotations_cache[self.fe_key] = [
292
keys, ann_keys = self.ann._get_needed_keys(self.ff_key)
293
self.assertEqual([self.ff_key], sorted(keys))
294
self.assertEqual({self.fd_key: 1,
297
}, self.ann._num_needed_children)
298
self.assertEqual([], sorted(ann_keys))
300
def test_record_annotation_removes_texts(self):
301
self.make_many_way_common_merge_text()
302
# Populate the caches
303
for x in self.ann._get_needed_texts(self.ff_key):
305
self.assertEqual({self.fa_key: 3,
311
}, self.ann._num_needed_children)
312
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
313
self.fd_key, self.fe_key, self.ff_key,
314
], sorted(self.ann._text_cache.keys()))
315
self.ann._record_annotation(self.fa_key, [], [])
316
self.ann._record_annotation(self.fb_key, [self.fa_key], [])
317
self.assertEqual({self.fa_key: 2,
323
}, self.ann._num_needed_children)
324
self.assertTrue(self.fa_key in self.ann._text_cache)
325
self.assertTrue(self.fa_key in self.ann._annotations_cache)
326
self.ann._record_annotation(self.fc_key, [self.fa_key], [])
327
self.ann._record_annotation(self.fd_key, [self.fb_key, self.fc_key], [])
328
self.assertEqual({self.fa_key: 1,
334
}, self.ann._num_needed_children)
335
self.assertTrue(self.fa_key in self.ann._text_cache)
336
self.assertTrue(self.fa_key in self.ann._annotations_cache)
337
self.assertFalse(self.fb_key in self.ann._text_cache)
338
self.assertFalse(self.fb_key in self.ann._annotations_cache)
339
self.assertFalse(self.fc_key in self.ann._text_cache)
340
self.assertFalse(self.fc_key in self.ann._annotations_cache)
342
def test_annotate_special_text(self):
343
# Things like WT and PreviewTree want to annotate an arbitrary text
344
# ('current:') so we need a way to add that to the group of files to be
346
self.make_many_way_common_merge_text()
347
# A-. 'simple|content|'
349
# B | | 'simple|new content|'
351
# | C | 'simple|new content|'
353
# D | 'simple|new content|'
355
# | E 'simple|new content|'
357
# SPEC 'simple|new content|locally modified|'
358
spec_key = ('f-id', revision.CURRENT_REVISION)
359
spec_text = 'simple\nnew content\nlocally modified\n'
360
self.ann.add_special_text(spec_key, [self.fd_key, self.fe_key],
362
self.assertAnnotateEqual([(self.fa_key,),
363
(self.fb_key, self.fc_key, self.fe_key),
368
def test_no_graph(self):
369
self.make_no_graph_texts()
370
self.assertAnnotateEqual([(self.fa_key,),
373
self.assertAnnotateEqual([(self.fb_key,),
added
removed
bzrlib/tests/test__annotator.py
