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."""
31
def load_tests(loader, standard_tests, pattern):
32
"""Parameterize tests for all versions of groupcompress."""
33
suite, _ = tests.permute_tests_for_extension(standard_tests, loader,
34
'breezy._annotator_py', 'breezy._annotator_pyx')
38
class TestAnnotator(tests.TestCaseWithMemoryTransport):
40
module = None # Set by load_tests
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
def make_no_graph_texts(self):
50
factory = knit.make_pack_factory(False, False, 2)
51
self.vf = factory(self.get_transport())
52
self.ann = self.module.Annotator(self.vf)
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
def make_simple_text(self):
57
# TODO: all we really need is a VersionedFile instance, we'd like to
58
# avoid creating all the intermediate stuff
59
factory = knit.make_pack_factory(True, True, 2)
60
self.vf = factory(self.get_transport())
61
# This assumes nothing special happens during __init__, which may be
63
self.ann = self.module.Annotator(self.vf)
66
# B 'simple|new content|'
67
self.vf.add_lines(self.fa_key, [], ['simple\n', 'content\n'])
68
self.vf.add_lines(self.fb_key, [self.fa_key],
69
['simple\n', 'new content\n'])
71
def make_merge_text(self):
72
self.make_simple_text()
75
# B | 'simple|new content|'
77
# | C 'simple|from c|content|'
79
# D 'simple|from c|new content|introduced in merge|'
80
self.vf.add_lines(self.fc_key, [self.fa_key],
81
['simple\n', 'from c\n', 'content\n'])
82
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
83
['simple\n', 'from c\n', 'new content\n',
84
'introduced in merge\n'])
86
def make_common_merge_text(self):
87
"""Both sides of the merge will have introduced a line."""
88
self.make_simple_text()
91
# B | 'simple|new content|'
93
# | C 'simple|new content|'
95
# D 'simple|new content|'
96
self.vf.add_lines(self.fc_key, [self.fa_key],
97
['simple\n', 'new content\n'])
98
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
99
['simple\n', 'new content\n'])
101
def make_many_way_common_merge_text(self):
102
self.make_simple_text()
103
# A-. 'simple|content|'
105
# B | | 'simple|new content|'
107
# | C | 'simple|new content|'
109
# D | 'simple|new content|'
111
# | E 'simple|new content|'
113
# F-' 'simple|new content|'
114
self.vf.add_lines(self.fc_key, [self.fa_key],
115
['simple\n', 'new content\n'])
116
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
117
['simple\n', 'new content\n'])
118
self.vf.add_lines(self.fe_key, [self.fa_key],
119
['simple\n', 'new content\n'])
120
self.vf.add_lines(self.ff_key, [self.fd_key, self.fe_key],
121
['simple\n', 'new content\n'])
123
def make_merge_and_restored_text(self):
124
self.make_simple_text()
125
# A 'simple|content|'
127
# B | 'simple|new content|'
129
# C | 'simple|content|' # reverted to A
131
# D 'simple|content|'
132
# c reverts back to 'a' for the new content line
133
self.vf.add_lines(self.fc_key, [self.fb_key],
134
['simple\n', 'content\n'])
135
# d merges 'a' and 'c', to find both claim last modified
136
self.vf.add_lines(self.fd_key, [self.fa_key, self.fc_key],
137
['simple\n', 'content\n'])
139
def assertAnnotateEqual(self, expected_annotation, key, exp_text=None):
140
annotation, lines = self.ann.annotate(key)
141
self.assertEqual(expected_annotation, annotation)
143
record = next(self.vf.get_record_stream([key], 'unordered', True))
144
exp_text = record.get_bytes_as('fulltext')
145
self.assertEqualDiff(exp_text, ''.join(lines))
147
def test_annotate_missing(self):
148
self.make_simple_text()
149
self.assertRaises(errors.RevisionNotPresent,
150
self.ann.annotate, ('not', 'present'))
152
def test_annotate_simple(self):
153
self.make_simple_text()
154
self.assertAnnotateEqual([(self.fa_key,)]*2, self.fa_key)
155
self.assertAnnotateEqual([(self.fa_key,), (self.fb_key,)], self.fb_key)
157
def test_annotate_merge_text(self):
158
self.make_merge_text()
159
self.assertAnnotateEqual([(self.fa_key,), (self.fc_key,),
160
(self.fb_key,), (self.fd_key,)],
163
def test_annotate_common_merge_text(self):
164
self.make_common_merge_text()
165
self.assertAnnotateEqual([(self.fa_key,), (self.fb_key, self.fc_key)],
168
def test_annotate_many_way_common_merge_text(self):
169
self.make_many_way_common_merge_text()
170
self.assertAnnotateEqual([(self.fa_key,),
171
(self.fb_key, self.fc_key, self.fe_key)],
174
def test_annotate_merge_and_restored(self):
175
self.make_merge_and_restored_text()
176
self.assertAnnotateEqual([(self.fa_key,), (self.fa_key, self.fc_key)],
179
def test_annotate_flat_simple(self):
180
self.make_simple_text()
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
def test_annotate_flat_merge_and_restored_text(self):
189
self.make_merge_and_restored_text()
190
# fc is a simple dominator of fa
191
self.assertEqual([(self.fa_key, 'simple\n'),
192
(self.fc_key, 'content\n'),
193
], self.ann.annotate_flat(self.fd_key))
195
def test_annotate_common_merge_text(self):
196
self.make_common_merge_text()
197
# there is no common point, so we just pick the lexicographical lowest
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
def test_annotate_many_way_common_merge_text(self):
204
self.make_many_way_common_merge_text()
205
self.assertEqual([(self.fa_key, 'simple\n'),
206
(self.fb_key, 'new content\n')],
207
self.ann.annotate_flat(self.ff_key))
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
239
def test_needed_keys_simple(self):
240
self.make_simple_text()
241
keys, ann_keys = self.ann._get_needed_keys(self.fb_key)
242
self.assertEqual([self.fa_key, self.fb_key], sorted(keys))
243
self.assertEqual({self.fa_key: 1, self.fb_key: 1},
244
self.ann._num_needed_children)
245
self.assertEqual(set(), ann_keys)
247
def test_needed_keys_many(self):
248
self.make_many_way_common_merge_text()
249
keys, ann_keys = self.ann._get_needed_keys(self.ff_key)
250
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
251
self.fd_key, self.fe_key, self.ff_key,
253
self.assertEqual({self.fa_key: 3,
259
}, self.ann._num_needed_children)
260
self.assertEqual(set(), ann_keys)
262
def test_needed_keys_with_special_text(self):
263
self.make_many_way_common_merge_text()
264
spec_key = ('f-id', revision.CURRENT_REVISION)
265
spec_text = 'simple\nnew content\nlocally modified\n'
266
self.ann.add_special_text(spec_key, [self.fd_key, self.fe_key],
268
keys, ann_keys = self.ann._get_needed_keys(spec_key)
269
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
270
self.fd_key, self.fe_key,
272
self.assertEqual([spec_key], sorted(ann_keys))
274
def test_needed_keys_with_parent_texts(self):
275
self.make_many_way_common_merge_text()
276
# If 'D' and 'E' are already annotated, we don't need to extract all
278
# D | 'simple|new content|'
280
# | E 'simple|new content|'
282
# F-' 'simple|new content|'
283
self.ann._parent_map[self.fd_key] = (self.fb_key, self.fc_key)
284
self.ann._text_cache[self.fd_key] = ['simple\n', 'new content\n']
285
self.ann._annotations_cache[self.fd_key] = [
287
(self.fb_key, self.fc_key),
289
self.ann._parent_map[self.fe_key] = (self.fa_key,)
290
self.ann._text_cache[self.fe_key] = ['simple\n', 'new content\n']
291
self.ann._annotations_cache[self.fe_key] = [
295
keys, ann_keys = self.ann._get_needed_keys(self.ff_key)
296
self.assertEqual([self.ff_key], sorted(keys))
297
self.assertEqual({self.fd_key: 1,
300
}, self.ann._num_needed_children)
301
self.assertEqual([], sorted(ann_keys))
303
def test_record_annotation_removes_texts(self):
304
self.make_many_way_common_merge_text()
305
# Populate the caches
306
for x in self.ann._get_needed_texts(self.ff_key):
308
self.assertEqual({self.fa_key: 3,
314
}, self.ann._num_needed_children)
315
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
316
self.fd_key, self.fe_key, self.ff_key,
317
], sorted(self.ann._text_cache.keys()))
318
self.ann._record_annotation(self.fa_key, [], [])
319
self.ann._record_annotation(self.fb_key, [self.fa_key], [])
320
self.assertEqual({self.fa_key: 2,
326
}, self.ann._num_needed_children)
327
self.assertTrue(self.fa_key in self.ann._text_cache)
328
self.assertTrue(self.fa_key in self.ann._annotations_cache)
329
self.ann._record_annotation(self.fc_key, [self.fa_key], [])
330
self.ann._record_annotation(self.fd_key, [self.fb_key, self.fc_key], [])
331
self.assertEqual({self.fa_key: 1,
337
}, self.ann._num_needed_children)
338
self.assertTrue(self.fa_key in self.ann._text_cache)
339
self.assertTrue(self.fa_key in self.ann._annotations_cache)
340
self.assertFalse(self.fb_key in self.ann._text_cache)
341
self.assertFalse(self.fb_key in self.ann._annotations_cache)
342
self.assertFalse(self.fc_key in self.ann._text_cache)
343
self.assertFalse(self.fc_key in self.ann._annotations_cache)
345
def test_annotate_special_text(self):
346
# Things like WT and PreviewTree want to annotate an arbitrary text
347
# ('current:') so we need a way to add that to the group of files to be
349
self.make_many_way_common_merge_text()
350
# A-. 'simple|content|'
352
# B | | 'simple|new content|'
354
# | C | 'simple|new content|'
356
# D | 'simple|new content|'
358
# | E 'simple|new content|'
360
# SPEC 'simple|new content|locally modified|'
361
spec_key = ('f-id', revision.CURRENT_REVISION)
362
spec_text = 'simple\nnew content\nlocally modified\n'
363
self.ann.add_special_text(spec_key, [self.fd_key, self.fe_key],
365
self.assertAnnotateEqual([(self.fa_key,),
366
(self.fb_key, self.fc_key, self.fe_key),
371
def test_no_graph(self):
372
self.make_no_graph_texts()
373
self.assertAnnotateEqual([(self.fa_key,),
376
self.assertAnnotateEqual([(self.fb_key,),