Merge with my development branch:

* ubuntutools/requestsync/{lp,mail}.py: Replace the relative imports with
  absolute ones (follow PEP8)
* Add support for the other LP service roots (edge is still default)
* Also check package sets for upload permissions.
* ubuntutools/lp/lpapicache.py: Add support for anonymous login (currently
  unused).
* ubuntutools/lp/lpapicache.py: Turn PersonTeam.getMe() into a class property
  (PersonTeam.me).
* requestsync: Add an error message when Ubuntu has a newer version than
  Debian.
This commit is contained in:
Michael Bienia 2010-02-20 18:46:21 +01:00
commit 3df7282fc5
11 changed files with 137 additions and 102 deletions

5
debian/changelog vendored
View File

@ -27,12 +27,15 @@ ubuntu-dev-tools (0.93) UNRELEASED; urgency=low
[ Michael Bienia ]
* ubuntutools/requestsync/mail.py: Encode the report to utf-8 before passing
it to gpg for signing (lp: #522316).
* Add support for the other LP service roots (edge is still default)
* Depend on python-launchpadlib >= 1.5.4
* Also check package sets for upload permissions.
[ Michael Vogt ]
* edit-patch: add wrapper around cdbs-edit-patch, dpatch-edit-patch, quilt
to transparently deal with the various patch systems.
-- Kees Cook <kees@ubuntu.com> Thu, 18 Feb 2010 11:10:45 -0800
-- Michael Bienia <geser@ubuntu.com> Sat, 20 Feb 2010 18:38:17 +0100
ubuntu-dev-tools (0.92) lucid; urgency=low

2
debian/control vendored
View File

@ -13,7 +13,7 @@ Standards-Version: 3.8.3
Package: ubuntu-dev-tools
Architecture: all
Depends: ${python:Depends}, ${misc:Depends}, binutils, devscripts, sudo,
python-debian, python-launchpadlib, dctrl-tools, lsb-release, diffstat,
python-debian, python-launchpadlib (>= 1.5.4), dctrl-tools, lsb-release, diffstat,
dpkg-dev, python-apt (>= 0.7.9), python-lazr.restfulclient
Recommends: bzr, pbuilder | cowdancer | sbuild, reportbug (>= 3.39ubuntu1),
ca-certificates, debootstrap, genisoimage, perl-modules, libwww-perl,

View File

@ -22,6 +22,7 @@
import os
import sys
from optparse import OptionParser, make_option
from launchpadlib.uris import lookup_service_root
from ubuntutools.lp.libsupport import *
class CmdOptions(OptionParser):
@ -77,7 +78,7 @@ class CmdOptions(OptionParser):
optional_options = given_options - set(sum(self.TOOLS[tool], ()))
if optional_options:
self.error("The following options are not allowed for this tool: %s" %", ".join(optional_options))
options.service = translate_service(options.service)
options.service = lookup_service_root(options.service)
if options.level in LEVEL:
options.level = LEVEL[options.level]
elif options.level.upper() in LEVEL.values():

View File

@ -78,7 +78,7 @@ if __name__ == '__main__':
# import the needed requestsync module
if options.lpapi:
from ubuntutools.requestsync.lp import *
from ubuntutools.lp.lpapicache import Distribution, PersonTeam
from ubuntutools.lp.lpapicache import Distribution
# See if we have LP credentials and exit if we don't - cannot continue in this case
try:
Launchpad.login()
@ -140,15 +140,19 @@ if __name__ == '__main__':
print >> sys.stderr, "E: %s" % e
sys.exit(1)
# Debian and Ubuntu versions are the same - stop
# Stop if Ubuntu has already the version from Debian or a newer version
if ubuntu_version == debian_version:
print >> sys.stderr, \
'E: The versions in Debian and Ubuntu are the same already (%s). Aborting.' % ubuntu_version
sys.exit(1)
if ubuntu_version > debian_version:
print >> sys.stderr, \
'E: The version in Ubuntu (%s) is newer than the version in Debian (%s). Aborting.' % (ubuntu_version, debian_version)
sys.exit(1)
# -s flag not specified - check if we do need sponsorship
if not sponsorship:
sponsorship = needSponsorship(srcpkg, ubuntu_component)
sponsorship = needSponsorship(srcpkg, ubuntu_component, release)
# Check for existing package reports
if not newsource:
@ -183,12 +187,6 @@ if __name__ == '__main__':
if need_interaction:
raw_input_exit_on_ctrlc('Press [Enter] to continue. Press [Ctrl-C] to abort now. ')
# Check if they have a per-package upload permission.
if lpapi:
ubuntu_archive = Distribution('ubuntu').getArchive()
if PersonTeam.getMe().isPerPackageUploader(ubuntu_archive, srcpkg):
report += 'Note that I have per-package upload permissions for %s.\n\n' % srcpkg
base_version = force_base_version or ubuntu_version
if newsource:

View File

@ -132,6 +132,7 @@ if not options.batch:
# Get list of published sources for package in question.
try:
sources = ubuntu_archive.getSourcePackage(package, release, pocket)
distroseries = Distribution('ubuntu').getSeries(release)
except (SeriesNotFoundException, PackageNotFoundException), e:
print e
sys.exit(1)
@ -144,10 +145,10 @@ if not options.batch:
# Operations that are remaining may only be done by Ubuntu developers (retry)
# or buildd admins (rescore). Check if the proper permissions are in place.
me = PersonTeam.getMe()
me = PersonTeam.me
if op == "rescore": necessaryPrivs = me.isLpTeamMember('launchpad-buildd-admins')
if op == "retry": necessaryPrivs = me.canUploadPackage(
ubuntu_archive, sources.getPackageName(), sources.getComponent())
ubuntu_archive, distroseries, sources.getPackageName(), sources.getComponent())
if op in ('rescore', 'retry') and not necessaryPrivs:
print >> sys.stderr, "You cannot perform the %s operation on a %s package " \
@ -215,7 +216,7 @@ if release and '-' in release:
sys.exit(1)
ubuntu_archive = Distribution('ubuntu').getArchive()
me = PersonTeam.getMe()
me = PersonTeam.me
# Check permisions (part 1): Rescoring can only be done by buildd admins
can_rescore = options.priority and me.isLpTeamMember('launchpad-buildd-admins') or False

View File

@ -1,3 +1,5 @@
##
## ubuntu-dev-tools Launchpad Python modules.
##
service = 'edge'

View File

@ -28,7 +28,7 @@ import httplib2
try:
from launchpadlib.credentials import Credentials
from launchpadlib.launchpad import Launchpad, STAGING_SERVICE_ROOT, EDGE_SERVICE_ROOT
from launchpadlib.launchpad import Launchpad
from launchpadlib.errors import HTTPError
except ImportError:
print "Unable to import launchpadlib module, is python-launchpadlib installed?"
@ -37,6 +37,8 @@ except:
Credentials = None
Launchpad = None
from ubuntutools.lp import service
def find_credentials(consumer, files, level=None):
""" search for credentials matching 'consumer' in path for given access level. """
if Credentials is None:
@ -73,7 +75,7 @@ def get_credentials(consumer, cred_file=None, level=None):
return find_credentials(consumer, files, level)
def get_launchpad(consumer, server=EDGE_SERVICE_ROOT, cache=None,
def get_launchpad(consumer, server=service, cache=None,
cred_file=None, level=None):
credentials = get_credentials(consumer, cred_file, level)
cache = cache or os.environ.get("LPCACHE", None)
@ -140,14 +142,3 @@ def approve_application(credentials, email, password, level, web_root,
raise HTTPError(response, content)
credentials.exchange_request_token_for_access_token(web_root)
return credentials
def translate_service(service):
_service = service.lower()
if _service in (STAGING_SERVICE_ROOT, EDGE_SERVICE_ROOT):
return _service
elif _service == "edge":
return EDGE_SERVICE_ROOT
elif _service == "staging":
return STAGING_SERVICE_ROOT
else:
raise ValueError("unknown service '%s'" %service)

View File

@ -3,7 +3,7 @@
# lpapicache.py - wrapper classes around the LP API implementing caching
# for usage in the ubuntu-dev-tools package
#
# Copyright © 2009 Michael Bienia <geser@ubuntu.com>
# Copyright © 2009-2010 Michael Bienia <geser@ubuntu.com>
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
@ -25,34 +25,55 @@
#httplib2.debuglevel = 1
import sys
import libsupport
import launchpadlib.launchpad as launchpad
from launchpadlib.errors import HTTPError
from launchpadlib.uris import lookup_service_root
from lazr.restfulclient.resource import Entry
from udtexceptions import *
import ubuntutools.lp.libsupport as libsupport
from ubuntutools.lp import service
from ubuntutools.lp.udtexceptions import *
__all__ = [
'Archive',
'Build',
'Distribution',
'DistributionSourcePackage',
'DistroSeries',
'Launchpad',
'PersonTeam',
'SourcePackagePublishingHistory',
]
class Launchpad(object):
''' Singleton for LP API access. '''
__lp = None
'''Singleton for LP API access.'''
def login(self):
'''
Enforce a login through the LP API.
'''
if not self.__lp:
try:
self.__lp = libsupport.get_launchpad('ubuntu-dev-tools')
except IOError, error:
print >> sys.stderr, 'E: %s' % error
raise error
return self
def login(self):
'''Enforce a non-anonymous login.'''
if '_Launchpad__lp' not in self.__dict__:
try:
self.__lp = libsupport.get_launchpad('ubuntu-dev-tools')
except IOError, error:
print >> sys.stderr, 'E: %s' % error
raise
else:
raise AlreadyLoggedInError('Already logged in to Launchpad.')
def __getattr__(self, attr):
if not self.__lp:
self.login()
return getattr(self.__lp, attr)
def login_anonymously(self):
'''Enforce an anonymous login.'''
if '_Launchpad__lp' not in self.__dict__:
self.__lp = launchpad.Launchpad.login_anonymously('ubuntu-dev-tools', service)
else:
raise AlreadyLoggedInError('Already logged in to Launchpad.')
def __call__(self):
return self
def __getattr__(self, attr):
if '_Launchpad__lp' not in self.__dict__:
self.login()
return getattr(self.__lp, attr)
def __call__(self):
return self
Launchpad = Launchpad()
@ -63,7 +84,7 @@ class MetaWrapper(type):
def __init__(cls, name, bases, attrd):
super(MetaWrapper, cls).__init__(name, bases, attrd)
if 'resource_type' not in attrd:
raise TypeError('Class needs an associated resource type')
raise TypeError('Class "%s" needs an associated resource type' % name)
cls._cache = dict()
@ -75,7 +96,7 @@ class BaseWrapper(object):
resource_type = None # it's a base class after all
def __new__(cls, data):
if isinstance(data, basestring) and data.startswith('https://api.edge.launchpad.net/beta/'):
if isinstance(data, basestring) and data.startswith(lookup_service_root(service) + 'beta/'):
# looks like a LP API URL
# check if it's already cached
cached = cls._cache.get(data)
@ -120,12 +141,17 @@ class BaseWrapper(object):
def __getattr__(self, attr):
return getattr(self._lpobject, attr)
def __repr__(self):
if hasattr(str, 'format'):
return '<{0}: {1!r}>'.format(self.__class__.__name__, self._lpobject)
else:
return '<%s: %r>' % (self.__class__.__name__, self._lpobject)
class Distribution(BaseWrapper):
'''
Wrapper class around a LP distribution object.
'''
resource_type = 'https://api.edge.launchpad.net/beta/#distribution'
resource_type = lookup_service_root(service) + 'beta/#distribution'
def __init__(self, *args):
# Don't share _series and _archives between different Distributions
@ -207,14 +233,14 @@ class DistroSeries(BaseWrapper):
'''
Wrapper class around a LP distro series object.
'''
resource_type = 'https://api.edge.launchpad.net/beta/#distro_series'
resource_type = lookup_service_root(service) + 'beta/#distro_series'
class Archive(BaseWrapper):
'''
Wrapper class around a LP archive object.
'''
resource_type = 'https://api.edge.launchpad.net/beta/#archive'
resource_type = lookup_service_root(service) + 'beta/#archive'
def __init__(self, *args):
# Don't share _srcpkgs between different Archives
@ -275,7 +301,7 @@ class SourcePackagePublishingHistory(BaseWrapper):
'''
Wrapper class around a LP source package object.
'''
resource_type = 'https://api.edge.launchpad.net/beta/#source_package_publishing_history'
resource_type = lookup_service_root(service) + 'beta/#source_package_publishing_history'
def __init__(self, *args):
# Don't share _builds between different SourcePackagePublishingHistory objects
@ -352,13 +378,33 @@ class SourcePackagePublishingHistory(BaseWrapper):
self.getPackageName(), '\n'.join(res))
class MetaPersonTeam(MetaWrapper):
@property
def me(cls):
'''The PersonTeam object of the currently authenticated LP user or
None when anonymously logged in.
'''
if '_me' not in cls.__dict__:
try:
cls._me = PersonTeam(Launchpad.me)
except HTTPError, error:
if error.response.status == 401:
# Anonymous login
cls._me = None
else:
raise
return cls._me
class PersonTeam(BaseWrapper):
'''
Wrapper class around a LP person or team object.
'''
resource_type = ('https://api.edge.launchpad.net/beta/#person', 'https://api.edge.launchpad.net/beta/#team')
__metaclass__ = MetaPersonTeam
_me = None # the PersonTeam object of the currently authenticated LP user
resource_type = (
lookup_service_root(service) + 'beta/#person',
lookup_service_root(service) + 'beta/#team',
)
def __init__(self, *args):
# Don't share _upload_{pkg,comp} between different PersonTeams
@ -385,15 +431,6 @@ class PersonTeam(BaseWrapper):
cached = PersonTeam(Launchpad.people[person_or_team])
return cached
@classmethod
def getMe(cls):
'''
Returns a PersonTeam object of the currently authenticated LP user.
'''
if not cls._me:
cls._me = PersonTeam(Launchpad.me)
return cls._me
def isLpTeamMember(self, team):
'''
Checks if the user is a member of a certain team on Launchpad.
@ -402,29 +439,38 @@ class PersonTeam(BaseWrapper):
'''
return any(t.name == team for t in self.super_teams)
def canUploadPackage(self, archive, package, component):
'''
Check if the person or team has upload rights for the source package
to the specified 'archive' either through component upload
rights or per-package upload rights.
def canUploadPackage(self, archive, distroseries, package, component):
'''Check if the person or team has upload rights for the source
package to the specified 'archive' and 'distrorelease' either
through package sets, component or or per-package upload rights.
Either a source package name or a component has the specified.
'archive' has to be a Archive object.
'distroseries' has to be an DistroSeries object.
'''
if not isinstance(archive, Archive):
raise TypeError("'%r' is not an Archive object." % archive)
if package and not isinstance(package, basestring):
if not isinstance(distroseries, DistroSeries):
raise TypeError("'%r' is not a DistroSeries object." % distroseries)
if package is not None and not isinstance(package, basestring):
raise TypeError('A source package name expected.')
if component and not isinstance(component, basestring):
if component is not None and not isinstance(component, basestring):
raise TypeError('A component name expected.')
if not package and not component:
if package is None and component is None:
raise ValueError('Either a source package name or a component has to be specified.')
upload_comp = self._upload_comp.get((archive, component))
upload_pkg = self._upload_pkg.get((archive, package))
if upload_comp == None and upload_pkg == None:
for perm in archive.getPermissionsForPerson(person = self()):
if upload_comp is None and upload_pkg is None:
# archive.isSourceUploadAllowed() checks only package sets permission
if package is not None and archive.isSourceUploadAllowed(
distroseries=distroseries(), person=self(), sourcepackagename=package):
# TODO: also cache the release it applies to
self._upload_pkg[(archive, package)] = True
return True
# check for component or per-package upload rights
for perm in archive.getPermissionsForPerson(person=self()):
if perm.permission != 'Archive Upload Rights':
continue
if component and perm.component_name == component:
@ -442,24 +488,12 @@ class PersonTeam(BaseWrapper):
else:
return upload_comp or upload_pkg
# TODO: check if this is still needed after ArchiveReorg (or at all)
def isPerPackageUploader(self, archive, package):
'''
Check if the user has PerPackageUpload rights for package.
'''
if isinstance(package, SourcePackagePublishingHistory):
pkg = package.getPackageName()
else:
pkg = package
return self.canUploadPackage(archive, pkg, None)
class Build(BaseWrapper):
'''
Wrapper class around a build object.
'''
resource_type = 'https://api.edge.launchpad.net/beta/#build'
resource_type = lookup_service_root(service) + 'beta/#build'
def __str__(self):
return u'%s: %s' % (self.arch_tag, self.buildstate)
@ -481,4 +515,4 @@ class DistributionSourcePackage(BaseWrapper):
'''
Caching class for distribution_source_package objects.
'''
resource_type = 'https://api.edge.launchpad.net/beta/#distribution_source_package'
resource_type = lookup_service_root(service) + 'beta/#distribution_source_package'

View File

@ -13,3 +13,7 @@ class PocketDoesNotExistException(BaseException):
class ArchiveNotFoundException(BaseException):
""" Thrown when an archive for a distibution is not found """
pass
class AlreadyLoggedInError(Exception):
'''Raised when a second login is attempted.'''
pass

View File

@ -20,10 +20,10 @@
# Please see the /usr/share/common-licenses/GPL-2 file for the full text
# of the GNU General Public License license.
from .common import raw_input_exit_on_ctrlc
from ..lp.lpapicache import Launchpad, Distribution, PersonTeam, DistributionSourcePackage
from ..lp.udtexceptions import PackageNotFoundException, SeriesNotFoundException, PocketDoesNotExistException, ArchiveNotFoundException
from ..lp.libsupport import translate_api_web
from ubuntutools.requestsync.common import raw_input_exit_on_ctrlc
from ubuntutools.lp.lpapicache import Launchpad, Distribution, PersonTeam, DistributionSourcePackage
from ubuntutools.lp.udtexceptions import PackageNotFoundException, SeriesNotFoundException, PocketDoesNotExistException, ArchiveNotFoundException
from ubuntutools.lp.libsupport import translate_api_web
def getDebianSrcPkg(name, release):
debian = Distribution('debian')
@ -44,14 +44,15 @@ def getUbuntuSrcPkg(name, release):
return ubuntu_archive.getSourcePackage(name, release)
def needSponsorship(name, component):
def needSponsorship(name, component, release):
'''
Check if the user has upload permissions for either the package
itself or the component
'''
archive = Distribution('ubuntu').getArchive()
distroseries = Distribution('ubuntu').getSeries(release)
need_sponsor = not PersonTeam.getMe().canUploadPackage(archive, name, component)
need_sponsor = not PersonTeam.me.canUploadPackage(archive, distroseries, name, component)
if need_sponsor:
print '''You are not able to upload this package directly to Ubuntu.
Your sync request shall require an approval by a member of the appropriate
@ -104,7 +105,7 @@ def postBug(srcpkg, subscribe, status, bugtitle, bugtext):
# newly created bugreports have only one task
task = bug.bug_tasks[0]
# only members of ubuntu-bugcontrol can set importance
if PersonTeam.getMe().isLpTeamMember('ubuntu-bugcontrol'):
if PersonTeam.me.isLpTeamMember('ubuntu-bugcontrol'):
task.importance = 'Wishlist'
task.status = status
task.lp_save()

View File

@ -25,8 +25,8 @@ import subprocess
import smtplib
import socket
from debian_bundle.changelog import Version
from .common import raw_input_exit_on_ctrlc
from ..lp.udtexceptions import PackageNotFoundException
from ubuntutools.requestsync.common import raw_input_exit_on_ctrlc
from ubuntutools.lp.udtexceptions import PackageNotFoundException
__all__ = [
'getDebianSrcPkg',
@ -118,7 +118,7 @@ def getEmailAddress():
'mail the sync request.'
return myemailaddr
def needSponsorship(name, component):
def needSponsorship(name, component, release):
'''
Ask the user if he has upload permissions for the package or the
component.
@ -126,7 +126,7 @@ def needSponsorship(name, component):
while True:
print "Do you have upload permissions for the '%s' component " \
"or the package '%s'?" % (component, name)
"or the package '%s' in Ubuntu %s?" % (component, name, release)
val = raw_input_exit_on_ctrlc("If in doubt answer 'n'. [y/N]? ")
if val.lower() in ('y', 'yes'):
return False