* lpapiwrapper.py:

- Add _UbuntuSeries and _SourcePackage wrapper classes for proper type checking
* functions.py: Removed
  - Move the remaining functions to lpapiwrapper.py
* buildd, requestsync:
  - Updated for the new function location
This commit is contained in:
Michael Bienia 2009-07-02 13:05:16 +02:00
parent e139944b3a
commit a0ae5a2e23
4 changed files with 89 additions and 167 deletions

11
buildd
View File

@ -27,7 +27,6 @@ from optparse import OptionGroup
from optparse import OptionParser from optparse import OptionParser
from ubuntutools.lp.udtexceptions import SeriesNotFoundException, PackageNotFoundException from ubuntutools.lp.udtexceptions import SeriesNotFoundException, PackageNotFoundException
from ubuntutools.lp.lpapiwrapper import LpApiWrapper from ubuntutools.lp.lpapiwrapper import LpApiWrapper
import ubuntutools.lp.functions as lp_functions
# Usage. # Usage.
usage = "%prog <srcpackage> <release> <operation>\n\n" usage = "%prog <srcpackage> <release> <operation>\n\n"
@ -117,15 +116,13 @@ except (SeriesNotFoundException, PackageNotFoundException), e:
# Get list of builds for that package. # Get list of builds for that package.
builds = sources.getBuilds() builds = sources.getBuilds()
# Find out the version in given release. # Find out the version and component in given release.
version = sources.source_package_version version = sources.getVersion()
component = sources.getComponent()
# Get the component the package is in.
component = sources.component_name
# Operations that are remaining may only be done by Ubuntu developers (retry) # Operations that are remaining may only be done by Ubuntu developers (retry)
# or buildd admins (rescore). Check if the proper permissions are in place. # or buildd admins (rescore). Check if the proper permissions are in place.
if op == "rescore": necessaryPrivs = lp_functions.isLPTeamMember('launchpad-buildd-admins') if op == "rescore": necessaryPrivs = lpapiwrapper.isLpTeamMember('launchpad-buildd-admins')
if op == "retry": necessaryPrivs = lpapiwrapper.canUploadPackage(package, release) if op == "retry": necessaryPrivs = lpapiwrapper.canUploadPackage(package, release)
if op in ('rescore', 'retry') and not necessaryPrivs: if op in ('rescore', 'retry') and not necessaryPrivs:

View File

@ -36,7 +36,6 @@ from optparse import OptionParser
from time import sleep from time import sleep
# ubuntu-dev-tools modules. # ubuntu-dev-tools modules.
import ubuntutools.lp.functions as lp_functions
import ubuntutools.lp.libsupport as lp_libsupport import ubuntutools.lp.libsupport as lp_libsupport
import ubuntutools.lp.udtexceptions as udtexceptions import ubuntutools.lp.udtexceptions as udtexceptions
from ubuntutools.lp.lpapiwrapper import Launchpad, LpApiWrapper from ubuntutools.lp.lpapiwrapper import Launchpad, LpApiWrapper
@ -122,10 +121,8 @@ def checkExistingReports(package):
def cur_version_component(sourcepkg, release): def cur_version_component(sourcepkg, release):
try: try:
version = lp_functions.packageVersion(sourcepkg, release) src = LpApiWrapper.getUbuntuSourcePackage(sourcepkg, release)
component = lp_functions.packageComponent(sourcepkg, release) return (src.getVersion(), src.getComponent())
return (version, component)
except udtexceptions.PackageNotFoundException: except udtexceptions.PackageNotFoundException:
@ -363,7 +360,7 @@ def post_bug(source_package, subscribe, status, bugtitle, bugtext):
#newly created bugreports have one task #newly created bugreports have one task
task = bug.bug_tasks[0] task = bug.bug_tasks[0]
# Only members of ubuntu-bugcontrol can set importance # Only members of ubuntu-bugcontrol can set importance
if lp_functions.isLPTeamMember('ubuntu-bugcontrol'): if LpApiWrapper.isLpTeamMember('ubuntu-bugcontrol'):
task.importance = 'Wishlist' task.importance = 'Wishlist'
task.status = status task.status = status
task.lp_save() task.lp_save()
@ -552,7 +549,7 @@ if __name__ == '__main__':
# Check if they have a per-package upload permission. # Check if they have a per-package upload permission.
if lp_functions.isPerPackageUploader(srcpkg): if LpApiWrapper.isPerPackageUploader(srcpkg):
report += 'Note that I have per-package upload permissions for %s.\n\n' % srcpkg report += 'Note that I have per-package upload permissions for %s.\n\n' % srcpkg
uidx = base_ver.find('build') uidx = base_ver.find('build')

View File

@ -1,138 +0,0 @@
#
# functions.py - various Launchpad-related functions for the Ubuntu Developer
# Tools package
#
# Copyright (C) 2008, 2009 Jonathan Davies <jpds@ubuntu.com>
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 3
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# Please see the /usr/share/common-licenses/GPL file for the full text of
# the GNU General Public License license.
#
import urllib2
import sys
from udtexceptions import PackageNotFoundException, SeriesNotFoundException
from lpapiwrapper import Launchpad, LpApiWrapper
import launchpadlib
from re import findall
import warnings
# http://wiki.python.org/moin/PythonDecoratorLibrary#GeneratingDeprecationWarnings
def deprecated(func):
"""This is a decorator which can be used to mark functions
as deprecated. It will result in a warning being emitted
when the function is used."""
def new_func(*args, **kwargs):
warnings.warn("Call to deprecated function %s." % func.__name__, DeprecationWarning, 2)
return func(*args, **kwargs)
new_func.__name__ = func.__name__
new_func.__doc__ = func.__doc__
new_func.__dict__.update(func.__dict__)
return new_func
# Singleton to access LP API
launchpad = Launchpad
def _ubuntuSourcePackage(package, series, pocket = 'Release'):
""" Finds an Ubuntu source package on LP
returns LP API repr of the source package
If the package does not exist: raise PackageNotFoundException
"""
lpapiwrapper = LpApiWrapper()
return lpapiwrapper.getUbuntuSourcePackage(package, series, pocket)
def packageVersion(package, series=None):
""" Retrieves the version of a given source package in the current
development distroseries
returns unicode string repr of source package version
If the package does not exist: raise PackageNotFoundException
"""
if not series:
series = LpApiWrapper.getUbuntuDevelopmentSeries()
return _ubuntuSourcePackage(package, series).source_package_version
def packageComponent(package, series=None):
""" Retrieves the component for a given source package
returns unicode string representation of component
If the package does not exist: raise PackageNotFoundException
"""
if not series:
series = LpApiWrapper.getUbuntuDevelopmentSeries()
return _ubuntuSourcePackage(package, series).component_name
def canUploadPackage(package, series=None):
""" Checks whether the user can upload package to Ubuntu's main archive
Uses LP API to do this.
If the user can upload the package: return True.
If the user cannot upload the package: return False.
If the package does not exist: raise PackageNotFoundException
"""
if not series:
series = LpApiWrapper.getUbuntuDevelopmentSeries()
u_archive = LpApiWrapper.getUbuntuArchive()
uploaders = u_archive.getUploadersForComponent(component_name=packageComponent(package, series))
for permission in uploaders:
current_uploader = permission.person
if _findMember(current_uploader):
return True
return False
def _findMember(haystack):
""" Find a person in a haystack. A haystack can consist of either people or teams.
If the needle is in the haystack: return True
If the needle is not in the haystack: return False
"""
if not haystack.is_team:
return (str(haystack) == str(launchpad.me))
elif haystack.is_valid: # is a team
return isLPTeamMember(haystack.name)
return False
def isLPTeamMember(team):
""" Checks if the user is a member of a certain team on Launchpad.
Uses the LP API.
If the user is a member of the team: return True.
If the user is not a member of the team: return False.
"""
return any(t.name == team for t in launchpad.me.super_teams)
def isPerPackageUploader(package):
# Checks if the user has upload privileges for a certain package.
me = findall('~(\S+)', '%s' % launchpad.me)[0]
main_archive = LpApiWrapper.getUbuntuArchive()
try:
perms = main_archive.getUploadersForPackage(source_package_name=package)
except launchpadlib.errors.HTTPError:
return False
for perm in perms:
if perm.person.name == me:
return True

View File

@ -109,8 +109,8 @@ class LpApiWrapper(object):
try: try:
series = cls.getUbuntuDistribution().getSeries(name_or_version = name_or_version) series = cls.getUbuntuDistribution().getSeries(name_or_version = name_or_version)
# Cache with name and version # Cache with name and version
cls._series[series.name] = series cls._series[series.name] = _UbuntuSeries(series)
cls._series[series.version] = series cls._series[series.version] = _UbuntuSeries(series)
except HTTPError: except HTTPError:
raise SeriesNotFoundException("Error: Unknown Ubuntu release: '%s'." % name_or_version) raise SeriesNotFoundException("Error: Unknown Ubuntu release: '%s'." % name_or_version)
@ -125,11 +125,11 @@ class LpApiWrapper(object):
if not cls._devel_series: if not cls._devel_series:
dev = cls.getUbuntuDistribution().current_series dev = cls.getUbuntuDistribution().current_series
cls._devel_series = dev cls._devel_series = _UbuntuSeries(dev)
# Cache it in _series if not already done # Cache it in _series if not already done
if dev.name not in cls._series: if dev.name not in cls._series:
cls._series[dev.name] = dev cls._series[dev.name] = cls._devel_series
cls._series[dev.version] = dev cls._series[dev.version] = cls._devel_series
return cls._devel_series return cls._devel_series
@ -138,7 +138,7 @@ class LpApiWrapper(object):
''' '''
Finds an Ubuntu source package on LP. Finds an Ubuntu source package on LP.
Returns LP representation of the source package. Returns a wrapped LP representation of the source package.
If the package does not exist: raise PackageNotFoundException If the package does not exist: raise PackageNotFoundException
''' '''
@ -147,7 +147,7 @@ class LpApiWrapper(object):
raise PocketDoesNotExist("Pocket '%s' does not exist." % pocket) raise PocketDoesNotExist("Pocket '%s' does not exist." % pocket)
# Check if we have already a LP representation of an Ubuntu series or not # Check if we have already a LP representation of an Ubuntu series or not
if not isinstance(series, Entry): if not isinstance(series, _UbuntuSeries):
series = cls.getUbuntuSeries(str(series)) series = cls.getUbuntuSeries(str(series))
if (name, series, pocket) not in cls._src_pkg: if (name, series, pocket) not in cls._src_pkg:
@ -155,7 +155,7 @@ class LpApiWrapper(object):
srcpkg = cls.getUbuntuArchive().getPublishedSources( srcpkg = cls.getUbuntuArchive().getPublishedSources(
source_name = name, distro_series = series, pocket = pocket, source_name = name, distro_series = series, pocket = pocket,
status = 'Published', exact_match = True)[0] status = 'Published', exact_match = True)[0]
cls._src_pkg[(name, series, pocket)] = srcpkg cls._src_pkg[(name, series, pocket)] = _SourcePackage(srcpkg)
except IndexError: except IndexError:
if pocket == 'Release': if pocket == 'Release':
msg = "The package '%s' does not exist in the Ubuntu main archive in '%s'" % \ msg = "The package '%s' does not exist in the Ubuntu main archive in '%s'" % \
@ -175,21 +175,21 @@ class LpApiWrapper(object):
for package either through component upload rights or for package either through component upload rights or
per-package upload rights. per-package upload rights.
'package' can either be a LP representation of a source package 'package' can either be a wrapped LP representation of a source
or a string and an Ubuntu series. If 'package' doesn't exist package or a string and an Ubuntu series. If 'package' doesn't
yet in Ubuntu assume 'universe' for component. exist yet in Ubuntu assume 'universe' for component.
''' '''
if isinstance(package, Entry): if isinstance(package, _SourcePackage):
component = package.component_name component = package.getComponent()
package = package.source_package_name package = package.getPackageName()
else: else:
if not series: if not series:
# Fall-back to current Ubuntu development series # Fall-back to current Ubuntu development series
series = cls.getUbuntuDevelopmentSeries() series = cls.getUbuntuDevelopmentSeries()
try: try:
component = cls.getUbuntuSourcePackage(package, series).component_name component = cls.getUbuntuSourcePackage(package, series).getComponent()
except PackageNotFoundException: except PackageNotFoundException:
# Probably a new package, assume "universe" as component # Probably a new package, assume "universe" as component
component = 'universe' component = 'universe'
@ -211,3 +211,69 @@ class LpApiWrapper(object):
return cls._upload_comp[component] return cls._upload_comp[component]
else: else:
return cls._upload_pkg[package] return cls._upload_pkg[package]
# TODO: check if this is still needed after ArchiveReorg (or at all)
@classmethod
def isPerPackageUploader(cls, package, series = None):
'''
Check if the user has PerPackageUpload rights for package.
'''
if isinstance(package, _SourcePackage):
pkg = package.getPackageName()
else:
pkg = package
return cls.canUploadPackage(package, series) and pkg in cls._upload_pkg
@classmethod
def isLpTeamMember(cls, team):
'''
Checks if the user is a member of a certain team on Launchpad.
Returns True if the user is a member of the team otherwise False.
'''
return any(t.name == team for t in cls.getMe().super_teams)
class _UbuntuSeries(object):
'''
Wrapper class around a LP Ubuntu series object.
'''
def __init__(self, series):
if isinstance(series, Entry) and series.resource_type_link == 'https://api.edge.launchpad.net/beta/#distro_series':
self._series = series
else:
raise TypeError('A LP API Ubuntu series representation expected.')
def __getattr__(self, attr):
return getattr(self._series, attr)
class _SourcePackage(object):
'''
Wrapper class around a LP source package object.
'''
def __init__(self, srcpkg):
if isinstance(srcpkg, Entry) and srcpkg.resource_type_link == 'https://api.edge.launchpad.net/beta/#source_package_publishing_history':
self._srcpkg = srcpkg
else:
raise TypeError('A LP API source package representation expected.')
def getPackageName(self):
'''
Returns the source package name.
'''
return self._srcpkg.source_package_name
def getVersion(self):
'''
Returns the version of the source package.
'''
return self._srcpkg.source_package_version
def getComponent(self):
'''
Returns the component of the source package.
'''
return self._srcpkg.component_name
def __getattr__(self, attr):
return getattr(self._srcpkg, attr)