93
102
lp_instance = service._lp_instance
95
return LAUNCHPAD_API_URLS[lp_instance]
104
return lookup_service_root(lp_instance)
97
106
raise InvalidLaunchpadInstance(lp_instance)
100
def login(service, timeout=None, proxy_info=None):
109
class NoLaunchpadBranch(errors.BzrError):
110
_fmt = 'No launchpad branch could be found for branch "%(url)s".'
112
def __init__(self, branch):
113
errors.BzrError.__init__(self, branch=branch, url=branch.base)
116
def login(service, timeout=None, proxy_info=None,
117
version=Launchpad.DEFAULT_VERSION):
101
118
"""Log in to the Launchpad API.
103
120
:return: The root `Launchpad` object from launchpadlib.
122
if proxy_info is None:
123
proxy_info = httplib2.proxy_info_from_environment('https')
105
124
cache_directory = get_cache_directory()
106
125
launchpad = Launchpad.login_with(
107
126
'bzr', _get_api_url(service), cache_directory, timeout=timeout,
108
proxy_info=proxy_info)
109
# XXX: Work-around a minor security bug in launchpadlib 1.5.1, which would
110
# create this directory with default umask.
111
os.chmod(cache_directory, 0700)
127
proxy_info=proxy_info, version=version)
128
# XXX: Work-around a minor security bug in launchpadlib < 1.6.3, which
129
# would create this directory with default umask.
130
osutils.chmod_if_possible(cache_directory, 0700)
115
def load_branch(launchpad, branch):
116
"""Return the launchpadlib Branch object corresponding to 'branch'.
118
:param launchpad: The root `Launchpad` object from launchpadlib.
119
:param branch: A `bzrlib.branch.Branch`.
120
:raise NotLaunchpadBranch: If we cannot determine the Launchpad URL of
122
:return: A launchpadlib Branch object.
124
# XXX: This duplicates the "What are possible URLs for the branch that
125
# Launchpad might recognize" logic found in cmd_lp_open.
127
# XXX: This makes multiple roundtrips to Launchpad for what is
128
# conceptually a single operation -- get me the branches that match these
129
# URLs. Unfortunately, Launchpad's support for such operations is poor, so
130
# we have to allow multiple roundtrips.
131
for url in branch.get_public_branch(), branch.get_push_location():
134
class LaunchpadBranch(object):
135
"""Provide bzr and lp API access to a Launchpad branch."""
137
def __init__(self, lp_branch, bzr_url, bzr_branch=None, check_update=True):
140
:param lp_branch: The Launchpad branch.
141
:param bzr_url: The URL of the Bazaar branch.
142
:param bzr_branch: An instance of the Bazaar branch.
144
self.bzr_url = bzr_url
145
self._bzr = bzr_branch
146
self._push_bzr = None
147
self._check_update = check_update
152
"""Return the bzr branch for this branch."""
153
if self._bzr is None:
154
self._bzr = branch.Branch.open(self.bzr_url)
159
"""Return the push branch for this branch."""
160
if self._push_bzr is None:
161
self._push_bzr = branch.Branch.open(self.lp.bzr_identity)
162
return self._push_bzr
165
def plausible_launchpad_url(url):
166
"""Is 'url' something that could conceivably be pushed to LP?
168
:param url: A URL that may refer to a Launchpad branch.
173
if url.startswith('lp:'):
175
regex = re.compile('([a-z]*\+)*(bzr\+ssh|http)'
176
'://bazaar.*.launchpad.net')
177
return bool(regex.match(url))
180
def candidate_urls(bzr_branch):
181
"""Iterate through related URLs that might be Launchpad URLs.
183
:param bzr_branch: A Bazaar branch to find URLs from.
184
:return: a generator of URL strings.
186
url = bzr_branch.get_public_branch()
189
url = bzr_branch.get_push_location()
192
url = bzr_branch.get_parent()
195
yield bzr_branch.base
198
def tweak_url(url, launchpad):
199
"""Adjust a URL to work with staging, if needed."""
200
if str(launchpad._root_uri) == STAGING_SERVICE_ROOT:
201
return url.replace('bazaar.launchpad.net',
202
'bazaar.staging.launchpad.net')
203
elif str(launchpad._root_uri) == lookup_service_root('qastaging'):
204
return url.replace('bazaar.launchpad.net',
205
'bazaar.qastaging.launchpad.net')
209
def from_bzr(cls, launchpad, bzr_branch, create_missing=True):
210
"""Find a Launchpad branch from a bzr branch."""
212
for url in cls.candidate_urls(bzr_branch):
213
url = cls.tweak_url(url, launchpad)
214
if not cls.plausible_launchpad_url(url):
216
lp_branch = launchpad.branches.getByUrl(url=url)
217
if lp_branch is not None:
220
if not create_missing:
221
raise NoLaunchpadBranch(bzr_branch)
222
lp_branch = cls.create_now(launchpad, bzr_branch)
224
return cls(lp_branch, bzr_branch.base, bzr_branch, check_update)
227
def create_now(cls, launchpad, bzr_branch):
228
"""Create a Bazaar branch on Launchpad for the supplied branch."""
229
url = cls.tweak_url(bzr_branch.get_push_location(), launchpad)
230
if not cls.plausible_launchpad_url(url):
231
raise errors.BzrError(gettext('%s is not registered on Launchpad') %
233
bzr_branch.create_clone_on_transport(transport.get_transport(url))
132
234
lp_branch = launchpad.branches.getByUrl(url=url)
135
raise NotLaunchpadBranch(url)
235
if lp_branch is None:
236
raise errors.BzrError(gettext('%s is not registered on Launchpad') %
240
def get_target(self):
241
"""Return the 'LaunchpadBranch' for the target of this one."""
243
if lp_branch.project is not None:
244
dev_focus = lp_branch.project.development_focus
245
if dev_focus is None:
246
raise errors.BzrError(gettext('%s has no development focus.') %
247
lp_branch.bzr_identity)
248
target = dev_focus.branch
250
raise errors.BzrError(gettext(
251
'development focus %s has no branch.') % dev_focus)
252
elif lp_branch.sourcepackage is not None:
253
target = lp_branch.sourcepackage.getBranch(pocket="Release")
255
raise errors.BzrError(gettext(
256
'source package %s has no branch.') %
257
lp_branch.sourcepackage)
259
raise errors.BzrError(gettext(
260
'%s has no associated product or source package.') %
261
lp_branch.bzr_identity)
262
return LaunchpadBranch(target, target.bzr_identity)
265
"""Update the Launchpad copy of this branch."""
266
if not self._check_update:
270
if self.lp.last_scanned_id is not None:
271
if self.bzr.last_revision() == self.lp.last_scanned_id:
272
trace.note(gettext('%s is already up-to-date.') %
273
self.lp.bzr_identity)
275
graph = self.bzr.repository.get_graph()
276
if not graph.is_ancestor(self.lp.last_scanned_id,
277
self.bzr.last_revision()):
278
raise errors.DivergedBranches(self.bzr, self.push_bzr)
279
trace.note(gettext('Pushing to %s') % self.lp.bzr_identity)
280
self.bzr.push(self.push_bzr)
284
def find_lca_tree(self, other):
285
"""Find the revision tree for the LCA of this branch and other.
287
:param other: Another LaunchpadBranch
288
:return: The RevisionTree of the LCA of this branch and other.
290
graph = self.bzr.repository.get_graph(other.bzr.repository)
291
lca = graph.find_unique_lca(self.bzr.last_revision(),
292
other.bzr.last_revision())
293
return self.bzr.repository.revision_tree(lca)
296
def canonical_url(object):
297
"""Return the canonical URL for a branch."""
298
scheme, netloc, path, params, query, fragment = urlparse.urlparse(
299
str(object.self_link))
300
path = '/'.join(path.split('/')[2:])
301
netloc = netloc.replace('api.', 'code.')
302
return urlparse.urlunparse((scheme, netloc, path, params, query,