~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/branch_implementations/test_push.py

  • Committer: Robert Collins
  • Date: 2007-03-08 04:06:06 UTC
  • mfrom: (2323.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 2442.
  • Revision ID: robertc@robertcollins.net-20070308040606-84gsniv56huiyjt4
Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2004, 2005, 2007 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.push behaviour."""
 
18
 
 
19
import os
 
20
 
 
21
from bzrlib.branch import Branch
 
22
from bzrlib import errors
 
23
from bzrlib.memorytree import MemoryTree
 
24
from bzrlib.revision import NULL_REVISION
 
25
from bzrlib.tests.branch_implementations.test_branch import TestCaseWithBranch
 
26
 
 
27
 
 
28
class TestPush(TestCaseWithBranch):
 
29
 
 
30
    def test_push_convergence_simple(self):
 
31
        # when revisions are pushed, the left-most accessible parents must 
 
32
        # become the revision-history.
 
33
        mine = self.make_branch_and_tree('mine')
 
34
        mine.commit('1st post', rev_id='P1', allow_pointless=True)
 
35
        other = mine.bzrdir.sprout('other').open_workingtree()
 
36
        other.commit('my change', rev_id='M1', allow_pointless=True)
 
37
        mine.merge_from_branch(other.branch)
 
38
        mine.commit('merge my change', rev_id='P2')
 
39
        result = mine.branch.push(other.branch)
 
40
        self.assertEqual(['P1', 'P2'], other.branch.revision_history())
 
41
        # result object contains some structured data
 
42
        self.assertEqual(result.old_revid, 'M1')
 
43
        self.assertEqual(result.new_revid, 'P2')
 
44
        # and it can be treated as an integer for compatibility
 
45
        self.assertEqual(int(result), 0)
 
46
 
 
47
    def test_push_merged_indirect(self):
 
48
        # it should be possible to do a push from one branch into another
 
49
        # when the tip of the target was merged into the source branch
 
50
        # via a third branch - so its buried in the ancestry and is not
 
51
        # directly accessible.
 
52
        mine = self.make_branch_and_tree('mine')
 
53
        mine.commit('1st post', rev_id='P1', allow_pointless=True)
 
54
        target = mine.bzrdir.sprout('target').open_workingtree()
 
55
        target.commit('my change', rev_id='M1', allow_pointless=True)
 
56
        other = mine.bzrdir.sprout('other').open_workingtree()
 
57
        other.merge_from_branch(target.branch)
 
58
        other.commit('merge my change', rev_id='O2')
 
59
        mine.merge_from_branch(other.branch)
 
60
        mine.commit('merge other', rev_id='P2')
 
61
        mine.branch.push(target.branch)
 
62
        self.assertEqual(['P1', 'P2'], target.branch.revision_history())
 
63
 
 
64
    def test_push_to_checkout_updates_master(self):
 
65
        """Pushing into a checkout updates the checkout and the master branch"""
 
66
        master_tree = self.make_branch_and_tree('master')
 
67
        checkout = self.make_branch_and_tree('checkout')
 
68
        try:
 
69
            checkout.branch.bind(master_tree.branch)
 
70
        except errors.UpgradeRequired:
 
71
            # cant bind this format, the test is irrelevant.
 
72
            return
 
73
        rev1 = checkout.commit('master')
 
74
 
 
75
        other = master_tree.branch.bzrdir.sprout('other').open_workingtree()
 
76
        rev2 = other.commit('other commit')
 
77
        # now push, which should update both checkout and master.
 
78
        other.branch.push(checkout.branch)
 
79
        self.assertEqual([rev1, rev2], checkout.branch.revision_history())
 
80
        self.assertEqual([rev1, rev2], master_tree.branch.revision_history())
 
81
 
 
82
    def test_push_raises_specific_error_on_master_connection_error(self):
 
83
        master_tree = self.make_branch_and_tree('master')
 
84
        checkout = self.make_branch_and_tree('checkout')
 
85
        try:
 
86
            checkout.branch.bind(master_tree.branch)
 
87
        except errors.UpgradeRequired:
 
88
            # cant bind this format, the test is irrelevant.
 
89
            return
 
90
        other = master_tree.branch.bzrdir.sprout('other').open_workingtree()
 
91
        # move the branch out of the way on disk to cause a connection
 
92
        # error.
 
93
        os.rename('master', 'master_gone')
 
94
        # try to push, which should raise a BoundBranchConnectionFailure.
 
95
        self.assertRaises(errors.BoundBranchConnectionFailure,
 
96
                other.branch.push, checkout.branch)
 
97
 
 
98
    def test_push_uses_read_lock(self):
 
99
        """Push should only need a read lock on the source side."""
 
100
        source = self.make_branch_and_tree('source')
 
101
        target = self.make_branch('target')
 
102
 
 
103
        self.build_tree(['source/a'])
 
104
        source.add(['a'])
 
105
        source.commit('a')
 
106
 
 
107
        source.branch.lock_read()
 
108
        try:
 
109
            target.lock_write()
 
110
            try:
 
111
                source.branch.push(target, stop_revision=source.last_revision())
 
112
            finally:
 
113
                target.unlock()
 
114
        finally:
 
115
            source.branch.unlock()
 
116
 
 
117
    def test_push_within_repository(self):
 
118
        """Push from one branch to another inside the same repository."""
 
119
        try:
 
120
            repo = self.make_repository('repo', shared=True)
 
121
        except (errors.IncompatibleFormat, errors.UninitializableFormat):
 
122
            # This Branch format cannot create shared repositories
 
123
            return
 
124
        # This is a little bit trickier because make_branch_and_tree will not
 
125
        # re-use a shared repository.
 
126
        a_bzrdir = self.make_bzrdir('repo/tree')
 
127
        try:
 
128
            a_branch = self.branch_format.initialize(a_bzrdir)
 
129
        except (errors.UninitializableFormat):
 
130
            # Cannot create these branches
 
131
            return
 
132
        tree = a_branch.bzrdir.create_workingtree()
 
133
        self.build_tree(['repo/tree/a'])
 
134
        tree.add(['a'])
 
135
        tree.commit('a')
 
136
 
 
137
        to_bzrdir = self.make_bzrdir('repo/branch')
 
138
        to_branch = self.branch_format.initialize(to_bzrdir)
 
139
        tree.branch.push(to_branch)
 
140
 
 
141
        self.assertEqual(tree.branch.last_revision(),
 
142
                         to_branch.last_revision())
 
143
 
 
144
 
 
145
class TestPushHook(TestCaseWithBranch):
 
146
 
 
147
    def setUp(self):
 
148
        self.hook_calls = []
 
149
        TestCaseWithBranch.setUp(self)
 
150
 
 
151
    def capture_post_push_hook(self, result):
 
152
        """Capture post push hook calls to self.hook_calls.
 
153
        
 
154
        The call is logged, as is some state of the two branches.
 
155
        """
 
156
        if result.local_branch:
 
157
            local_locked = result.local_branch.is_locked()
 
158
            local_base = result.local_branch.base
 
159
        else:
 
160
            local_locked = None
 
161
            local_base = None
 
162
        self.hook_calls.append(
 
163
            ('post_push', result.source_branch, local_base,
 
164
             result.master_branch.base,
 
165
             result.old_revno, result.old_revid,
 
166
             result.new_revno, result.new_revid,
 
167
             result.source_branch.is_locked(), local_locked,
 
168
             result.master_branch.is_locked()))
 
169
 
 
170
    def test_post_push_empty_history(self):
 
171
        target = self.make_branch('target')
 
172
        source = self.make_branch('source')
 
173
        Branch.hooks.install_hook('post_push', self.capture_post_push_hook)
 
174
        source.push(target)
 
175
        # with nothing there we should still get a notification, and
 
176
        # have both branches locked at the notification time.
 
177
        self.assertEqual([
 
178
            ('post_push', source, None, target.base, 0, NULL_REVISION,
 
179
             0, NULL_REVISION, True, None, True)
 
180
            ],
 
181
            self.hook_calls)
 
182
 
 
183
    def test_post_push_bound_branch(self):
 
184
        # pushing to a bound branch should pass in the master branch to the
 
185
        # hook, allowing the correct number of emails to be sent, while still
 
186
        # allowing hooks that want to modify the target to do so to both 
 
187
        # instances.
 
188
        target = self.make_branch('target')
 
189
        local = self.make_branch('local')
 
190
        try:
 
191
            local.bind(target)
 
192
        except errors.UpgradeRequired:
 
193
            # cant bind this format, the test is irrelevant.
 
194
            return
 
195
        source = self.make_branch('source')
 
196
        Branch.hooks.install_hook('post_push', self.capture_post_push_hook)
 
197
        source.push(local)
 
198
        # with nothing there we should still get a notification, and
 
199
        # have both branches locked at the notification time.
 
200
        self.assertEqual([
 
201
            ('post_push', source, local.base, target.base, 0, NULL_REVISION,
 
202
             0, NULL_REVISION, True, True, True)
 
203
            ],
 
204
            self.hook_calls)
 
205
 
 
206
    def test_post_push_nonempty_history(self):
 
207
        target = self.make_branch_and_memory_tree('target')
 
208
        target.lock_write()
 
209
        target.add('')
 
210
        rev1 = target.commit('rev 1')
 
211
        target.unlock()
 
212
        sourcedir = target.bzrdir.clone(self.get_url('source'))
 
213
        source = MemoryTree.create_on_branch(sourcedir.open_branch())
 
214
        rev2 = source.commit('rev 2')
 
215
        Branch.hooks.install_hook('post_push', self.capture_post_push_hook)
 
216
        source.branch.push(target.branch)
 
217
        # with nothing there we should still get a notification, and
 
218
        # have both branches locked at the notification time.
 
219
        self.assertEqual([
 
220
            ('post_push', source.branch, None, target.branch.base, 1, rev1,
 
221
             2, rev2, True, None, True)
 
222
            ],
 
223
            self.hook_calls)