1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
# Disable pylint 'should have "self" as first argument' warnings.
# pylint: disable-msg=E0213
"""Branch XMLRPC API."""
__metaclass__ = type
__all__ = [
'BranchSetAPI',
'IBranchSetAPI',
'IPublicCodehostingAPI',
'PublicCodehostingAPI',
]
from xmlrpclib import Fault
from bzrlib import urlutils
from zope.component import getUtility
from zope.interface import (
implements,
Interface,
)
from canonical.config import config
from canonical.launchpad.webapp import (
canonical_url,
LaunchpadXMLRPCView,
)
from canonical.launchpad.webapp.interfaces import ILaunchBag
from lp.xmlrpc import faults
from lp.xmlrpc.helpers import return_fault
from lp.app.errors import NotFoundError
from lp.app.validators import LaunchpadValidationError
from lp.bugs.interfaces.bug import IBugSet
from lp.code.enums import BranchType
from lp.code.errors import (
BranchCreationException,
BranchCreationForbidden,
CannotHaveLinkedBranch,
InvalidNamespace,
NoLinkedBranch,
NoSuchBranch,
)
from lp.code.interfaces.branch import IBranch
from lp.code.interfaces.branchlookup import IBranchLookup
from lp.code.interfaces.branchnamespace import get_branch_namespace
from lp.code.interfaces.codehosting import (
BRANCH_ALIAS_PREFIX,
compose_public_url,
SUPPORTED_SCHEMES,
)
from lp.registry.errors import (
NoSuchDistroSeries,
NoSuchSourcePackageName,
)
from lp.registry.interfaces.person import (
IPersonSet,
NoSuchPerson,
)
from lp.registry.interfaces.product import (
InvalidProductName,
IProductSet,
NoSuchProduct,
)
from lp.registry.interfaces.productseries import NoSuchProductSeries
class IBranchSetAPI(Interface):
"""An XMLRPC interface for dealing with branches.
This XML-RPC interface was introduced to support Bazaar 0.8-2, which is
included in Ubuntu 6.06. This interface cannot be removed until Ubuntu
6.06 is end-of-lifed.
"""
def register_branch(branch_url, branch_name, branch_title,
branch_description, author_email, product_name,
owner_name=''):
"""Register a new branch in Launchpad."""
def link_branch_to_bug(branch_url, bug_id):
"""Link the branch to the bug."""
class BranchSetAPI(LaunchpadXMLRPCView):
implements(IBranchSetAPI)
def register_branch(self, branch_url, branch_name, branch_title,
branch_description, author_email, product_name,
owner_name=''):
"""See IBranchSetAPI."""
registrant = getUtility(ILaunchBag).user
assert registrant is not None, (
"register_branch shouldn't be accessible to unauthenicated"
" requests.")
person_set = getUtility(IPersonSet)
if owner_name:
owner = person_set.getByName(owner_name)
if owner is None:
return faults.NoSuchPersonWithName(owner_name)
if not registrant.inTeam(owner):
return faults.NotInTeam(registrant.name, owner_name)
else:
owner = registrant
if product_name:
product = getUtility(IProductSet).getByName(product_name)
if product is None:
return faults.NoSuchProduct(product_name)
else:
product = None
# Branch URLs in Launchpad do not end in a slash, so strip any
# slashes from the end of the URL.
branch_url = branch_url.rstrip('/')
branch_lookup = getUtility(IBranchLookup)
existing_branch = branch_lookup.getByUrl(branch_url)
if existing_branch is not None:
return faults.BranchAlreadyRegistered(branch_url)
try:
unicode_branch_url = branch_url.decode('utf-8')
IBranch['url'].validate(unicode_branch_url)
except LaunchpadValidationError, exc:
return faults.InvalidBranchUrl(branch_url, exc)
# We want it to be None in the database, not ''.
if not branch_description:
branch_description = None
if not branch_title:
branch_title = None
if not branch_name:
branch_name = unicode_branch_url.split('/')[-1]
try:
if branch_url:
branch_type = BranchType.MIRRORED
else:
branch_type = BranchType.HOSTED
namespace = get_branch_namespace(owner, product)
branch = namespace.createBranch(
branch_type=branch_type,
name=branch_name, registrant=registrant,
url=branch_url, title=branch_title,
summary=branch_description)
if branch_type == BranchType.MIRRORED:
branch.requestMirror()
except BranchCreationForbidden:
return faults.BranchCreationForbidden(product.displayname)
except BranchCreationException, err:
return faults.BranchNameInUse(err)
except LaunchpadValidationError, err:
return faults.InvalidBranchName(err)
return canonical_url(branch)
def link_branch_to_bug(self, branch_url, bug_id):
"""See IBranchSetAPI."""
branch = getUtility(IBranchLookup).getByUrl(url=branch_url)
if branch is None:
return faults.NoSuchBranch(branch_url)
try:
bug = getUtility(IBugSet).get(bug_id)
except NotFoundError:
return faults.NoSuchBug(bug_id)
# Since this API is controlled using launchpad.AnyPerson there must be
# an authenticated person, so use this person as the registrant.
registrant = getUtility(ILaunchBag).user
bug.linkBranch(branch, registrant=registrant)
return canonical_url(bug)
class IPublicCodehostingAPI(Interface):
"""The public codehosting API."""
def resolve_lp_path(path):
"""Expand the path segment of an lp: URL into a list of branch URLs.
This method is added to Bazaar in 0.93.
:return: A dict containing a single 'urls' key that maps to a list of
URLs. Clients should use the first URL in the list that they can
support. Returns a Fault if the path does not resolve to a
branch.
"""
class _NonexistentBranch:
"""Used to represent a branch that was requested but doesn't exist."""
def __init__(self, unique_name):
self.unique_name = unique_name
self.branch_type = None
class PublicCodehostingAPI(LaunchpadXMLRPCView):
"""See `IPublicCodehostingAPI`."""
implements(IPublicCodehostingAPI)
def _compose_http_url(unique_name, path, suffix):
return compose_public_url('http', unique_name, suffix)
def _compose_bzr_ssh_url(unique_name, path, suffix):
if not path.startswith('~'):
path = '%s/%s' % (BRANCH_ALIAS_PREFIX, path)
return compose_public_url('bzr+ssh', path, suffix)
scheme_funcs = {
'bzr+ssh': _compose_bzr_ssh_url,
'http': _compose_http_url,
}
def _getUrlsForBranch(self, branch, lp_path, suffix=None,
supported_schemes=None):
"""Return a list of URLs for the given branch.
:param branch: A Branch object.
:param lp_path: The path that was used to traverse to the branch.
:param suffix: The section of the path that follows the branch
specification.
:return: {'urls': [list_of_branch_urls]}.
"""
if branch.branch_type == BranchType.REMOTE:
if branch.url is None:
raise faults.NoUrlForBranch(branch.unique_name)
return [branch.url]
else:
return self._getUniqueNameResultDict(
branch.unique_name, suffix, supported_schemes, lp_path)
def _getUniqueNameResultDict(self, unique_name, suffix=None,
supported_schemes=None, path=None):
if supported_schemes is None:
supported_schemes = SUPPORTED_SCHEMES
if path is None:
path = unique_name
return [self.scheme_funcs[scheme](unique_name, path, suffix)
for scheme in supported_schemes]
@return_fault
def _resolve_lp_path(self, path):
"""See `IPublicCodehostingAPI`."""
# Separate method because Zope's mapply raises errors if we use
# decorators in XMLRPC methods. mapply checks that the passed
# arguments match the formal parameters. Decorators normally have
# *args and **kwargs, which mapply fails on.
strip_path = path.strip('/')
if strip_path == '':
raise faults.InvalidBranchIdentifier(path)
supported_schemes = list(SUPPORTED_SCHEMES)
hot_products = [product.strip() for product
in config.codehosting.hot_products.split(',')]
# If we have been given something that looks like a branch name, just
# look that up.
if strip_path.startswith('~'):
urls = self._getBranchPaths(strip_path, supported_schemes)
else:
# We only check the hot product code when accessed through the
# short name, so we can check it here.
if strip_path in hot_products:
supported_schemes = ['http']
urls = []
else:
urls = [self.scheme_funcs['bzr+ssh'](None, strip_path, None)]
supported_schemes.remove('bzr+ssh')
# Try to look up the branch at that url and add alternative URLs.
# This may well fail, and if it does, we just return the aliased
# url.
try:
urls.extend(
self._getBranchPaths(strip_path, supported_schemes))
except Fault:
pass
return dict(urls=urls)
def _getBranchPaths(self, strip_path, supported_schemes):
"""Get the specific paths for a branch.
If the branch is not found, but it looks like a branch name, then we
return a writable URL for it. If it doesn't look like a branch name a
fault is raised.
"""
branch_set = getUtility(IBranchLookup)
try:
branch, suffix = branch_set.getByLPPath(strip_path)
except NoSuchBranch:
# If the branch isn't found, but it looks like a valid name, then
# resolve it anyway, treating the path like a branch's unique
# name. This lets people push new branches up to Launchpad using
# lp: URL syntax.
supported_schemes = ['bzr+ssh']
return self._getUniqueNameResultDict(
strip_path, supported_schemes=supported_schemes)
# XXX: JonathanLange 2009-03-21 bug=347728: All of this is repetitive
# and thus error prone. Alternatives are directly raising faults from
# the model code(blech) or some automated way of reraising as faults
# or using a narrower range of faults (e.g. only one "NoSuch" fault).
except InvalidProductName, e:
raise faults.InvalidProductIdentifier(urlutils.escape(e.name))
except NoSuchProductSeries, e:
raise faults.NoSuchProductSeries(
urlutils.escape(e.name), e.product)
except NoSuchPerson, e:
raise faults.NoSuchPersonWithName(urlutils.escape(e.name))
except NoSuchProduct, e:
raise faults.NoSuchProduct(urlutils.escape(e.name))
except NoSuchDistroSeries, e:
raise faults.NoSuchDistroSeries(urlutils.escape(e.name))
except NoSuchSourcePackageName, e:
raise faults.NoSuchSourcePackageName(urlutils.escape(e.name))
except NoLinkedBranch, e:
raise faults.NoLinkedBranch(e.component)
except CannotHaveLinkedBranch, e:
raise faults.CannotHaveLinkedBranch(e.component)
except InvalidNamespace, e:
raise faults.InvalidBranchUniqueName(urlutils.escape(e.name))
# Reverse engineer the actual lp_path that is used, so we need to
# remove any suffix that may be there from the strip_path.
lp_path = strip_path
if suffix is not None:
# E.g. 'project/trunk/filename.txt' the suffix is 'filename.txt'
# we want lp_path to be 'project/trunk'.
lp_path = lp_path[:-(len(suffix)+1)]
return self._getUrlsForBranch(
branch, lp_path, suffix, supported_schemes)
def resolve_lp_path(self, path):
"""See `IPublicCodehostingAPI`."""
return self._resolve_lp_path(path)
|