1
# Copyright (C) 2010, 2011 Canonical Ltd
2
# Copyright (C) 2018 Breezy Developers
4
# This program is free software; you can redistribute it and/or modify
5
# it under the terms of the GNU General Public License as published by
6
# the Free Software Foundation; either version 2 of the License, or
7
# (at your option) any later version.
9
# This program is distributed in the hope that it will be useful,
10
# but WITHOUT ANY WARRANTY; without even the implied warranty of
11
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12
# GNU General Public License for more details.
14
# You should have received a copy of the GNU General Public License
15
# along with this program; if not, write to the Free Software
16
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
18
"""Support for Launchpad."""
20
from __future__ import absolute_import
24
from .propose import (
34
branch as _mod_branch,
40
from ...git.refs import ref_to_branch_name
41
from ...lazy_import import lazy_import
42
lazy_import(globals(), """
43
from breezy.plugins.launchpad import (
48
from ...transport import get_transport
51
# TODO(jelmer): Make selection of launchpad staging a configuration option.
53
def status_to_lp_mp_statuses(status):
55
if status in ('open', 'all'):
60
'Code failed to merge',
62
if status in ('closed', 'all'):
63
statuses.extend(['Rejected', 'Superseded'])
64
if status in ('merged', 'all'):
65
statuses.append('Merged')
69
def plausible_launchpad_url(url):
72
if url.startswith('lp:'):
74
regex = re.compile('([a-z]*\+)*(bzr\+ssh|http|ssh|git|https)'
75
'://(bazaar|git).*.launchpad.net')
76
return bool(regex.match(url))
79
class WebserviceFailure(Exception):
81
def __init__(self, message):
82
self.message = message
85
def _call_webservice(call, *args, **kwargs):
86
"""Make a call to the webservice, wrapping failures.
88
:param call: The call to make.
89
:param *args: *args for the call.
90
:param **kwargs: **kwargs for the call.
91
:return: The result of calling call(*args, *kwargs).
93
from lazr.restfulclient import errors as restful_errors
95
return call(*args, **kwargs)
96
except restful_errors.HTTPError as e:
98
for line in e.content.splitlines():
99
if line.startswith(b'Traceback (most recent call last):'):
101
error_lines.append(line)
102
raise WebserviceFailure(b''.join(error_lines))
105
class LaunchpadMergeProposal(MergeProposal):
107
def __init__(self, mp):
110
def get_source_branch_url(self):
111
if self._mp.source_branch:
112
return self._mp.source_branch.bzr_identity
114
branch_name = ref_to_branch_name(
115
self._mp.source_git_path.encode('utf-8'))
116
return urlutils.join_segment_parameters(
117
self._mp.source_git_repository.git_identity,
118
{"branch": branch_name})
120
def get_target_branch_url(self):
121
if self._mp.target_branch:
122
return self._mp.target_branch.bzr_identity
124
branch_name = ref_to_branch_name(
125
self._mp.target_git_path.encode('utf-8'))
126
return urlutils.join_segment_parameters(
127
self._mp.target_git_repository.git_identity,
128
{"branch": branch_name})
132
return lp_api.canonical_url(self._mp)
135
return (self._mp.queue_status == 'Merged')
138
class Launchpad(Hoster):
139
"""The Launchpad hosting service."""
143
# https://bugs.launchpad.net/launchpad/+bug/397676
144
supports_merge_proposal_labels = False
146
def __init__(self, staging=False):
147
self._staging = staging
149
lp_instance = 'staging'
151
lp_instance = 'production'
152
self.launchpad = connect_launchpad(lp_instance)
155
return "Launchpad(staging=%s)" % self._staging
157
def hosts(self, branch):
158
# TODO(jelmer): staging vs non-staging?
159
return plausible_launchpad_url(branch.user_url)
162
def probe(cls, branch):
163
if plausible_launchpad_url(branch.user_url):
165
raise UnsupportedHoster(branch)
167
def _get_lp_git_ref_from_branch(self, branch):
168
url, params = urlutils.split_segment_parameters(branch.user_url)
169
(scheme, user, password, host, port, path) = urlutils.parse_url(
171
repo_lp = self.launchpad.git_repositories.getByPath(
172
path=path.strip('/'))
174
ref_path = params['ref']
176
branch_name = params.get('branch', branch.name)
178
ref_path = 'refs/heads/%s' % branch_name
180
ref_path = repo_lp.default_branch
181
ref_lp = repo_lp.getRefByPath(path=ref_path)
182
return (repo_lp, ref_lp)
184
def _get_lp_bzr_branch_from_branch(self, branch):
185
return self.launchpad.branches.getByUrl(
186
url=urlutils.unescape(branch.user_url))
188
def _get_derived_git_path(self, base_path, owner, project):
189
base_repo = self.launchpad.git_repositories.getByPath(path=base_path)
191
project = '/'.join(base_repo.unique_name.split('/')[1:])
192
# TODO(jelmer): Surely there is a better way of creating one of these
194
return "~%s/%s" % (owner, project)
196
def _publish_git(self, local_branch, base_path, name, owner, project=None,
197
revision_id=None, overwrite=False, allow_lossy=True):
198
to_path = self._get_derived_git_path(base_path, owner, project)
199
to_transport = get_transport("git+ssh://git.launchpad.net/" + to_path)
201
dir_to = controldir.ControlDir.open_from_transport(to_transport)
202
except errors.NotBranchError:
203
# Didn't find anything
208
br_to = local_branch.create_clone_on_transport(
209
to_transport, revision_id=revision_id, name=name)
210
except errors.NoRoundtrippingSupport:
211
br_to = local_branch.create_clone_on_transport(
212
to_transport, revision_id=revision_id, name=name,
216
dir_to = dir_to.push_branch(
217
local_branch, revision_id, overwrite=overwrite, name=name)
218
except errors.NoRoundtrippingSupport:
221
dir_to = dir_to.push_branch(
222
local_branch, revision_id, overwrite=overwrite, name=name,
224
br_to = dir_to.target_branch
226
"https://git.launchpad.net/%s/+ref/%s" % (to_path, name))
228
def _get_derived_bzr_path(self, base_branch, name, owner, project):
230
base_branch_lp = self._get_lp_bzr_branch_from_branch(base_branch)
231
project = '/'.join(base_branch_lp.unique_name.split('/')[1:-1])
232
# TODO(jelmer): Surely there is a better way of creating one of these
234
return "~%s/%s/%s" % (owner, project, name)
236
def get_push_url(self, branch):
237
(vcs, user, password, path, params) = self._split_url(branch.user_url)
239
branch_lp = self._get_lp_bzr_branch_from_branch(branch)
240
return branch_lp.bzr_identity
242
return urlutils.join_segment_parameters(
243
"git+ssh://git.launchpad.net/" + path, params)
247
def _publish_bzr(self, local_branch, base_branch, name, owner,
248
project=None, revision_id=None, overwrite=False,
250
to_path = self._get_derived_bzr_path(base_branch, name, owner, project)
251
to_transport = get_transport("lp:" + to_path)
253
dir_to = controldir.ControlDir.open_from_transport(to_transport)
254
except errors.NotBranchError:
255
# Didn't find anything
259
br_to = local_branch.create_clone_on_transport(
260
to_transport, revision_id=revision_id)
262
br_to = dir_to.push_branch(
263
local_branch, revision_id, overwrite=overwrite).target_branch
264
return br_to, ("https://code.launchpad.net/" + to_path)
266
def _split_url(self, url):
267
url, params = urlutils.split_segment_parameters(url)
268
(scheme, user, password, host, port, path) = urlutils.parse_url(url)
269
path = path.strip('/')
270
if host.startswith('bazaar.'):
272
elif host.startswith('git.'):
275
raise ValueError("unknown host %s" % host)
276
return (vcs, user, password, path, params)
278
def publish_derived(self, local_branch, base_branch, name, project=None,
279
owner=None, revision_id=None, overwrite=False,
281
"""Publish a branch to the site, derived from base_branch.
283
:param base_branch: branch to derive the new branch from
284
:param new_branch: branch to publish
285
:param name: Name of the new branch on the remote host
286
:param project: Optional project name
287
:param owner: Optional owner
288
:return: resulting branch
291
owner = self.launchpad.me.name
292
(base_vcs, base_user, base_password, base_path,
293
base_params) = self._split_url(base_branch.user_url)
294
# TODO(jelmer): Prevent publishing to development focus
295
if base_vcs == 'bzr':
296
return self._publish_bzr(
297
local_branch, base_branch, name, project=project, owner=owner,
298
revision_id=revision_id, overwrite=overwrite,
299
allow_lossy=allow_lossy)
300
elif base_vcs == 'git':
301
return self._publish_git(
302
local_branch, base_path, name, project=project, owner=owner,
303
revision_id=revision_id, overwrite=overwrite,
304
allow_lossy=allow_lossy)
306
raise AssertionError('not a valid Launchpad URL')
308
def get_derived_branch(self, base_branch, name, project=None, owner=None):
310
owner = self.launchpad.me.name
311
(base_vcs, base_user, base_password, base_path,
312
base_params) = self._split_url(base_branch.user_url)
313
if base_vcs == 'bzr':
314
to_path = self._get_derived_bzr_path(
315
base_branch, name, owner, project)
316
return _mod_branch.Branch.open("lp:" + to_path)
317
elif base_vcs == 'git':
318
to_path = self._get_derived_git_path(
319
base_path.strip('/'), owner, project)
320
return _mod_branch.Branch.open(
321
"git+ssh://git.launchpad.net/" + to_path, name)
323
raise AssertionError('not a valid Launchpad URL')
325
def iter_proposals(self, source_branch, target_branch, status='open'):
326
(base_vcs, base_user, base_password, base_path,
327
base_params) = self._split_url(target_branch.user_url)
328
statuses = status_to_lp_mp_statuses(status)
329
if base_vcs == 'bzr':
330
target_branch_lp = self.launchpad.branches.getByUrl(
331
url=target_branch.user_url)
332
source_branch_lp = self.launchpad.branches.getByUrl(
333
url=source_branch.user_url)
334
for mp in target_branch_lp.getMergeProposals(status=statuses):
335
if mp.source_branch_link != source_branch_lp.self_link:
337
yield LaunchpadMergeProposal(mp)
338
elif base_vcs == 'git':
339
(source_repo_lp, source_branch_lp) = (
340
self.lp_host._get_lp_git_ref_from_branch(source_branch))
341
(target_repo_lp, target_branch_lp) = (
342
self.lp_host._get_lp_git_ref_from_branch(target_branch))
343
for mp in target_branch_lp.getMergeProposals(status=statuses):
344
if (target_branch_lp.path != mp.target_git_path or
345
target_repo_lp != mp.target_git_repository or
346
source_branch_lp.path != mp.source_git_path or
347
source_repo_lp != mp.source_git_repository):
349
yield LaunchpadMergeProposal(mp)
351
raise AssertionError('not a valid Launchpad URL')
353
def get_proposer(self, source_branch, target_branch):
354
(base_vcs, base_user, base_password, base_path,
355
base_params) = self._split_url(target_branch.user_url)
356
if base_vcs == 'bzr':
357
return LaunchpadBazaarMergeProposalBuilder(
358
self, source_branch, target_branch)
359
elif base_vcs == 'git':
360
return LaunchpadGitMergeProposalBuilder(
361
self, source_branch, target_branch)
363
raise AssertionError('not a valid Launchpad URL')
366
def iter_instances(cls):
369
def iter_my_proposals(self, status='open'):
370
statuses = status_to_lp_mp_statuses(status)
371
for mp in self.launchpad.me.getMergeProposals(status=statuses):
372
yield LaunchpadMergeProposal(mp)
375
def connect_launchpad(lp_instance='production'):
376
service = lp_registration.LaunchpadService(lp_instance=lp_instance)
377
return lp_api.login(service, version='devel')
380
class LaunchpadBazaarMergeProposalBuilder(MergeProposalBuilder):
382
def __init__(self, lp_host, source_branch, target_branch, message=None,
383
staging=None, approve=None, fixes=None):
386
:param source_branch: The branch to propose for merging.
387
:param target_branch: The branch to merge into.
388
:param message: The commit message to use. (May be None.)
389
:param staging: If True, propose the merge against staging instead of
391
:param approve: If True, mark the new proposal as approved immediately.
392
This is useful when a project permits some things to be approved
393
by the submitter (e.g. merges between release and deployment
396
self.lp_host = lp_host
397
self.launchpad = lp_host.launchpad
398
self.source_branch = source_branch
399
self.source_branch_lp = self.launchpad.branches.getByUrl(
400
url=source_branch.user_url)
401
if target_branch is None:
402
self.target_branch_lp = self.source_branch_lp.get_target()
403
self.target_branch = _mod_branch.Branch.open(
404
self.target_branch_lp.bzr_identity)
406
self.target_branch = target_branch
407
self.target_branch_lp = self.launchpad.branches.getByUrl(
408
url=target_branch.user_url)
409
self.commit_message = message
410
self.approve = approve
413
def get_infotext(self):
414
"""Determine the initial comment for the merge proposal."""
415
if self.commit_message is not None:
416
return self.commit_message.strip().encode('utf-8')
417
info = ["Source: %s\n" % self.source_branch_lp.bzr_identity]
418
info.append("Target: %s\n" % self.target_branch_lp.bzr_identity)
421
def get_initial_body(self):
422
"""Get a body for the proposal for the user to modify.
424
:return: a str or None.
426
if not self.hooks['merge_proposal_body']:
429
def list_modified_files():
430
lca_tree = self.source_branch_lp.find_lca_tree(
431
self.target_branch_lp)
432
source_tree = self.source_branch.basis_tree()
433
files = modified_files(lca_tree, source_tree)
435
with self.target_branch.lock_read(), \
436
self.source_branch.lock_read():
438
for hook in self.hooks['merge_proposal_body']:
440
'target_branch': self.target_branch_lp.bzr_identity,
441
'modified_files_callback': list_modified_files,
446
def check_proposal(self):
447
"""Check that the submission is sensible."""
448
if self.source_branch_lp.self_link == self.target_branch_lp.self_link:
449
raise errors.BzrCommandError(
450
'Source and target branches must be different.')
451
for mp in self.source_branch_lp.landing_targets:
452
if mp.queue_status in ('Merged', 'Rejected'):
454
if mp.target_branch.self_link == self.target_branch_lp.self_link:
455
raise MergeProposalExists(lp_api.canonical_url(mp))
457
def approve_proposal(self, mp):
458
with self.source_branch.lock_read():
462
subject='', # Use the default subject.
463
content=u"Rubberstamp! Proposer approves of own proposal.")
464
_call_webservice(mp.setStatus, status=u'Approved',
465
revid=self.source_branch.last_revision())
467
def create_proposal(self, description, reviewers=None, labels=None,
468
prerequisite_branch=None):
469
"""Perform the submission."""
471
raise LabelsUnsupported()
472
if prerequisite_branch is not None:
473
prereq = self.launchpad.branches.getByUrl(
474
url=prerequisite_branch.user_url)
477
if reviewers is None:
480
mp = _call_webservice(
481
self.source_branch_lp.createMergeProposal,
482
target_branch=self.target_branch_lp,
483
prerequisite_branch=prereq,
484
initial_comment=description.strip(),
485
commit_message=self.commit_message,
486
reviewers=[self.launchpad.people[reviewer].self_link
487
for reviewer in reviewers],
488
review_types=[None for reviewer in reviewers])
489
except WebserviceFailure as e:
491
if (b'There is already a branch merge proposal '
492
b'registered for branch ') in e.message:
493
raise MergeProposalExists(self.source_branch.user_url)
497
self.approve_proposal(mp)
499
if self.fixes.startswith('lp:'):
500
self.fixes = self.fixes[3:]
503
bug=self.launchpad.bugs[int(self.fixes)])
504
return LaunchpadMergeProposal(mp)
507
class LaunchpadGitMergeProposalBuilder(MergeProposalBuilder):
509
def __init__(self, lp_host, source_branch, target_branch, message=None,
510
staging=None, approve=None, fixes=None):
513
:param source_branch: The branch to propose for merging.
514
:param target_branch: The branch to merge into.
515
:param message: The commit message to use. (May be None.)
516
:param staging: If True, propose the merge against staging instead of
518
:param approve: If True, mark the new proposal as approved immediately.
519
This is useful when a project permits some things to be approved
520
by the submitter (e.g. merges between release and deployment
523
self.lp_host = lp_host
524
self.launchpad = lp_host.launchpad
525
self.source_branch = source_branch
526
(self.source_repo_lp,
527
self.source_branch_lp) = self.lp_host._get_lp_git_ref_from_branch(
529
if target_branch is None:
530
self.target_branch_lp = self.source_branch.get_target()
531
self.target_branch = _mod_branch.Branch.open(
532
self.target_branch_lp.git_https_url)
534
self.target_branch = target_branch
535
(self.target_repo_lp, self.target_branch_lp) = (
536
self.lp_host._get_lp_git_ref_from_branch(target_branch))
537
self.commit_message = message
538
self.approve = approve
541
def get_infotext(self):
542
"""Determine the initial comment for the merge proposal."""
543
if self.commit_message is not None:
544
return self.commit_message.strip().encode('utf-8')
545
info = ["Source: %s\n" % self.source_branch.user_url]
546
info.append("Target: %s\n" % self.target_branch.user_url)
549
def get_initial_body(self):
550
"""Get a body for the proposal for the user to modify.
552
:return: a str or None.
554
if not self.hooks['merge_proposal_body']:
557
def list_modified_files():
558
lca_tree = self.source_branch_lp.find_lca_tree(
559
self.target_branch_lp)
560
source_tree = self.source_branch.basis_tree()
561
files = modified_files(lca_tree, source_tree)
563
with self.target_branch.lock_read(), \
564
self.source_branch.lock_read():
566
for hook in self.hooks['merge_proposal_body']:
568
'target_branch': self.target_branch,
569
'modified_files_callback': list_modified_files,
574
def check_proposal(self):
575
"""Check that the submission is sensible."""
576
if self.source_branch_lp.self_link == self.target_branch_lp.self_link:
577
raise errors.BzrCommandError(
578
'Source and target branches must be different.')
579
for mp in self.source_branch_lp.landing_targets:
580
if mp.queue_status in ('Merged', 'Rejected'):
582
if mp.target_branch.self_link == self.target_branch_lp.self_link:
583
raise MergeProposalExists(lp_api.canonical_url(mp))
585
def approve_proposal(self, mp):
586
with self.source_branch.lock_read():
590
subject='', # Use the default subject.
591
content=u"Rubberstamp! Proposer approves of own proposal.")
593
mp.setStatus, status=u'Approved',
594
revid=self.source_branch.last_revision())
596
def create_proposal(self, description, reviewers=None, labels=None,
597
prerequisite_branch=None):
598
"""Perform the submission."""
600
raise LabelsUnsupported()
601
if prerequisite_branch is not None:
602
(prereq_repo_lp, prereq_branch_lp) = (
603
self.lp_host._get_lp_git_ref_from_branch(prerequisite_branch))
605
prereq_branch_lp = None
606
if reviewers is None:
609
mp = _call_webservice(
610
self.source_branch_lp.createMergeProposal,
611
merge_target=self.target_branch_lp,
612
merge_prerequisite=prereq_branch_lp,
613
initial_comment=description.strip().encode('utf-8'),
614
commit_message=self.commit_message,
616
reviewers=[self.launchpad.people[reviewer].self_link
617
for reviewer in reviewers],
618
review_types=[None for reviewer in reviewers])
619
except WebserviceFailure as e:
621
if ('There is already a branch merge proposal '
622
'registered for branch ') in e.message:
623
raise MergeProposalExists(self.source_branch.user_url)
626
self.approve_proposal(mp)
628
if self.fixes.startswith('lp:'):
629
self.fixes = self.fixes[3:]
632
bug=self.launchpad.bugs[int(self.fixes)])
633
return LaunchpadMergeProposal(mp)
636
def modified_files(old_tree, new_tree):
637
"""Return a list of paths in the new tree with modified contents."""
638
for f, (op, path), c, v, p, n, (ok, k), e in new_tree.iter_changes(
640
if c and k == 'file':
added
removed
breezy/plugins/propose/launchpad.py
