1
# Copyright (C) 2007 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 fetch between repositories of the same type."""
27
from bzrlib.inventory import ROOT_ID
28
from bzrlib.tests import TestSkipped
29
from bzrlib.tests.per_repository import TestCaseWithRepository
30
from bzrlib.transport import get_transport
33
class TestFetchSameRepository(TestCaseWithRepository):
36
# smoke test fetch to ensure that the convenience function works.
37
# it is defined as a convenience function with the underlying
38
# functionality provided by an InterRepository
39
tree_a = self.make_branch_and_tree('a')
40
self.build_tree(['a/foo'])
41
tree_a.add('foo', 'file1')
42
tree_a.commit('rev1', rev_id='rev1')
43
# fetch with a default limit (grab everything)
44
repo = self.make_repository('b')
45
if (tree_a.branch.repository.supports_rich_root() and not
46
repo.supports_rich_root()):
47
raise TestSkipped('Cannot fetch from model2 to model1')
48
repo.fetch(tree_a.branch.repository,
51
def test_fetch_fails_in_write_group(self):
52
# fetch() manages a write group itself, fetching within one isn't safe.
53
repo = self.make_repository('a')
55
self.addCleanup(repo.unlock)
56
repo.start_write_group()
57
self.addCleanup(repo.abort_write_group)
58
# Don't need a specific class - not expecting flow control based on
60
self.assertRaises(errors.BzrError, repo.fetch, repo)
62
def test_fetch_to_knit3(self):
63
# create a repository of the sort we are testing.
64
tree_a = self.make_branch_and_tree('a')
65
self.build_tree(['a/foo'])
66
tree_a.add('foo', 'file1')
67
tree_a.commit('rev1', rev_id='rev1')
68
# create a knit-3 based format to fetch into
69
f = bzrdir.format_registry.make_bzrdir('dirstate-with-subtree')
71
format = tree_a.branch.repository._format
72
format.check_conversion_target(f.repository_format)
73
# if we cannot convert data to knit3, skip the test.
74
except errors.BadConversionTarget, e:
75
raise TestSkipped(str(e))
76
self.get_transport().mkdir('b')
77
b_bzrdir = f.initialize(self.get_url('b'))
78
knit3_repo = b_bzrdir.create_repository()
79
# fetch with a default limit (grab everything)
80
knit3_repo.fetch(tree_a.branch.repository, revision_id=None)
81
# Reopen to avoid any in-memory caching - ensure its reading from
83
knit3_repo = b_bzrdir.open_repository()
84
rev1_tree = knit3_repo.revision_tree('rev1')
87
lines = rev1_tree.get_file_lines(rev1_tree.get_root_id())
90
self.assertEqual([], lines)
91
b_branch = b_bzrdir.create_branch()
92
b_branch.pull(tree_a.branch)
94
tree_b = b_bzrdir.create_workingtree()
95
except errors.NotLocalUrl:
97
tree_b = b_branch.create_checkout('b', lightweight=True)
98
except errors.NotLocalUrl:
99
raise TestSkipped("cannot make working tree with transport %r"
100
% b_bzrdir.transport)
101
tree_b.commit('no change', rev_id='rev2')
102
rev2_tree = knit3_repo.revision_tree('rev2')
103
self.assertEqual('rev1', rev2_tree.inventory.root.revision)
105
def do_test_fetch_to_rich_root_sets_parents_correctly(self, result,
106
snapshots, root_id=ROOT_ID, allow_lefthand_ghost=False):
107
"""Assert that result is the parents of 'tip' after fetching snapshots.
109
This helper constructs a 1.9 format source, and a test-format target
110
and fetches the result of building snapshots in the source, then
111
asserts that the parents of tip are result.
113
:param result: A parents list for the inventories.get_parent_map call.
114
:param snapshots: An iterable of snapshot parameters for
115
BranchBuilder.build_snapshot.
117
# This overlaps slightly with the tests for commit builder about graph
120
repo = self.make_repository('target')
121
remote_format = isinstance(repo, remote.RemoteRepository)
122
if not repo._format.rich_root_data and not remote_format:
123
return # not relevant
124
builder = self.make_branch_builder('source', format='1.9')
125
builder.start_series()
126
for revision_id, parent_ids, actions in snapshots:
127
builder.build_snapshot(revision_id, parent_ids, actions,
128
allow_leftmost_as_ghost=allow_lefthand_ghost)
129
builder.finish_series()
130
source = builder.get_branch()
131
if remote_format and not repo._format.rich_root_data:
132
# use a manual rich root format to ensure the code path is tested.
133
repo = self.make_repository('remote-target',
134
format='1.9-rich-root')
136
self.addCleanup(repo.unlock)
137
repo.fetch(source.repository)
138
self.assertEqual(result,
139
repo.texts.get_parent_map([(root_id, 'tip')])[(root_id, 'tip')])
141
def test_fetch_to_rich_root_set_parent_no_parents(self):
142
# No parents rev -> No parents
143
self.do_test_fetch_to_rich_root_sets_parents_correctly((),
144
[('tip', None, [('add', ('', ROOT_ID, 'directory', ''))]),
147
def test_fetch_to_rich_root_set_parent_1_parent(self):
148
# 1 parent rev -> 1 parent
149
self.do_test_fetch_to_rich_root_sets_parents_correctly(
150
((ROOT_ID, 'base'),),
151
[('base', None, [('add', ('', ROOT_ID, 'directory', ''))]),
155
def test_fetch_to_rich_root_set_parent_1_ghost_parent(self):
156
# 1 ghost parent -> No parents
157
self.do_test_fetch_to_rich_root_sets_parents_correctly((),
158
[('tip', ['ghost'], [('add', ('', ROOT_ID, 'directory', ''))]),
159
], allow_lefthand_ghost=True)
161
def test_fetch_to_rich_root_set_parent_2_head_parents(self):
162
# 2 parents both heads -> 2 parents
163
self.do_test_fetch_to_rich_root_sets_parents_correctly(
164
((ROOT_ID, 'left'), (ROOT_ID, 'right')),
165
[('base', None, [('add', ('', ROOT_ID, 'directory', ''))]),
167
('right', ['base'], []),
168
('tip', ['left', 'right'], []),
171
def test_fetch_to_rich_root_set_parent_2_parents_1_head(self):
172
# 2 parents one head -> 1 parent
173
self.do_test_fetch_to_rich_root_sets_parents_correctly(
174
((ROOT_ID, 'right'),),
175
[('left', None, [('add', ('', ROOT_ID, 'directory', ''))]),
177
('tip', ['left', 'right'], []),
180
def test_fetch_to_rich_root_set_parent_1_parent_different_id_gone(self):
181
# 1 parent different fileid, ours missing -> no parents
182
self.do_test_fetch_to_rich_root_sets_parents_correctly(
184
[('base', None, [('add', ('', ROOT_ID, 'directory', ''))]),
185
('tip', None, [('unversion', ROOT_ID),
186
('add', ('', 'my-root', 'directory', '')),
188
], root_id='my-root')
190
def test_fetch_to_rich_root_set_parent_1_parent_different_id_moved(self):
191
# 1 parent different fileid, ours moved -> 1 parent
192
# (and that parent honours the changing revid of the other location)
193
self.do_test_fetch_to_rich_root_sets_parents_correctly(
194
(('my-root', 'origin'),),
195
[('origin', None, [('add', ('', ROOT_ID, 'directory', '')),
196
('add', ('child', 'my-root', 'directory', ''))]),
198
('tip', None, [('unversion', 'my-root'),
199
('unversion', ROOT_ID),
200
('add', ('', 'my-root', 'directory', '')),
202
], root_id='my-root')
204
def test_fetch_to_rich_root_set_parent_2_parent_1_different_id_gone(self):
205
# 2 parents, 1 different fileid, our second missing -> 1 parent
206
self.do_test_fetch_to_rich_root_sets_parents_correctly(
207
(('my-root', 'right'),),
208
[('base', None, [('add', ('', ROOT_ID, 'directory', ''))]),
209
('right', None, [('unversion', ROOT_ID),
210
('add', ('', 'my-root', 'directory', ''))]),
211
('tip', ['base', 'right'], [('unversion', ROOT_ID),
212
('add', ('', 'my-root', 'directory', '')),
214
], root_id='my-root')
216
def test_fetch_to_rich_root_set_parent_2_parent_2_different_id_moved(self):
217
# 2 parents, 1 different fileid, our second moved -> 2 parent
218
# (and that parent honours the changing revid of the other location)
219
self.do_test_fetch_to_rich_root_sets_parents_correctly(
220
(('my-root', 'right'),),
221
# 'my-root' at 'child'.
222
[('origin', None, [('add', ('', ROOT_ID, 'directory', '')),
223
('add', ('child', 'my-root', 'directory', ''))]),
226
('right', None, [('unversion', 'my-root'),
227
('unversion', ROOT_ID),
228
('add', ('', 'my-root', 'directory', ''))]),
229
('tip', ['base', 'right'], [('unversion', 'my-root'),
230
('unversion', ROOT_ID),
231
('add', ('', 'my-root', 'directory', '')),
233
], root_id='my-root')
235
def test_fetch_all_from_self(self):
236
tree = self.make_branch_and_tree('.')
237
rev_id = tree.commit('one')
238
# This needs to be a new copy of the repository, if this changes, the
239
# test needs to be rewritten
240
repo = tree.branch.repository.bzrdir.open_repository()
241
# This fetch should be a no-op see bug #158333
242
tree.branch.repository.fetch(repo, None)
244
def test_fetch_from_self(self):
245
tree = self.make_branch_and_tree('.')
246
rev_id = tree.commit('one')
247
repo = tree.branch.repository.bzrdir.open_repository()
248
# This fetch should be a no-op see bug #158333
249
tree.branch.repository.fetch(repo, rev_id)
251
def test_fetch_missing_from_self(self):
252
tree = self.make_branch_and_tree('.')
253
rev_id = tree.commit('one')
254
# Even though the fetch() is a NO-OP it should assert the revision id
256
repo = tree.branch.repository.bzrdir.open_repository()
257
self.assertRaises(errors.NoSuchRevision, tree.branch.repository.fetch,
258
repo, 'no-such-revision')
260
def makeARepoWithSignatures(self):
261
wt = self.make_branch_and_tree('a-repo-with-sigs')
262
wt.commit('rev1', allow_pointless=True, rev_id='rev1')
263
repo = wt.branch.repository
265
repo.start_write_group()
266
repo.sign_revision('rev1', gpg.LoopbackGPGStrategy(None))
267
repo.commit_write_group()
271
def test_fetch_copies_signatures(self):
272
source_repo = self.makeARepoWithSignatures()
273
target_repo = self.make_repository('target')
274
target_repo.fetch(source_repo, revision_id=None)
276
source_repo.get_signature_text('rev1'),
277
target_repo.get_signature_text('rev1'))
279
def make_repository_with_one_revision(self):
280
wt = self.make_branch_and_tree('source')
281
wt.commit('rev1', allow_pointless=True, rev_id='rev1')
282
return wt.branch.repository
284
def test_fetch_revision_already_exists(self):
285
# Make a repository with one revision.
286
source_repo = self.make_repository_with_one_revision()
287
# Fetch that revision into a second repository.
288
target_repo = self.make_repository('target')
289
target_repo.fetch(source_repo, revision_id='rev1')
290
# Now fetch again; there will be nothing to do. This should work
291
# without causing any errors.
292
target_repo.fetch(source_repo, revision_id='rev1')
294
def test_fetch_all_same_revisions_twice(self):
295
# Blind-fetching all the same revisions twice should succeed and be a
296
# no-op the second time.
297
repo = self.make_repository('repo')
298
tree = self.make_branch_and_tree('tree')
299
revision_id = tree.commit('test')
300
repo.fetch(tree.branch.repository)
301
repo.fetch(tree.branch.repository)
304
class TestSource(TestCaseWithRepository):
305
"""Tests for/about the results of Repository._get_source."""
307
def test_no_absent_records_in_stream_with_ghosts(self):
308
# XXX: Arguably should be in interrepository_implementations but
309
# doesn't actually gain coverage there; need a specific set of
310
# permutations to cover it.
311
# bug lp:376255 was reported about this.
312
builder = self.make_branch_builder('repo')
313
builder.start_series()
314
builder.build_snapshot('tip', ['ghost'],
315
[('add', ('', 'ROOT_ID', 'directory', ''))],
316
allow_leftmost_as_ghost=True)
317
builder.finish_series()
318
b = builder.get_branch()
320
self.addCleanup(b.unlock)
322
source = repo._get_source(repo._format)
323
search = graph.PendingAncestryResult(['tip'], repo)
324
stream = source.get_stream(search)
325
for substream_type, substream in stream:
326
for record in substream:
327
self.assertNotEqual('absent', record.storage_kind,
328
"Absent record for %s" % (((substream_type,) + record.key),))
added
removed
bzrlib/tests/per_repository/test_fetch.py
