~abentley/bzrtools/bzrtools.dev

« back to all changes in this revision

Viewing changes to zap.py

  • Committer: Aaron Bentley
  • Date: 2014-01-13 00:50:22 UTC
  • Revision ID: aaron@aaronbentley.com-20140113005022-q40nvybtox1mt19o
Tags: release-2.6.0
Update for release 2.6.0

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2006-2007, 2010-2011 Aaron Bentley <aaron@aaronbentley.com>
 
2
# Copyright (C) 2013 Aaron Bentley <aaron@aaronbentley.com>
 
3
# Copyright (C) 2007 Charlie Shepherd <masterdriverz@gentoo.org>
 
4
# Copyright (C) 2011 Canonical Ltd.
 
5
#
 
6
#    This program is free software; you can redistribute it and/or modify
 
7
#    it under the terms of the GNU General Public License as published by
 
8
#    the Free Software Foundation; either version 2 of the License, or
 
9
#    (at your option) any later version.
 
10
#
 
11
#    This program is distributed in the hope that it will be useful,
 
12
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
 
13
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
14
#    GNU General Public License for more details.
 
15
#
 
16
#    You should have received a copy of the GNU General Public License
 
17
#    along with this program; if not, write to the Free Software
 
18
#    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
19
 
1
20
from shutil import rmtree
2
21
 
3
22
from bzrlib import (
5
24
    revision as _mod_revision,
6
25
    )
7
26
from bzrlib.branch import Branch
8
 
from bzrlib.errors import NoWorkingTree, NotLocalUrl, NotBranchError
 
27
from bzrlib.errors import BzrCommandError, NoWorkingTree, NotBranchError
 
28
from bzrlib import registry
9
29
from bzrlib.workingtree import WorkingTree
10
30
 
11
31
from errors import (NotCheckout, UncommittedCheckout, ParentMissingRevisions,
12
32
                    NoParent)
13
 
 
14
 
 
15
 
def zap(path, remove_branch=False, allow_modified=False):
 
33
from bzrlib.plugins.bzrtools.bzrtools import read_locked
 
34
 
 
35
 
 
36
class AllowChanged(object):
 
37
 
 
38
    @classmethod
 
39
    def check_changed(klass, wt, remove_branch):
 
40
        pass
 
41
 
 
42
 
 
43
class CheckChanged(object):
 
44
 
 
45
    @classmethod
 
46
    def check_changed(klass, wt, remove_branch):
 
47
        delta = wt.changes_from(wt.basis_tree(), want_unchanged=False)
 
48
        if delta.has_changed():
 
49
            klass.handle_changed(wt, remove_branch)
 
50
 
 
51
 
 
52
class HaltOnChange(CheckChanged):
 
53
 
 
54
    @staticmethod
 
55
    def handle_changed(wt, remove_branch):
 
56
        raise UncommittedCheckout()
 
57
 
 
58
 
 
59
class StoreChanges(CheckChanged):
 
60
 
 
61
    @staticmethod
 
62
    def handle_changed(wt, remove_branch):
 
63
        from bzrlib.plugins.pipeline.pipeline import PipeManager
 
64
        if remove_branch:
 
65
            raise BzrCommandError('Cannot store changes in deleted branch.')
 
66
        PipeManager.from_checkout(wt).store_uncommitted()
 
67
 
 
68
 
 
69
change_policy_registry = registry.Registry()
 
70
 
 
71
 
 
72
change_policy_registry.register('force', AllowChanged,
 
73
                                'Delete tree even if contents are modified.')
 
74
 
 
75
 
 
76
change_policy_registry.register('store', StoreChanges,
 
77
                                'Store changes in branch.  (Requires'
 
78
                                ' bzr-pipeline.)')
 
79
 
 
80
 
 
81
change_policy_registry.register('check', HaltOnChange,
 
82
                                'Stop if tree contents are modified.')
 
83
 
 
84
 
 
85
change_policy_registry.default_key = 'check'
 
86
 
 
87
 
 
88
 
 
89
def zap(path, remove_branch=False, policy=HaltOnChange):
16
90
    try:
17
91
        wt = bzrdir.BzrDir.open(path).open_workingtree(path,
18
92
                                                       recommend_upgrade=False)
23
97
    branch_base = branch.bzrdir.transport.base
24
98
    if tree_base == branch_base:
25
99
        raise NotCheckout(path)
26
 
    if not allow_modified:
27
 
        delta = wt.changes_from(wt.basis_tree(), want_unchanged=False)
28
 
        if delta.has_changed():
29
 
            raise UncommittedCheckout()
 
100
    policy.check_changed(wt, remove_branch)
30
101
    if remove_branch:
31
102
        parent_loc = branch.get_parent()
32
103
        if parent_loc is None:
33
104
            raise NoParent()
34
 
        parent = Branch.open(parent_loc)
35
 
        last_revision = _mod_revision.ensure_null(parent.last_revision())
36
 
        p_ancestry = parent.repository.get_ancestry(last_revision)
37
 
        if (last_revision != _mod_revision.NULL_REVISION and
38
 
            branch.last_revision() not in p_ancestry):
39
 
            raise ParentMissingRevisions(branch.get_parent())
 
105
        with read_locked(Branch.open(parent_loc)) as parent:
 
106
            last_revision = _mod_revision.ensure_null(parent.last_revision())
 
107
            if last_revision != _mod_revision.NULL_REVISION:
 
108
                graph = parent.repository.get_graph()
 
109
                heads = graph.heads([last_revision, branch.last_revision()])
 
110
                if branch.last_revision() in heads:
 
111
                    raise ParentMissingRevisions(branch.get_parent())
40
112
    rmtree(path)
41
113
    if remove_branch:
42
114
        t = branch.bzrdir.transport
50
122
    import os
51
123
    from unittest import makeSuite
52
124
 
53
 
    from bzrlib.bzrdir import BzrDir, BzrDirMetaFormat1
54
 
    from bzrlib.branch import BranchReferenceFormat
55
 
    from bzrlib.tests import TestCaseInTempDir
56
 
 
57
 
    class TestZap(TestCaseInTempDir):
 
125
    from bzrlib.tests import TestCaseWithTransport
 
126
 
 
127
 
 
128
    class PipelinePluginFeature:
 
129
 
 
130
        @staticmethod
 
131
        def available():
 
132
            try:
 
133
                import bzrlib.plugins.pipeline
 
134
            except ImportError:
 
135
                return False
 
136
            else:
 
137
                return True
 
138
 
 
139
 
 
140
    class TestZap(TestCaseWithTransport):
58
141
 
59
142
        def make_checkout(self):
60
 
            wt = BzrDir.create_standalone_workingtree('source')
61
 
            os.mkdir('checkout')
62
 
            checkout = BzrDirMetaFormat1().initialize('checkout')
63
 
            BranchReferenceFormat().initialize(checkout, wt.branch)
64
 
            return checkout.create_workingtree()
 
143
            wt = bzrdir.BzrDir.create_standalone_workingtree('source')
 
144
            return wt.branch.create_checkout('checkout', lightweight=True)
65
145
 
66
146
        def make_checkout2(self):
67
147
            wt = self.make_checkout()
68
148
            wt2 = wt.branch.bzrdir.sprout('source2').open_workingtree()
69
 
            os.mkdir('checkout2')
70
 
            checkout = BzrDirMetaFormat1().initialize('checkout2')
71
 
            BranchReferenceFormat().initialize(checkout, wt2.branch)
72
 
            return checkout.create_workingtree()
 
149
            return wt2.branch.create_checkout('checkout2', lightweight=True)
73
150
 
74
151
        def test_is_checkout(self):
75
152
            self.assertRaises(NotCheckout, zap, '.')
76
 
            wt = BzrDir.create_standalone_workingtree('.')
 
153
            wt = bzrdir.BzrDir.create_standalone_workingtree('.')
77
154
            self.assertRaises(NotCheckout, zap, '.')
78
155
 
79
156
        def test_zap_works(self):
83
160
            self.assertIs(False, os.path.exists('checkout'))
84
161
            self.assertIs(True, os.path.exists('source'))
85
162
 
86
 
        def test_zap_branch(self):
 
163
        def test_zap_branch(self):
87
164
            self.make_checkout2()
88
165
            base = WorkingTree.open('checkout').branch.base
89
166
            self.assertIs(True, os.path.exists('checkout'))
102
179
            checkout.commit('commit changes to branch')
103
180
            zap('checkout')
104
181
 
 
182
        def make_modified_checkout(self):
 
183
            checkout = self.make_checkout()
 
184
            os.mkdir('checkout/foo')
 
185
            checkout.add('foo')
 
186
            return checkout
 
187
 
105
188
        def test_allow_modified(self):
 
189
            self.make_modified_checkout()
 
190
            self.assertRaises(UncommittedCheckout, zap, 'checkout')
 
191
            zap('checkout', policy=AllowChanged)
 
192
 
 
193
        def test_store(self):
 
194
            self.requireFeature(PipelinePluginFeature)
 
195
            checkout = self.make_modified_checkout()
 
196
            zap('checkout', policy=StoreChanges)
 
197
            self.assertIn('stored-transform',
 
198
                checkout.branch.bzrdir.transport.list_dir('branch'))
 
199
 
 
200
        def test_store_remove_branch(self):
 
201
            self.requireFeature(PipelinePluginFeature)
 
202
            checkout = self.make_modified_checkout()
 
203
            branch = self.make_branch('branch')
 
204
            checkout.branch.set_parent(branch.base)
 
205
            e = self.assertRaises(BzrCommandError, zap, 'checkout',
 
206
                                  policy=StoreChanges, remove_branch=True)
 
207
            self.assertEqual('Cannot store changes in deleted branch.', str(e))
 
208
 
 
209
        def test_store_remove_branch_unmodified(self):
 
210
            self.requireFeature(PipelinePluginFeature)
106
211
            checkout = self.make_checkout()
107
 
            os.mkdir('checkout/foo')
108
 
            checkout.add('foo')
109
 
            self.assertRaises(UncommittedCheckout, zap, 'checkout')
110
 
            zap('checkout', allow_modified=True)
 
212
            branch = self.make_branch('branch')
 
213
            checkout.branch.set_parent(branch.base)
 
214
            zap('checkout', policy=StoreChanges, remove_branch=True)
 
215
 
 
216
        def test_zap_branch_with_unique_revision(self):
 
217
            parent = self.make_branch_and_tree('parent')
 
218
            parent.commit('foo')
 
219
            new_branch = self.make_branch('new')
 
220
            new_branch.set_parent(parent.branch.base)
 
221
            checkout = new_branch.create_checkout('checkout', lightweight=True)
 
222
            checkout.commit('unique')
 
223
            self.assertRaises(ParentMissingRevisions, zap, 'checkout',
 
224
                              remove_branch=True)
111
225
 
112
226
    return makeSuite(TestZap)