1
# Copyright (C) 2005 by 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
17
from bzrlib.selftest.testrevision import make_branches
18
from bzrlib.trace import mutter, note
19
from bzrlib.branch import Branch
20
from bzrlib.progress import ProgressBar
24
def greedy_fetch(to_branch, from_branch, revision=None, pb=None):
25
"""Copy a revision and all available ancestors from one branch to another
26
If no revision is specified, uses the last revision in the source branch's
29
from_history = from_branch.revision_history()
30
required_revisions = set(from_history)
32
if revision is not None:
33
required_revisions.add(revision)
35
rev_index = from_history.index(revision)
38
if rev_index is not None:
39
from_history = from_history[:rev_index + 1]
41
from_history = [revision]
42
to_history = to_branch.revision_history()
44
for rev_id in from_history:
45
if not has_revision(to_branch, rev_id):
46
missing.append(rev_id)
49
while len(missing) > 0:
50
installed, failed = to_branch.install_revisions(from_branch,
54
required_failed = failed.intersection(required_revisions)
55
if len(required_failed) > 0:
56
raise bzrlib.errors.InstallFailed(required_failed)
58
note("Failed to install %s" % rev_id)
59
all_failed.update(failed)
61
for rev_id in missing:
63
revision = from_branch.get_revision(rev_id)
64
except bzrlib.errors.NoSuchRevision:
65
if revision in from_history:
69
for parent in [p.revision_id for p in revision.parents]:
70
if not has_revision(to_branch, parent):
71
new_missing.append(parent)
73
return count, all_failed
76
from testsweet import InTempDir
77
from bzrlib.commit import commit
78
def has_revision(branch, revision_id):
80
branch.get_revision_xml(revision_id)
82
except bzrlib.errors.NoSuchRevision:
85
class TestFetch(InTempDir):
89
return Branch(name, init=True)
91
#highest indices a: 5, b: 7
92
br_a, br_b = make_branches()
93
assert not has_revision(br_b, br_a.revision_history()[3])
94
assert has_revision(br_b, br_a.revision_history()[2])
95
assert len(br_b.revision_history()) == 7
96
assert greedy_fetch(br_b, br_a, br_a.revision_history()[2])[0] == 0
98
# greedy_fetch is not supposed to alter the revision history
99
assert len(br_b.revision_history()) == 7
100
assert not has_revision(br_b, br_a.revision_history()[3])
102
assert len(br_b.revision_history()) == 7
103
assert greedy_fetch(br_b, br_a, br_a.revision_history()[3])[0] == 1
104
assert has_revision(br_b, br_a.revision_history()[3])
105
assert not has_revision(br_a, br_b.revision_history()[3])
106
assert not has_revision(br_a, br_b.revision_history()[4])
108
# When a non-branch ancestor is missing, it should be a failure, not
110
br_a4 = new_branch('br_a4')
111
count, failures = greedy_fetch(br_a4, br_a)
113
assert failures == set((br_b.revision_history()[4],
114
br_b.revision_history()[5]))
116
assert greedy_fetch(br_a, br_b)[0] == 4
117
assert has_revision(br_a, br_b.revision_history()[3])
118
assert has_revision(br_a, br_b.revision_history()[4])
120
br_b2 = new_branch('br_b2')
121
assert greedy_fetch(br_b2, br_b)[0] == 7
122
assert has_revision(br_b2, br_b.revision_history()[4])
123
assert has_revision(br_b2, br_a.revision_history()[2])
124
assert not has_revision(br_b2, br_a.revision_history()[3])
126
br_a2 = new_branch('br_a2')
127
assert greedy_fetch(br_a2, br_a)[0] == 9
128
assert has_revision(br_a2, br_b.revision_history()[4])
129
assert has_revision(br_a2, br_a.revision_history()[3])
131
br_a3 = new_branch('br_a3')
132
assert greedy_fetch(br_a3, br_a2)[0] == 0
133
for revno in range(4):
134
assert not has_revision(br_a3, br_a.revision_history()[revno])
135
assert greedy_fetch(br_a3, br_a2, br_a.revision_history()[2])[0] == 3
136
fetched = greedy_fetch(br_a3, br_a2, br_a.revision_history()[3])[0]
137
assert fetched == 3, "fetched %d instead of 3" % fetched
138
# InstallFailed should be raised if the branch is missing the revision
139
# that was requested.
140
self.assertRaises(bzrlib.errors.InstallFailed, greedy_fetch, br_a3,
142
# InstallFailed should be raised if the branch is missing a revision
143
# from its own revision history
144
br_a2.append_revision('a-b-c')
145
self.assertRaises(bzrlib.errors.InstallFailed, greedy_fetch, br_a3,
150
if __name__ == '__main__':
152
sys.exit(run_suite(unittest.makeSuite()))
added
removed
bzrlib/fetch.py
