1
# Copyright (C) 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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18
"""Direct tests of the loggerhead/history.py module"""
20
from datetime import datetime
21
from bzrlib import tag, tests
23
from loggerhead import history as _mod_history
26
class TestCaseWithExamples(tests.TestCaseWithMemoryTransport):
28
def make_linear_ancestry(self):
35
builder = self.make_branch_builder('branch')
36
builder.start_series()
37
builder.build_snapshot('rev-1', None, [
38
('add', ('', 'root-id', 'directory', None))])
39
builder.build_snapshot('rev-2', ['rev-1'], [])
40
builder.build_snapshot('rev-3', ['rev-2'], [])
41
builder.finish_series()
42
b = builder.get_branch()
43
self.addCleanup(b.lock_read().unlock)
44
return _mod_history.History(b, {})
46
def make_long_linear_ancestry(self):
47
builder = self.make_branch_builder('branch')
48
builder.start_series()
49
builder.build_snapshot('A', None, [
50
('add', ('', 'root-id', 'directory', None))])
51
for r in "BCDEFGHIJKLMNOPQRSTUVWXYZ":
52
builder.build_snapshot(r, None, [])
53
builder.finish_series()
54
b = builder.get_branch()
55
self.addCleanup(b.lock_read().unlock)
56
return _mod_history.History(b, {})
58
def make_merged_ancestry(self):
65
builder = self.make_branch_builder('branch')
66
builder.start_series()
67
builder.build_snapshot('rev-1', None, [
68
('add', ('', 'root-id', 'directory', None))])
69
builder.build_snapshot('rev-2', ['rev-1'], [])
70
builder.build_snapshot('rev-3', ['rev-1', 'rev-2'], [])
71
builder.finish_series()
72
b = builder.get_branch()
73
self.addCleanup(b.lock_read().unlock)
74
return _mod_history.History(b, {})
76
def make_deep_merged_ancestry(self):
87
builder = self.make_branch_builder('branch')
88
builder.start_series()
89
builder.build_snapshot('A', None, [
90
('add', ('', 'root-id', 'directory', None))])
91
builder.build_snapshot('B', ['A'], [])
92
builder.build_snapshot('C', ['A'], [])
93
builder.build_snapshot('D', ['C'], [])
94
builder.build_snapshot('E', ['C', 'D'], [])
95
builder.build_snapshot('F', ['B', 'E'], [])
96
builder.finish_series()
97
b = builder.get_branch()
98
self.addCleanup(b.lock_read().unlock)
99
return _mod_history.History(b, {})
101
def assertRevidsFrom(self, expected, history, search_revs, tip_rev):
102
self.assertEqual(expected,
103
list(history.get_revids_from(search_revs, tip_rev)))
106
class _DictProxy(object):
108
def __init__(self, d):
110
self._accessed = set()
111
self.__setitem__ = d.__setitem__
113
def __getitem__(self, name):
114
self._accessed.add(name)
121
def track_rev_info_accesses(h):
122
"""Track __getitem__ access to History._rev_info,
124
:return: set of items accessed
126
h._rev_info = _DictProxy(h._rev_info)
127
return h._rev_info._accessed
130
class TestHistoryGetRevidsFrom(TestCaseWithExamples):
132
def test_get_revids_from_simple_mainline(self):
133
history = self.make_linear_ancestry()
134
self.assertRevidsFrom(['rev-3', 'rev-2', 'rev-1'],
135
history, None, 'rev-3')
137
def test_get_revids_from_merged_mainline(self):
138
history = self.make_merged_ancestry()
139
self.assertRevidsFrom(['rev-3', 'rev-1'],
140
history, None, 'rev-3')
142
def test_get_revids_given_one_rev(self):
143
history = self.make_merged_ancestry()
144
# rev-3 was the first mainline revision to see rev-2.
145
self.assertRevidsFrom(['rev-3'], history, ['rev-2'], 'rev-3')
147
def test_get_revids_deep_ancestry(self):
148
history = self.make_deep_merged_ancestry()
149
self.assertRevidsFrom(['F'], history, ['F'], 'F')
150
self.assertRevidsFrom(['F'], history, ['E'], 'F')
151
self.assertRevidsFrom(['F'], history, ['D'], 'F')
152
self.assertRevidsFrom(['F'], history, ['C'], 'F')
153
self.assertRevidsFrom(['B'], history, ['B'], 'F')
154
self.assertRevidsFrom(['A'], history, ['A'], 'F')
156
def test_get_revids_doesnt_over_produce_simple_mainline(self):
157
# get_revids_from shouldn't walk the whole ancestry just to get the
158
# answers for the first few revisions.
159
history = self.make_long_linear_ancestry()
160
accessed = track_rev_info_accesses(history)
161
result = history.get_revids_from(None, 'Z')
162
self.assertEqual(set(), accessed)
163
self.assertEqual('Z', result.next())
164
# We already know 'Z' because we passed it in.
165
self.assertEqual(set(), accessed)
166
self.assertEqual('Y', result.next())
167
self.assertEqual(set([history._rev_indices['Z']]), accessed)
169
def test_get_revids_doesnt_over_produce_for_merges(self):
170
# get_revids_from shouldn't walk the whole ancestry just to get the
171
# answers for the first few revisions.
172
history = self.make_long_linear_ancestry()
173
accessed = track_rev_info_accesses(history)
174
result = history.get_revids_from(['X', 'V'], 'Z')
175
self.assertEqual(set(), accessed)
176
self.assertEqual('X', result.next())
177
# We access 'W' because we are checking that W wasn't merged into X.
178
# The important bit is that we aren't getting the whole ancestry.
179
self.assertEqual(set([history._rev_indices[x] for x in 'ZYXW']),
181
self.assertEqual('V', result.next())
182
self.assertEqual(set([history._rev_indices[x] for x in 'ZYXWVU']),
184
self.assertRaises(StopIteration, result.next)
185
self.assertEqual(set([history._rev_indices[x] for x in 'ZYXWVU']),
190
class TestHistoryChangeFromRevision(tests.TestCaseWithTransport):
192
def make_single_commit(self):
193
tree = self.make_branch_and_tree('test')
194
rev_id = tree.commit('Commit Message', timestamp=1299838474.317,
195
timezone=3600, committer='Joe Example <joe@example.com>',
197
self.addCleanup(tree.branch.lock_write().unlock)
198
rev = tree.branch.repository.get_revision(rev_id)
199
history = _mod_history.History(tree.branch, {})
202
def test_simple(self):
203
history, rev = self.make_single_commit()
204
change = history._change_from_revision(rev)
205
self.assertEqual(rev.revision_id, change.revid)
206
self.assertEqual(datetime.fromtimestamp(1299838474.317),
208
self.assertEqual(datetime.utcfromtimestamp(1299838474.317),
210
self.assertEqual(['Joe Example <joe@example.com>'],
212
self.assertEqual('test', change.branch_nick)
213
self.assertEqual('Commit Message', change.short_comment)
214
self.assertEqual('Commit Message', change.comment)
215
self.assertEqual(['Commit Message'], change.comment_clean)
216
self.assertEqual([], change.parents)
217
self.assertEqual([], change.bugs)
218
self.assertEqual(None, change.tags)
221
history, rev = self.make_single_commit()
223
b.tags.set_tag('tag-1', rev.revision_id)
224
b.tags.set_tag('tag-2', rev.revision_id)
225
b.tags.set_tag('Tag-10', rev.revision_id)
226
change = history._change_from_revision(rev)
227
# If available, tags are 'naturally' sorted. (sorting numbers in order,
228
# and ignoring case, etc.)
229
if getattr(tag, 'sort_natural', None) is not None:
230
self.assertEqual('tag-1, tag-2, Tag-10', change.tags)
232
self.assertEqual('Tag-10, tag-1, tag-2', change.tags)
234
def test_committer_vs_authors(self):
235
tree = self.make_branch_and_tree('test')
236
rev_id = tree.commit('Commit Message', timestamp=1299838474.317,
237
timezone=3600, committer='Joe Example <joe@example.com>',
238
revprops={'authors': u'A Author <aauthor@example.com>\n'
239
u'B Author <bauthor@example.com>'})
240
self.addCleanup(tree.branch.lock_write().unlock)
241
rev = tree.branch.repository.get_revision(rev_id)
242
history = _mod_history.History(tree.branch, {})
243
change = history._change_from_revision(rev)
244
self.assertEqual(u'Joe Example <joe@example.com>',
246
self.assertEqual([u'A Author <aauthor@example.com>',
247
u'B Author <bauthor@example.com>'],
251
class TestHistory_IterateSufficiently(tests.TestCase):
253
def assertIterate(self, expected, iterable, stop_at, extra_rev_count):
254
self.assertEqual(expected, _mod_history.History._iterate_sufficiently(
255
iterable, stop_at, extra_rev_count))
257
def test_iter_no_extra(self):
258
lst = list('abcdefghijklmnopqrstuvwxyz')
259
self.assertIterate(['a', 'b', 'c'], iter(lst), 'c', 0)
260
self.assertIterate(['a', 'b', 'c', 'd'], iter(lst), 'd', 0)
262
def test_iter_not_found(self):
263
# If the key in question isn't found, we just exhaust the list
264
lst = list('abcdefghijklmnopqrstuvwxyz')
265
self.assertIterate(lst, iter(lst), 'not-there', 0)
267
def test_iter_with_extra(self):
268
lst = list('abcdefghijklmnopqrstuvwxyz')
269
self.assertIterate(['a', 'b', 'c'], iter(lst), 'b', 1)
270
self.assertIterate(['a', 'b', 'c', 'd', 'e'], iter(lst), 'c', 2)
272
def test_iter_with_too_many_extra(self):
273
lst = list('abcdefghijklmnopqrstuvwxyz')
274
self.assertIterate(lst, iter(lst), 'y', 10)
275
self.assertIterate(lst, iter(lst), 'z', 10)
277
def test_iter_with_extra_None(self):
278
lst = list('abcdefghijklmnopqrstuvwxyz')
279
self.assertIterate(lst, iter(lst), 'z', None)
283
class TestHistoryGetView(TestCaseWithExamples):
285
def test_get_view_limited_history(self):
286
# get_view should only load enough history to serve the result, not all
288
history = self.make_long_linear_ancestry()
289
accessed = track_rev_info_accesses(history)
290
revid, start_revid, revid_list = history.get_view('Z', 'Z', None,
292
self.assertEqual(['Z', 'Y', 'X', 'W', 'V', 'U'], revid_list)
293
self.assertEqual('Z', revid)
294
self.assertEqual('Z', start_revid)
295
self.assertEqual(set([history._rev_indices[x] for x in 'ZYXWVU']),
added
removed
loggerhead/tests/test_history.py
