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 = (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')
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, (), [b'simple\n', b'content\n'])
54
self.vf.add_lines(self.fb_key, (), [b'simple\n', b'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, [], [b'simple\n', b'content\n'])
68
self.vf.add_lines(self.fb_key, [self.fa_key],
69
[b'simple\n', b'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
[b'simple\n', b'from c\n', b'content\n'])
82
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
83
[b'simple\n', b'from c\n', b'new content\n',
84
b'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
[b'simple\n', b'new content\n'])
98
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
99
[b'simple\n', b'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
[b'simple\n', b'new content\n'])
116
self.vf.add_lines(self.fd_key, [self.fb_key, self.fc_key],
117
[b'simple\n', b'new content\n'])
118
self.vf.add_lines(self.fe_key, [self.fa_key],
119
[b'simple\n', b'new content\n'])
120
self.vf.add_lines(self.ff_key, [self.fd_key, self.fe_key],
121
[b'simple\n', b'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
[b'simple\n', b'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
[b'simple\n', b'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, b''.join(lines))
147
def test_annotate_missing(self):
148
self.make_simple_text()
149
self.assertRaises(errors.RevisionNotPresent,
150
self.ann.annotate, (b'not', b'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, 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))
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, b'simple\n'),
192
(self.fc_key, b'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'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))
203
def test_annotate_many_way_common_merge_text(self):
204
self.make_many_way_common_merge_text()
205
self.assertEqual([(self.fa_key, b'simple\n'),
206
(self.fb_key, b'new content\n')],
207
self.ann.annotate_flat(self.ff_key))
209
def test_annotate_flat_respects_break_ann_tie(self):
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)
235
def test_needed_keys_simple(self):
236
self.make_simple_text()
237
keys, ann_keys = self.ann._get_needed_keys(self.fb_key)
238
self.assertEqual([self.fa_key, self.fb_key], sorted(keys))
239
self.assertEqual({self.fa_key: 1, self.fb_key: 1},
240
self.ann._num_needed_children)
241
self.assertEqual(set(), ann_keys)
243
def test_needed_keys_many(self):
244
self.make_many_way_common_merge_text()
245
keys, ann_keys = self.ann._get_needed_keys(self.ff_key)
246
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
247
self.fd_key, self.fe_key, self.ff_key,
249
self.assertEqual({self.fa_key: 3,
255
}, self.ann._num_needed_children)
256
self.assertEqual(set(), ann_keys)
258
def test_needed_keys_with_special_text(self):
259
self.make_many_way_common_merge_text()
260
spec_key = (b'f-id', revision.CURRENT_REVISION)
261
spec_text = b'simple\nnew content\nlocally modified\n'
262
self.ann.add_special_text(spec_key, [self.fd_key, self.fe_key],
264
keys, ann_keys = self.ann._get_needed_keys(spec_key)
265
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
266
self.fd_key, self.fe_key,
268
self.assertEqual([spec_key], sorted(ann_keys))
270
def test_needed_keys_with_parent_texts(self):
271
self.make_many_way_common_merge_text()
272
# If 'D' and 'E' are already annotated, we don't need to extract all
274
# D | 'simple|new content|'
276
# | E 'simple|new content|'
278
# F-' 'simple|new content|'
279
self.ann._parent_map[self.fd_key] = (self.fb_key, self.fc_key)
280
self.ann._text_cache[self.fd_key] = [b'simple\n', b'new content\n']
281
self.ann._annotations_cache[self.fd_key] = [
283
(self.fb_key, self.fc_key),
285
self.ann._parent_map[self.fe_key] = (self.fa_key,)
286
self.ann._text_cache[self.fe_key] = [b'simple\n', b'new content\n']
287
self.ann._annotations_cache[self.fe_key] = [
291
keys, ann_keys = self.ann._get_needed_keys(self.ff_key)
292
self.assertEqual([self.ff_key], sorted(keys))
293
self.assertEqual({self.fd_key: 1,
296
}, self.ann._num_needed_children)
297
self.assertEqual([], sorted(ann_keys))
299
def test_record_annotation_removes_texts(self):
300
self.make_many_way_common_merge_text()
301
# Populate the caches
302
for x in self.ann._get_needed_texts(self.ff_key):
304
self.assertEqual({self.fa_key: 3,
310
}, self.ann._num_needed_children)
311
self.assertEqual([self.fa_key, self.fb_key, self.fc_key,
312
self.fd_key, self.fe_key, self.ff_key,
313
], sorted(self.ann._text_cache.keys()))
314
self.ann._record_annotation(self.fa_key, [], [])
315
self.ann._record_annotation(self.fb_key, [self.fa_key], [])
316
self.assertEqual({self.fa_key: 2,
322
}, self.ann._num_needed_children)
323
self.assertTrue(self.fa_key in self.ann._text_cache)
324
self.assertTrue(self.fa_key in self.ann._annotations_cache)
325
self.ann._record_annotation(self.fc_key, [self.fa_key], [])
326
self.ann._record_annotation(
327
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 = (b'f-id', revision.CURRENT_REVISION)
359
spec_text = b'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,),