bzr branch
http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar
| 
4044.1.1
by Robert Collins
 Create Branch.create_clone_on_transport helper method to combine bzr and branch creation for push.  | 
1  | 
# Copyright (C) 2009 Canonical Ltd
 | 
2  | 
#
 | 
|
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.
 | 
|
7  | 
#
 | 
|
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.
 | 
|
12  | 
#
 | 
|
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
 | 
|
16  | 
||
17  | 
"""Tests for branch.create_clone behaviour."""
 | 
|
18  | 
||
| 
4050.1.1
by Robert Collins
 Fix race condition with branch hooks during cloning when the new branch is stacked.  | 
19  | 
from bzrlib.branch import Branch  | 
| 
4053.2.2
by Andrew Bennetts
 Better fix, with test.  | 
20  | 
from bzrlib import errors  | 
| 
4053.2.5
by Andrew Bennetts
 Merge from bzr.dev.  | 
21  | 
from bzrlib import remote  | 
| 
4053.2.2
by Andrew Bennetts
 Better fix, with test.  | 
22  | 
from bzrlib import tests  | 
| 
4044.1.1
by Robert Collins
 Create Branch.create_clone_on_transport helper method to combine bzr and branch creation for push.  | 
23  | 
from bzrlib.tests.branch_implementations.test_branch import TestCaseWithBranch  | 
24  | 
||
25  | 
||
26  | 
class TestCreateClone(TestCaseWithBranch):  | 
|
27  | 
||
28  | 
def test_create_clone_on_transport_no_revision_id(self):  | 
|
29  | 
tree = self.make_branch_and_tree('source')  | 
|
30  | 
tree.commit('a commit')  | 
|
31  | 
source = tree.branch  | 
|
32  | 
target_transport = self.get_transport('target')  | 
|
33  | 
result = tree.branch.create_clone_on_transport(target_transport)  | 
|
34  | 
self.assertEqual(source.last_revision(), result.last_revision())  | 
|
35  | 
||
36  | 
def test_create_clone_on_transport_revision_id(self):  | 
|
37  | 
tree = self.make_branch_and_tree('source')  | 
|
38  | 
old_revid = tree.commit('a commit')  | 
|
39  | 
source_tip = tree.commit('a second commit')  | 
|
40  | 
source = tree.branch  | 
|
41  | 
target_transport = self.get_transport('target')  | 
|
42  | 
result = tree.branch.create_clone_on_transport(target_transport,  | 
|
43  | 
revision_id=old_revid)  | 
|
44  | 
self.assertEqual(old_revid, result.last_revision())  | 
|
45  | 
result.lock_read()  | 
|
46  | 
self.addCleanup(result.unlock)  | 
|
47  | 
self.assertFalse(result.repository.has_revision(source_tip))  | 
|
48  | 
||
49  | 
def test_create_clone_on_transport_stacked(self):  | 
|
50  | 
tree = self.make_branch_and_tree('source')  | 
|
51  | 
tree.commit('a commit')  | 
|
52  | 
trunk = tree.branch.create_clone_on_transport(  | 
|
53  | 
self.get_transport('trunk'))  | 
|
54  | 
revid = tree.commit('a second commit')  | 
|
55  | 
source = tree.branch  | 
|
56  | 
target_transport = self.get_transport('target')  | 
|
57  | 
result = tree.branch.create_clone_on_transport(target_transport,  | 
|
58  | 
stacked_on=trunk.base)  | 
|
59  | 
self.assertEqual(revid, result.last_revision())  | 
|
60  | 
self.assertEqual(trunk.base, result.get_stacked_on_url())  | 
|
| 
4053.2.2
by Andrew Bennetts
 Better fix, with test.  | 
61  | 
|
| 
4053.2.3
by Andrew Bennetts
 Fix some nits.  | 
62  | 
def test_create_clone_of_multiple_roots(self):  | 
| 
4053.2.2
by Andrew Bennetts
 Better fix, with test.  | 
63  | 
try:  | 
64  | 
builder = self.make_branch_builder('local')  | 
|
65  | 
except (errors.TransportNotPossible, errors.UninitializableFormat):  | 
|
66  | 
raise tests.TestNotApplicable('format not directly constructable')  | 
|
67  | 
builder.start_series()  | 
|
68  | 
builder.build_snapshot('rev1', None, [  | 
|
69  | 
('add', ('', 'root-id', 'directory', ''))])  | 
|
70  | 
builder.build_snapshot('rev2', ['rev1'], [])  | 
|
71  | 
builder.build_snapshot('other', None, [  | 
|
72  | 
('add', ('', 'root-id', 'directory', ''))])  | 
|
73  | 
builder.build_snapshot('rev3', ['rev2', 'other'], [])  | 
|
74  | 
builder.finish_series()  | 
|
75  | 
local = builder.get_branch()  | 
|
76  | 
local.bzrdir.clone(self.get_url('remote'), revision_id='rev3')  | 
|
77  | 
||
| 
4050.1.1
by Robert Collins
 Fix race condition with branch hooks during cloning when the new branch is stacked.  | 
78  | 
def assertBranchHookBranchIsStacked(self, pre_change_params):  | 
79  | 
        # Just calling will either succeed or fail.
 | 
|
80  | 
pre_change_params.branch.get_stacked_on_url()  | 
|
81  | 
self.hook_calls.append(pre_change_params)  | 
|
82  | 
||
83  | 
def test_create_clone_on_transport_stacked_hooks_get_stacked_branch(self):  | 
|
84  | 
tree = self.make_branch_and_tree('source')  | 
|
85  | 
tree.commit('a commit')  | 
|
86  | 
trunk = tree.branch.create_clone_on_transport(  | 
|
87  | 
self.get_transport('trunk'))  | 
|
88  | 
revid = tree.commit('a second commit')  | 
|
89  | 
source = tree.branch  | 
|
90  | 
target_transport = self.get_transport('target')  | 
|
91  | 
self.hook_calls = []  | 
|
92  | 
Branch.hooks.install_named_hook("pre_change_branch_tip",  | 
|
93  | 
self.assertBranchHookBranchIsStacked, None)  | 
|
94  | 
result = tree.branch.create_clone_on_transport(target_transport,  | 
|
95  | 
stacked_on=trunk.base)  | 
|
96  | 
self.assertEqual(revid, result.last_revision())  | 
|
97  | 
self.assertEqual(trunk.base, result.get_stacked_on_url())  | 
|
98  | 
        # Smart servers invoke hooks on both sides
 | 
|
99  | 
if isinstance(result, remote.RemoteBranch):  | 
|
100  | 
expected_calls = 2  | 
|
101  | 
else:  | 
|
102  | 
expected_calls = 1  | 
|
103  | 
self.assertEqual(expected_calls, len(self.hook_calls))  |