« back to all changes in this revision
Viewing changes to mojo/bzr.py
- browse files at revision 288
- compare with another revision
- download diff
- download tarball
- view history from revision 288
- Committer: Tom Haddon
- Date: 2016-02-15 09:28:35 UTC
- mfrom: (285.2.8 mojo)
- Revision ID: tom.haddon@canonical.com-20160215092835-0gsjqhdc60x29dbo
[timkuhlman,r=mthaddon] Moved to using codetree to parse/branch/fetch the spec url
added
removed
mojo/bzr.py
1
# Copyright 2014 Canonical Ltd. This software is licensed under the
2
# GNU General Public License version 3 (see the file LICENSE).
3
import os
4
import shutil
5
from bzrlib import (
6
errors,
7
urlutils,
8
)
9
from .utils import (
10
CacheDisk
11
)
12
from bzrlib.branch import Branch
13
from bzrlib.bzrdir import BzrDir
14
from bzrlib.plugin import load_plugins
15
from bzrlib.workingtree import WorkingTree
16
17
load_plugins()
18
19
20
class NotABranchError(Exception):
21
pass
22
23
24
class NoSuchFileError(Exception):
25
pass
26
27
28
class DivergedBranchError(Exception):
29
pass
30
31
32
class PermissionDeniedError(Exception):
33
pass
34
35
36
class NoWorkingTreeError(Exception):
37
pass
38
39
40
class AlreadyControlDirError(Exception):
41
pass
42
43
44
class BzrBranch(object):
45
"Simplified bazaar branch API"
46
def __init__(self, location):
47
if location.startswith("file://"):
48
location = location[7:]
49
if urlutils.is_url(location):
50
self.is_local = False
51
else:
52
if not os.path.isabs(location):
53
location = os.path.normpath(os.path.join(os.getcwd(),
54
location))
55
self.is_local = True
56
self.location = location
57
self._branch = None
58
self._tree = None
59
60
@property
61
def bzr_branch(self):
62
if not self._branch:
63
try:
64
self._branch = Branch.open(self.location)
65
except errors.NotBranchError:
66
return None
67
return self._branch
68
69
@property
70
def bzr_tree(self):
71
if self._tree:
72
return self._tree
73
if self.is_local:
74
try:
75
self._tree = WorkingTree.open(self.location)
76
return self._tree
77
except errors.NoWorkingTree:
78
return None
79
else:
80
return None
81
82
@property
83
def is_remote(self):
84
return not self.is_local
85
86
@property
87
def is_branch(self):
88
if self.is_local:
89
return self.is_local_branch()
90
else:
91
return self.is_remote_branch()
92
93
def is_local_branch(self):
94
if self.bzr_branch is None:
95
return False
96
return True
97
98
# only cache remote is_branch() calls
99
@CacheDisk.cache(unique_attrs=('location',))
100
def is_remote_branch(self):
101
if self.bzr_branch is None:
102
return False
103
return True
104
105
@property
106
def has_tree(self):
107
if self.bzr_tree is None:
108
return False
109
return True
110
111
@property
112
@CacheDisk.cache(unique_attrs=('location',))
113
def revno(self):
114
if self.bzr_branch:
115
return self.bzr_branch.revno()
116
117
@property
118
def revno_tree(self):
119
if self.is_remote or not self.bzr_tree:
120
return None
121
return self.bzr_branch.revision_id_to_revno(
122
self.bzr_tree.last_revision())
123
124
@property
125
def parent(self):
126
if self.is_remote or self.bzr_branch is None:
127
return None
128
return self.bzr_branch.get_parent()
129
130
def has_changes(self, unknowns=False, detritus=False):
131
"Check whether changes or stray files are present in the branch"
132
if self.is_remote:
133
return False
134
if not self.bzr_tree:
135
return False
136
if self.bzr_tree.has_changes():
137
return True
138
if unknowns and list(self.bzr_tree.unknowns()):
139
return True
140
if detritus and list(self.bzr_tree.ignored_files()):
141
return True
142
return False
143
144
def revert(self):
145
if self.is_remote or not self.bzr_tree:
146
raise NoWorkingTreeError
147
self.bzr_tree.revert()
148
149
def cleantree(self, unknowns=True, detritus=True):
150
if self.is_remote or not self.bzr_tree:
151
raise NoWorkingTreeError
152
targets = []
153
if unknowns:
154
targets.extend(list(self.bzr_tree.unknowns()))
155
if detritus:
156
self.bzr_tree.lock_read()
157
targets.extend([f[0] for f in self.bzr_tree.ignored_files()])
158
for target in [os.path.join(self.location, t) for t in targets]:
159
if os.path.isdir(target):
160
shutil.rmtree(target)
161
else:
162
os.unlink(target)
163
164
def clone(self, dest_dir, revno=None):
165
if not self.bzr_branch:
166
return None
167
if revno is not None:
168
revision_id = self.bzr_branch.get_rev_id(revno)
169
else:
170
revision_id = None
171
try:
172
self.bzr_branch.bzrdir.sprout(dest_dir, revision_id=revision_id)
173
except errors.PermissionDenied:
174
raise PermissionDeniedError("Could not create bzr repo in {}"
175
"".format(dest_dir))
176
except errors.NoSuchFile:
177
raise NoSuchFileError("No such file or directory: {}"
178
"".format(dest_dir))
179
cloned = BzrBranch(dest_dir)
180
CacheDisk.delete_cache_dir(cloned, unique_attrs=('location',))
181
return cloned
182
183
@CacheDisk.delete(unique_attrs=('location',))
184
def create(self):
185
"bzr init"
186
if self.is_local:
187
try:
188
BzrDir.create_branch_convenience(self.location)
189
except errors.AlreadyControlDirError:
190
raise AlreadyControlDirError
191
192
def pull(self, src_location=None, overwrite=False, revno=None):
193
if not self.is_branch:
194
raise NotABranchError("{} is not a branch".format(self.location))
195
if src_location is None and self.is_local:
196
src_location = self.parent
197
other = BzrBranch(src_location)
198
if not other.is_branch:
199
raise NotABranchError("{} is not a branch".format(other.location))
200
try:
201
if self.bzr_tree:
202
self.bzr_tree.pull(other.bzr_branch,
203
overwrite=overwrite,
204
stop_revision=revno)
205
self.bzr_tree.update(revision=revno)
206
else:
207
self.bzr_branch.pull(other.bzr_branch,
208
overwrite=overwrite,
209
stop_revision=revno)
210
except errors.DivergedBranches:
211
message = "Branches have diverged: {} and {}".format(
212
self.location, other.location)
213
raise DivergedBranchError(message)
214
return True
215
216
def push(self, dest_location, update=True, update_revno=None,
217
overwrite=False):
218
dest_branch = BzrBranch(dest_location)
219
if not dest_branch.is_branch:
220
return None
221
try:
222
self.bzr_branch.push(dest_branch.bzr_branch, overwrite=overwrite)
223
except errors.DivergedBranches:
224
raise DivergedBranchError
225
if update:
226
dest_branch.update(update_revno)
227
return dest_branch
228
229
def update(self, revno=None):
230
rev_id = None
231
if revno is not None:
232
rev_id = self.bzr_branch.get_rev_id(revno)
233
self.bzr_tree.update(revision=rev_id)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1