Merge in lplib changes from thekorn.

This commit is contained in:
Jonathan Davies 2009-01-13 20:25:16 +00:00
commit 2270f133cb
9 changed files with 378 additions and 137 deletions

118
common.py
View File

@ -31,6 +31,16 @@ import re
import subprocess
import sys
import urllib2
import urlparse
import urllib
try:
import httplib2
from launchpadlib.credentials import Credentials
from launchpadlib.launchpad import Launchpad, STAGING_SERVICE_ROOT, EDGE_SERVICE_ROOT
from launchpadlib.errors import HTTPError
except ImportError:
Credentials = None
Launchpad = None
# Clear https_proxy env var as it's not supported in urllib/urllib2; see
# LP #122551
@ -265,3 +275,111 @@ def packageComponent(package, release):
component = rel.split('/')[1]
return component.strip()
def find_credentials(consumer, files, level=None):
""" search for credentials matching 'consumer' in path for given access level. """
if Credentials is None:
raise ImportError
for f in files:
cred = Credentials()
try:
cred.load(open(f))
except:
continue
if cred.consumer.key == consumer:
return cred
raise IOError("No credentials found")
def get_credentials(consumer, cred_file=None, level=None):
files = list()
if cred_file:
files.append(cred_file)
if "LPCREDENTIALS" in os.environ:
files.append(os.environ["LPCREDENTIALS"])
files.extend([
os.path.join(os.getcwd(), "lp_credentials.txt"),
os.path.expanduser("~/lp_credentials.txt"),
])
return find_credentials(consumer, files, level)
def get_launchpad(consumer, server=STAGING_SERVICE_ROOT, cache=None,
cred_file=None, level=None):
credentials = get_credentials(consumer, cred_file, level)
cache = cache or os.environ.get("LPCACHE", None)
return Launchpad(credentials, server, cache)
def query_to_dict(query_string):
result = dict()
options = filter(None, query_string.split("&"))
for opt in options:
key, value = opt.split("=")
result.setdefault(key, set()).add(value)
return result
def translate_web_api(url, launchpad):
scheme, netloc, path, query, fragment = urlparse.urlsplit(url)
query = query_to_dict(query)
if not (("edge" in netloc and "edge" in str(launchpad._root_uri))
or ("staging" in netloc and "staging" in str(launchpad._root_uri))):
raise ValueError("url conflict (url: %s, root: %s" %(url, launchpad._root_uri))
if path.endswith("/+bugs"):
path = path[:-6]
if "ws.op" in query:
raise ValueError("Invalid web url, url: %s" %url)
query["ws.op"] = "searchTasks"
scheme, netloc, api_path, _, _ = urlparse.urlsplit(str(launchpad._root_uri))
query = urllib.urlencode(query)
url = urlparse.urlunsplit((scheme, netloc, api_path + path.lstrip("/"), query, fragment))
return url
def translate_api_web(self_url):
return self_url.replace("api.", "").replace("beta/", "")
LEVEL = {
0: "UNAUTHORIZED",
1: "READ_PUBLIC",
2: "WRITE_PUBLIC",
3: "READ_PRIVATE",
4: "WRITE_PRIVATE"
}
def approve_application(credentials, email, password, level, web_root,
context):
authorization_url = credentials.get_request_token(context, web_root)
if level in LEVEL:
level = 'field.actions.%s' %LEVEL[level]
elif level in LEVEL.values():
level = 'field.actions.%s' %level
elif str(level).startswith("field.actions") and str(level).split(".")[-1] in LEVEL:
pass
else:
raise ValueError("Unknown access level '%s'" %level)
params = {level: 1,
"oauth_token": credentials._request_token.key,
"lp.context": context or ""}
lp_creds = ":".join((email, password))
basic_auth = "Basic %s" %(lp_creds.encode('base64'))
headers = {'Authorization': basic_auth}
response, content = httplib2.Http().request(authorization_url,
method="POST", body=urllib.urlencode(params), headers=headers)
if int(response["status"]) != 200:
if not 300 <= int(response["status"]) <= 400: # this means redirection
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 == "staging":
return STAGING_SERVICE_ROOT
elif _service == "edge":
return EDGE_SERVICE_ROOT
else:
raise ValueError("unknown service '%s'" %service)

8
debian/changelog vendored
View File

@ -37,6 +37,14 @@ ubuntu-dev-tools (0.52) UNRELEASED; urgency=low
* requestsync:
- Catch AssertionError exception if rmadison returns with an error.
[ Markus Korn ]
* Added manage-credentials, a tool to create (and manage) credentials
which are used to access launchpad via the API.
* Ported: hugdaylist, massfile, grab-attachment and requestsync to
launchpadlib.
* Other misc. fixes and tweaks.
* Install common.py to correct location with py_modules.
-- Luca Falavigna <dktrkranz@ubuntu.com> Mon, 12 Jan 2009 20:28:01 +0100
ubuntu-dev-tools (0.51) jaunty; urgency=low

4
debian/control vendored
View File

@ -14,8 +14,8 @@ Package: ubuntu-dev-tools
Architecture: all
Section: devel
Depends: ${python:Depends}, binutils, devscripts, sudo, python-debian,
python-launchpad-bugs (>= 0.2.25), dctrl-tools, lsb-release, diffstat,
dpkg-dev, ${misc:Depends}
python-launchpad-bugs (>= 0.2.25), python-launchpadlib, dctrl-tools,
sb-release, diffstat, dpkg-dev, ${misc:Depends}
Recommends: bzr, pbuilder | cowdancer | sbuild, reportbug (>= 3.39ubuntu1),
ca-certificates, genisoimage, perl-modules, libwww-perl
Conflicts: devscripts (<< 2.10.7ubuntu5)

View File

@ -20,8 +20,7 @@
import os
import sys
import urllib
import launchpadbugs.connector as Connector
from common import get_launchpad
USAGE = "grab-attachments <bug numbers>"
@ -33,17 +32,21 @@ def main():
if sys.argv[1] in ["--help", "-h"]:
print USAGE
sys.exit(0)
Bug = Connector.ConnectBug(method="Text")
launchpad = get_launchpad("ubuntu-dev-tools")
for arg in sys.argv[1:]:
try:
number = int(arg)
except:
print >> sys.stderr, "'%s' is not a valid bug number." % arg
sys.exit(1)
b = Bug(number)
b = launchpad.bugs[number]
for a in b.attachments:
filename = os.path.join(os.getcwd(), a.url.split("/")[-1])
urllib.urlretrieve(a.url, filename)
f = a.data.open()
filename = os.path.join(os.getcwd(), f.filename)
local_file = open(filename, "w")
local_file.write(f.read())
f.close()
local_file.close()
if __name__ == '__main__':
main()

View File

@ -35,14 +35,7 @@ import string
import sys
from optparse import OptionParser
try:
import launchpadbugs.connector as Connector
BugList = Connector.ConnectBugList()
Bug = Connector.ConnectBug(method="Text")
except ImportError:
print >> sys.stderr, \
"python-launchpad-bugs (>= 0.2.25) needs to be installed to use hugdaylist."
sys.exit(1)
from common import get_launchpad, translate_web_api, translate_api_web
def check_args():
howmany = -1
@ -74,26 +67,40 @@ def check_args():
return (howmany, url)
def filter_unsolved(b):
bug = Bug(int(b))
def filter_unsolved(task):
# TODO: don't use this filter here, only check status and assignee of
# the given task
# Filter out special types of bugs:
# - https://wiki.ubuntu.com/Bugs/HowToTriage#Special%20types%20of%20bugs
return filter(lambda a: a.status != 'Fix Committed' and \
(a.assignee in ['motu','desktop-bugs'] or \
not a.assignee), bug.infotable) and \
'ubuntu-main-sponsors' not in [str(s) for s in bug.subscribers] and \
'ubuntu-universe-sponsors' not in [str(s) for s in bug.subscribers] and \
'ubuntu-archive' not in [str(s) for s in bug.subscribers]
subscriptions = set(s.person.name for s in task.bug.subscriptions) #this is expensive, parse name out of self_link instead?
if (task.status != "Fix Committed" and
(not task.assignee or task.assignee.name in ['motu','desktop-bugs']) and
'ubuntu-main-sponsors' not in subscriptions and
'ubuntu-universe-sponsors' not in subscriptions and
'ubuntu-archive' not in subscriptions):
return True
return False
def main():
(howmany, url) = check_args()
try:
bl = BugList(url)
except:
print >> sys.stderr, "The URL at '%s' does not appear to have a bug " \
"list." % url
if len(url.split("?", 1)) == 2:
# search options not supported, because there is no mapping web ui options <-> API options
print >> sys.stderr, "Options in url are not supported, url: %s" %url
sys.exit(1)
launchpad = get_launchpad("ubuntu-dev-tools")
api_url = translate_web_api(url, launchpad)
try:
product = launchpad.load(api_url)
except Exception, e:
x = getattr(e, "response", {})
if response.get("status", None) == "404":
print >> sys.stderr, "The URL at '%s' does not appear to be a valid url to a product" %url
sys.exit(1)
else:
raise
bl = product.searchTasks()
l = filter(filter_unsolved, bl)
@ -112,13 +119,14 @@ def main():
|| Bug || Subject || Triager ||"""
for i in list(l)[:howmany]:
bug = i.bug
print '||<rowbgcolor="#FFEBBB"> [%s %s] || %s || ||' % \
(i.url, i.bugnumber, i.summary)
(translate_api_web(bug.self_link), bug.id, bug.title)
if __name__ == '__main__':
try:
main()
main()
except KeyboardInterrupt:
print >> sys.stderr, "Aborted."
sys.exit(1)

123
manage-credentials Executable file
View File

@ -0,0 +1,123 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2009 Markus Korn <thekorn@gmx.de>
#
# ##################################################################
#
# 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.
#
# See file /usr/share/common-licenses/GPL for more details.
#
# ##################################################################
import sys
from optparse import OptionParser, make_option
from common import Credentials, Launchpad, translate_service
from common import LEVEL, translate_api_web, approve_application
class CmdOptions(OptionParser):
USAGE = (
"\t%prog create -c <consumer> [--email <email> --password <password>] [--service <staging|edge>]\n"
)
OPTIONS = (
make_option("-c", "--consumer", action="store", type="string",
dest="consumer", default=None),
make_option("-e", "--email", action="store", type="string",
dest="email", default=None),
make_option("-p", "--password", action="store", type="string",
dest="password", default=None),
make_option("-s", "--service", action="store", type="string",
dest="service", default="staging"),
make_option("--cache", action="store", type="string",
dest="cache", default=None),
make_option("-o", action="store", type="string",
dest="output", default=None),
make_option("-l", "--level", action="store", type="int",
dest="level", default=0,
help="integer representing the access-level (default: 0), mappping: %s" %LEVEL),
)
TOOLS = {
"create": ( ("consumer",),
("email", "password", "service", "cache", "output",
"level")),
"list": (tuple(), ("service", )),
}
def __init__(self):
OptionParser.__init__(self, option_list=self.OPTIONS)
self.set_usage(self.USAGE)
def parse_args(self, args=None, values=None):
options, args = OptionParser.parse_args(self, args, values)
given_options = set(i for i, k in self.defaults.iteritems() if not getattr(options, i) == k)
if not args:
self.error("Please define a sub-tool you would like to use")
if not len(args) == 1:
self.error("Only one sub-tool allowed")
else:
tool = args.pop()
if not tool in self.TOOLS:
self.error("Unknown tool '%s'" %tool)
needed_options = set(self.TOOLS[tool][0]) - given_options
if needed_options:
self.error("please define the following options: %s" %", ".join(needed_options))
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)
if options.level in LEVEL:
options.level = LEVEL[options.level]
elif options.level.upper() in LEVEL.values():
options.level = options.level.upper()
else:
self.error("unknown access-level '%s', level must be in %s" %(options.level, self.LEVEL))
return tool, options
def create_credentials(options):
if options.password and options.email:
# use hack
credentials = Credentials(options.consumer)
credentials = approve_application(credentials, options.email,
options.password, options.level,
translate_api_web(options.service), None)
else:
launchpad = Launchpad.get_token_and_login(options.consumer,
options.service, options.cache)
credentials = launchpad.credentials
if options.output:
f = file(options.output, "w")
else:
f = sys.stdout
credentials.save(f)
return
def list_tokens(options):
print "Not implemented yet"
print "To get a list of your tokens, please visit %speople/+me/+oauth-tokens" %translate_api_web(options.service)
return 1
def main():
cmdoptions = CmdOptions()
tool, options = cmdoptions.parse_args()
if tool == "create":
return create_credentials(options)
elif tool == "list":
return list_tokens(options)
if __name__ == "__main__":
sys.exit(main())

View File

@ -27,12 +27,7 @@ import email
import subprocess
import glob
import launchpadbugs.connector as Connector
sys.path.append('/usr/share/ubuntu-dev-tools/')
import common
cookie = common.prepareLaunchpadCookie()
from common import get_launchpad, translate_api_web, translate_web_api
def read_config():
instructions_file = open("instructions")
@ -57,10 +52,6 @@ def read_list():
listfile.close()
return pack_list
def file_bug():
Bug = Connector.ConnectBug()
Bug.authentication = cookie
def check_configfiles():
result = True
@ -87,43 +78,62 @@ def check_configfiles():
def file_bug(config):
Bug = Connector.ConnectBug()
Bug.authentication = cookie
launchpad = get_launchpad("ubuntu-dev-tools")
try:
summary = config["subject"].replace("$pack", config["sourcepackage"])
description = config["text"].replace("$pack", config["sourcepackage"])
bug = Bug.New(product={"name": config["sourcepackage"],
"target": "ubuntu"},
summary=summary,
description=description)
print "Successfully filed bug %s: https://launchpad.net/bugs/%s" % \
(bug.bugnumber, bug.bugnumber)
for sub in config["subscribers"].split(","):
if sub.strip("\n").strip():
bug.subscribers.add(sub.strip("\n").strip())
for tag in config["tags"].split(","):
if tag.strip("\n").strip():
bug.tags.append(tag.strip("\n").strip())
bug.assignee = config["assignee"]
product_url = "%subuntu/+source/%s" %(launchpad._root_uri, config["sourcepackage"])
tags = filter(None, map(lambda t: t.strip("\n").strip(), config["tags"].split(",")))
bug = launchpad.bugs.createBug(description=description, title=summary,
target=product_url, tags=tags)
print "Successfully filed bug %i: %s" %(bug.id, translate_api_web(bug.self_link))
subscribers = filter(None, map(lambda t: t.strip("\n").strip(), config["subscribers"].split(",")))
for sub in subscribers:
subscribe_url = "%s~%s" %(launchpad._root_uri, sub)
bug.subscribe(person=subscribe_url)
#newly created bugreports have one task
task = bug.bug_tasks[0]
if config["status"]:
bug.status = config["status"].capitalize()
status = config["status"].capitalize()
else:
bug.status = "Confirmed"
bug.commit()
status = "Confirmed"
task.transitionToStatus(status=status)
assignee = config["assignee"]
if assignee:
assignee_url = "%s~%s" %(launchpad._root_uri, assignee)
bug.transitionToAssignee(assignee=assignee_url)
except:
"Bug for '%s' was not filed." % config["sourcepackage"]
def read_buglist(url):
BugList = Connector.ConnectBugList()
if not url:
return set()
if len(url.split("?", 1)) == 2:
# search options not supported, because there is no mapping web ui options <-> API options
print >> sys.stderr, "Options in url are not supported, url: %s" %url
sys.exit(1)
launchpad = get_launchpad("ubuntu-dev-tools")
packages = set()
if url:
buglist = BugList(url)
for bug in buglist.bugs:
packages.add(bug.sourcepackage)
api_url = translate_web_api(url, launchpad)
# workaround LP #303414
# if this is fixed it should simply be: buglist = launchpad.load(api_url)
api_url = api_url.split("?", 1)[0]
project = launchpad.load(api_url)
buglist = project.searchTasks()
for bug in buglist:
packages.add(bug.bug_target_name)
return packages

View File

@ -34,8 +34,6 @@ from debian_bundle.changelog import Version
from optparse import OptionParser
from time import sleep
# Use functions from ubuntu-dev-tools to create Launchpad cookie file.
sys.path.append('/usr/share/ubuntu-dev-tools/')
import common
launchpad_cookiefile = common.prepareLaunchpadCookie()
@ -54,6 +52,11 @@ def checkNeedsSponsorship(component):
The prepareLaunchpadCookie function above shall ensure that a cookie
file exists first.
"""
# TODO: use launchpadlib here
# Once LP: #313233 has been fixed this can be implemented by either:
# >>> me = launchpad.me
# >>> me.inTeam(<TEAM>) #or
# >>> me in <TEAM>
urlopener = common.setupLaunchpadUrlOpener(launchpad_cookiefile)
# Check where the package is and assign the appropriate variables.
@ -85,52 +88,38 @@ def checkNeedsSponsorship(component):
# Is a team member, no sponsorship required.
return False
def checkExistingReports(package, isNewPackage):
def checkExistingReports(package):
""" Check existing bug reports on Launchpad for a possible sync request.
If found ask for confirmation on filing a request.
"""
# Determine if the package is new or not.
if isNewPackage:
# Package not in Ubuntu, check Ubuntu Archive Team's bug page for
# possible duplicate reports.
bugListUrl = "https://bugs.launchpad.net/~ubuntu-archive"
else:
# Package in Ubuntu, check the package's bug list for duplicate reports.
bugListUrl = "https://bugs.launchpad.net/ubuntu/+source/%s" % package
try:
import launchpadbugs.connector as Connector
except:
# Failed to import launchpadbugs - skip check.
print >> sys.stderr, "Unable to import launchpadbugs. Is " \
"python-launchpad-bugs installed?"
launchpad = common.get_launchpad("ubuntu-dev-tools")
except ImportError:
print >> sys.stderr, 'Importing launchpadlib failed. Is ' \
'python-launchpadlib installed?'
print >> sys.stderr, "Skipping existing report check, you should "\
"manually check at:"
print "-", bugListUrl
return
print "- https://bugs.launchpad.net/ubuntu/+source/%s" % package
return False
# Connect to the bug list.
bugList = Connector.ConnectBugList()
# Fetch data from Launchpad.
pkgBugList = bugList(bugListUrl)
if len(pkgBugList) == 0:
return # No bugs found.
# Fetch the package's bug list from Launchpad.
pkg = launchpad.distributions["ubuntu"].getSourcePackage(name=package)
pkgBugList = pkg.searchTasks()
# Search bug list for other sync requests.
matchingBugs = [bug for bug in pkgBugList if "Please sync %s" %
package in bug.summary]
package in bug.title]
if len(matchingBugs) == 0:
return # No sync bugs found.
print "The following bugs could possibly be duplicate sync request(s) on Launchpad:"
print "The following bugs could be possible duplicate sync bug(s) on Launchpad:"
for bug in matchingBugs:
print " *", bug.summary
print " -", bug.url
print " *", bug.title
print " -", common.translate_api_web(bug.self_link)
print "Please check the above URLs to verify this before filing a " \
"possible duplicate report."
@ -143,11 +132,7 @@ def cur_version_component(sourcepkg, release):
madison = subprocess.Popen(['rmadison', '-u', 'ubuntu', '-a', 'source', \
'-s', release, sourcepkg], stdout=subprocess.PIPE)
out = madison.communicate()[0]
try:
assert (madison.returncode == 0)
except AssertionError:
print out
sys.exit(1)
assert (madison.returncode == 0)
for l in out.splitlines():
(pkg, version, rel, builds) = l.split('|')
@ -321,7 +306,7 @@ def mail_bug(source_package, subscribe, status, bugtitle, bugtext, keyid = None)
(mailserver, mailserver_port, s[1], s[0])
print "The port %s may be firewalled. Please try using requestsync with" \
% mailserver_port
print "the '--lp' flag to file a sync request with the launchpadbugs " \
print "the '--lp' flag to file a sync request with the launchpadlib " \
"module."
return False
@ -347,24 +332,23 @@ def mail_bug(source_package, subscribe, status, bugtitle, bugtext, keyid = None)
return True
def post_bug(source_package, subscribe, status, bugtitle, bugtext):
'''Use python-launchpad-bugs to submit the sync request.
'''Use python-launchpadlib to submit the sync request.
Return True if successfully posted, otherwise False.'''
import glob, os.path
try:
import launchpadbugs.connector
from launchpadbugs.lpconstants import HTTPCONNECTION
launchpad = common.get_launchpad("ubuntu-dev-tools")
except ImportError:
print >> sys.stderr, 'Importing launchpadbugs failed. Is python-launchpad-bugs installed?'
print >> sys.stderr, 'Importing launchpadlib failed. Is python-launchpadlib installed?'
return False
if source_package:
product = {'name': source_package, 'target': 'ubuntu'}
product_url = "%subuntu/+source/%s" %(launchpad._root_uri, source_package)
else:
# new source package
product = {'name': 'ubuntu'}
product_url = "%subuntu" %launchpad._root_uri
in_confirm_loop = True
while in_confirm_loop:
print 'Summary:\n%s\n\nDescription:\n%s' % (bugtitle, bugtext)
@ -383,32 +367,17 @@ def post_bug(source_package, subscribe, status, bugtitle, bugtext):
print "Invalid answer."
# Create bug
Bug = launchpadbugs.connector.ConnectBug()
# Force the usage of stable Launchpad.
Bug.set_connection_mode(HTTPCONNECTION.MODE.STABLE)
# Use our cookie file for authentication.
Bug.authentication = launchpad_cookiefile
print "Using LP cookie at: %s for authentication." % launchpad_cookiefile
# Submit bug report.
bug = Bug.New(product = product, summary = bugtitle, description = bugtext)
sleep(2) # Wait in case of slow Launchpad.
try:
bug.importance = 'Wishlist'
except IOError, s:
print "Warning: setting importance failed: %s" % s
bug = launchpad.bugs.createBug(description=bugtext, title=bugtitle, target=product_url)
bug.status = status
bug.subscriptions.add(subscribe)
sleep(1) # Wait.
bug.commit()
sleep(1) # Wait.
print 'Sync request filed as bug #%i: https://launchpad.net/bugs/%i' % (bug.bugnumber, bug.bugnumber)
#newly created bugreports have one task
task = bug.bug_tasks[0]
task.transitionToImportance(importance='Wishlist')
task.transitionToStatus(status=status)
subscribe_url = "%s~%s" %(launchpad._root_uri, subscribe)
bug.subscribe(person=subscribe_url)
print 'Sync request filed as bug #%i: %s' % (bug.id, common.translate_api_web(bug.self_link))
return True
def edit_report(subject, body, changes_required=False):
@ -481,7 +450,7 @@ if __name__ == '__main__':
help = "Whether package to sync is a new package in Ubuntu.")
optParser.add_option("--lp", action = "store_true",
dest = "lpbugs", default = False,
help = "Specify whether to use the launchpadbugs module for filing " \
help = "Specify whether to use the launchpadlib module for filing " \
"report.")
optParser.add_option("-s", action = "store_true",
dest = "sponsor", default = False,
@ -528,8 +497,8 @@ if __name__ == '__main__':
# -s flag not specified - check if we do need sponsorship.
if not sponsorship: sponsorship = checkNeedsSponsorship(component)
# Check for existing sync requests.
checkExistingReports(srcpkg, newsource)
# Check for existing package reports.
if not newsource: checkExistingReports(srcpkg)
# Generate bug report.
subscribe = 'ubuntu-archive'

View File

@ -37,7 +37,9 @@ setup(name='ubuntu-dev-tools',
'suspicious-source',
'ubuntu-iso',
'update-maintainer',
'what-patch',
'what-patch',
'manage-credentials',
],
packages=['ubuntutools'],
py_modules = ['common'],
)