yudaike/openSUSE-release-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #266 from lnussel/master

Browse Source 
check script that makes sure packages are submitted to Factory
This commit is contained in:
Stephan Kulow 2014-11-25 15:04:35 +01:00
commit 48dd4eff6f
11 changed files with 1228 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
README.asciidoc
View File

@ -3,12 +3,10 @@ Factory workflow plugins
image:https://travis-ci.org/openSUSE/osc-plugin-factory.png?branch=master["Build Status", link="https://travis-ci.org/openSUSE/osc-plugin-factory"] image:https://coveralls.io/repos/openSUSE/osc-plugin-factory/badge.png?branch=master["Coverage Status", link="https://coveralls.io/r/openSUSE/osc-plugin-factory"]
This repository contains different OSC plugins that help and support
the maintenance of Factory. These plugins use the OBS API to check,
evaluate and manage the different submit request of packages that go
from a devel project to Factory.
This repository contains various OSC plugins and scripts used for the
maintenance of openSUSE distributions like Factory.
Plugins
Scripts
-------
* *link:docs/staging.asciidoc[Staging].* Plugin used to manage the
@ -19,15 +17,21 @@ Plugins
continuously for usual mistakes or problems in requests like, for
example, new dependency cycles or errors in the binary RPM.
* *link:docs/checksource.asciidoc[CheckSource].* Check the source
version of the RPM package.
* *link:docs/checksource.asciidoc[CheckSource].* This plugin checks for usual
mistakes and problems in the source packages submitted by the users.
* *link:docs/totest.asciidoc[ToTest].* A plugin that help the
publishing process of Factory.
* *link:docs/totest.asciidoc[ToTest].* A plugin that checks if Factory is ready
to be released as Tumbleweed.
* *link:docs/checkdups.asciidoc[CheckDups].* Plugin to detect
superseded or duplicated requests.
* *link:docs/factory-source.asciidoc[factory-source].* script that checks if the
submitted sources of a request are already accepted in Factory
* *link:docs/maintbot.asciidoc[maintbot].* script that checks maintenance
incidents to make sure the Factory maintainer submitted the package.
Installation
------------

281
ReviewBot.py Normal file
View File

@ -0,0 +1,281 @@
#!/usr/bin/python
# Copyright (c) 2014 SUSE Linux Products GmbH
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
from pprint import pprint
import os, sys, re
import logging
from optparse import OptionParser
import cmdln
try:
from xml.etree import cElementTree as ET
except ImportError:
import cElementTree as ET
import osc.conf
import osc.core
import urllib2
class ReviewBot(object):
"""
A generic obs request reviewer
Inherit from this class and implement check functions for each action type:
def check_action_<type>(self, req, action):
return (None|True|False)
"""
def __init__(self, apiurl = None, dryrun = False, logger = None, user = None):
self.apiurl = apiurl
self.dryrun = dryrun
self.logger = logger
self.review_user = user
self.requests = []
self.review_messages = { 'accepted' : 'ok', 'declined': 'review failed' }
def set_request_ids(self, ids):
for rqid in ids:
u = osc.core.makeurl(self.apiurl, [ 'request', rqid ], { 'withhistory' : '1' })
r = osc.core.http_GET(u)
root = ET.parse(r).getroot()
req = osc.core.Request()
req.read(root)
self.requests.append(req)
def check_requests(self):
for req in self.requests:
self.logger.debug("checking %s"%req.reqid)
good = self.check_one_request(req)
if good is None:
self.logger.info("ignoring")
elif good:
self.logger.info("%s is good"%req.reqid)
self._set_review(req, 'accepted')
else:
self.logger.info("%s is not acceptable"%req.reqid)
self._set_review(req, 'declined')
def _set_review(self, req, state):
if not self.review_user:
return
review_state = self.get_review_state(req.reqid, self.review_user)
if review_state == 'new':
self.logger.debug("setting %s to %s"%(req.reqid, state))
if not self.dryrun:
msg = self.review_messages[state] if state in self.review_messages else state
osc.core.change_review_state(apiurl = self.apiurl,
reqid = req.reqid, newstate = state,
by_user=self.review_user, message=msg)
elif review_state == '':
self.logger.info("can't change state, %s does not have '%s' as reviewer"%(req.reqid, self.review_user))
else:
self.logger.debug("%s review in state '%s' not changed"%(req.reqid, review_state))
def add_review(self, req, by_group=None, by_user=None, by_project = None, by_package = None, msg=None):
query = {
'cmd': 'addreview'
}
if by_group:
query['by_group'] = by_group
elif by_user:
query['by_user'] = by_user
elif by_project:
query['by_project'] = by_project
if by_package:
query['by_package'] = by_package
else:
raise osc.oscerr.WrongArgs("missing by_*")
u = osc.core.makeurl(self.apiurl, ['request', req.reqid], query)
if self.dryrun:
self.logger.info('POST %s' % u)
return True
try:
r = osc.core.http_POST(u, data=msg)
except urllib2.HTTPError, e:
self.logger.error(e)
return False
code = ET.parse(r).getroot().attrib['code']
if code != 'ok':
self.logger.error("invalid return code %s"%code)
return False
return True
def check_one_request(self, req):
"""
check all actions in one request.
calls helper functions for each action type
return None if nothing to do, True to accept, False to reject
"""
overall = None
for a in req.actions:
fn = 'check_action_%s'%a.type
if not hasattr(self, fn):
self.logger.error("unhandled request type %s"%a.type)
ret = None
else:
func = getattr(self, fn)
ret = func(req, a)
if ret == False or overall is None and ret is not None:
overall = ret
return overall
# XXX used in other modules
def _get_verifymd5(self, src_project, src_package, rev=None):
query = { 'view': 'info' }
if rev:
query['rev'] = rev
url = osc.core.makeurl(self.apiurl, ('source', src_project, src_package), query=query)
try:
root = ET.parse(osc.core.http_GET(url)).getroot()
except urllib2.HTTPError:
return None
if root is not None:
srcmd5 = root.get('verifymd5')
return srcmd5
# TODO: what if there is more than _link?
def _get_linktarget_self(self, src_project, src_package):
""" if it's a link to a package in the same project return the name of the package"""
prj, pkg = self._get_linktarget(src_project, src_package)
if prj is None or prj == src_project:
return pkg
def _get_linktarget(self, src_project, src_package):
query = {}
url = osc.core.makeurl(self.apiurl, ('source', src_project, src_package), query=query)
try:
root = ET.parse(osc.core.http_GET(url)).getroot()
except urllib2.HTTPError:
return (None, None)
if root is not None:
linkinfo = root.find("linkinfo")
if linkinfo is not None:
return (linkinfo.get('project'), linkinfo.get('package'))
return (None, None)
# XXX used in other modules
def get_review_state(self, request_id, user):
"""Return the current review state of the request."""
states = []
url = osc.core.makeurl(self.apiurl, ('request', str(request_id)))
try:
root = ET.parse(osc.core.http_GET(url)).getroot()
states = [review.get('state') for review in root.findall('review') if review.get('by_user') == user]
except urllib2.HTTPError, e:
print('ERROR in URL %s [%s]' % (url, e))
return states[0] if states else ''
def set_request_ids_search_review(self, user = None):
if user is None:
user = self.review_user
review = "@by_user='%s'+and+@state='new'"%user
url = osc.core.makeurl(self.apiurl, ('search', 'request'), "match=state/@name='review'+and+review[%s]&withhistory=1"%review)
root = ET.parse(osc.core.http_GET(url)).getroot()
for request in root.findall('request'):
req = osc.core.Request()
req.read(request)
self.requests.append(req)
class CommandLineInterface(cmdln.Cmdln):
def __init__(self, *args, **kwargs):
cmdln.Cmdln.__init__(self, args, kwargs)
def get_optparser(self):
parser = cmdln.CmdlnOptionParser(self)
parser.add_option("--apiurl", '-A', metavar="URL", help="api url")
parser.add_option("--user", metavar="USER", help="reviewer user name")
parser.add_option("--dry", action="store_true", help="dry run")
parser.add_option("--debug", action="store_true", help="debug output")
parser.add_option("--osc-debug", action="store_true", help="osc debug output")
parser.add_option("--verbose", action="store_true", help="verbose")
return parser
def postoptparse(self):
logging.basicConfig()
self.logger = logging.getLogger(self.optparser.prog)
if (self.options.debug):
self.logger.setLevel(logging.DEBUG)
elif (self.options.verbose):
self.logger.setLevel(logging.INFO)
osc.conf.get_config(override_apiurl = self.options.apiurl)
if (self.options.osc_debug):
osc.conf.config['debug'] = 1
self.checker = self.setup_checker()
def setup_checker(self):
""" reimplement this """
apiurl = osc.conf.config['apiurl']
if apiurl is None:
raise osc.oscerr.ConfigError("missing apiurl")
user = self.options.user
if user is None:
user = osc.conf.get_apiurl_usr(apiurl)
return ReviewBot(apiurl = apiurl, \
dryrun = self.options.dry, \
user = user, \
logger = self.logger)
def do_id(self, subcmd, opts, *args):
"""${cmd_name}: print the status of working copy files and directories
${cmd_usage}
${cmd_option_list}
"""
self.checker.set_request_ids(args)
self.checker.check_requests()
def do_review(self, subcmd, opts, *args):
"""${cmd_name}: print the status of working copy files and directories
${cmd_usage}
${cmd_option_list}
"""
if self.checker.review_user is None:
raise osc.oscerr.WrongArgs("missing user")
self.checker.set_request_ids_search_review()
self.checker.check_requests()
if __name__ == "__main__":
app = CommandLineInterface()
sys.exit( app.main() )
# vim: sw=4 et

150
check_maintenance_incidents.py Executable file
View File

@ -0,0 +1,150 @@
#!/usr/bin/python
# Copyright (c) 2014 SUSE Linux GmbH
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
from pprint import pprint
import os, sys, re
import logging
from optparse import OptionParser
import cmdln
try:
from xml.etree import cElementTree as ET
except ImportError:
import cElementTree as ET
import osc.conf
import osc.core
import urllib2
import ReviewBot
class MaintenanceChecker(ReviewBot.ReviewBot):
""" simple bot that adds other reviewers depending on target project
"""
def __init__(self, *args, **kwargs):
ReviewBot.ReviewBot.__init__(self, *args, **kwargs)
self.review_messages = {}
# XXX: share with checkrepo
def _maintainers(self, package):
"""Get the maintainer of the package involved in the package."""
query = {
'binary': package,
}
url = osc.core.makeurl(self.apiurl, ('search', 'owner'), query=query)
root = ET.parse(osc.core.http_GET(url)).getroot()
return [p.get('name') for p in root.findall('.//person') if p.get('role') == 'maintainer']
def add_devel_project_review(self, req, package):
""" add devel project/package as reviewer """
query = {
'binary': package,
}
url = osc.core.makeurl(self.apiurl, ('search', 'owner'), query=query)
root = ET.parse(osc.core.http_GET(url)).getroot()
package_reviews = set((r.by_project, r.by_package) for r in req.reviews if r.by_package)
for p in root.findall('./owner'):
prj = p.get("project")
pkg = p.get("package")
if ((pkg, prj) in package_reviews):
# there already is a review for this project/package
continue
self.add_review(req, by_project = prj, by_package = pkg,
msg = "Submission by someone who is not maintainer in the devel project. Please review")
def check_action_maintenance_incident(self, req, a):
known_maintainer = False
author = req.get_creator()
# check if there is a link and use that or the real package
# name as src_packge may end with something like
# .openSUSE_XX.Y_Update
pkgname = a.src_package
(linkprj, linkpkg) = self._get_linktarget(a.src_project, pkgname)
if linkpkg is not None:
pkgname = linkpkg
if pkgname == 'patchinfo':
return None
maintainers = set(self._maintainers(pkgname))
if maintainers:
for m in maintainers:
if author == m:
self.logger.debug("%s is maintainer"%author)
known_maintainer = True
if not known_maintainer:
for r in req.reviews:
if r.by_user in maintainers:
self.logger.debug("found %s as reviewer"%r.by_user)
known_maintainer = True
if not known_maintainer:
self.logger.info("author: %s, maintainers: %s => need review"%(author, ','.join(maintainers)))
self.needs_maintainer_review.add(pkgname)
else:
self.logger.warning("%s doesn't have maintainers"%pkgname)
if a.tgt_releaseproject == "openSUSE:CPE:SLE-12":
self.add_factory_source = True
return True
def check_one_request(self, req):
self.add_factory_source = False
self.needs_maintainer_review = set()
ret = ReviewBot.ReviewBot.check_one_request(self, req)
if self.add_factory_source:
self.logger.debug("%s needs review by factory-source"%req.reqid)
if self.add_review(req, by_user = "factory-source") != True:
ret = None
if self.needs_maintainer_review:
for p in self.needs_maintainer_review:
self.add_devel_project_review(req, p)
return ret
class CommandLineInterface(ReviewBot.CommandLineInterface):
def __init__(self, *args, **kwargs):
ReviewBot.CommandLineInterface.__init__(self, args, kwargs)
def setup_checker(self):
apiurl = osc.conf.config['apiurl']
if apiurl is None:
raise osc.oscerr.ConfigError("missing apiurl")
user = self.options.user
if user is None:
user = osc.conf.get_apiurl_usr(apiurl)
return MaintenanceChecker(apiurl = apiurl, \
dryrun = self.options.dry, \
user = user, \
logger = self.logger)
if __name__ == "__main__":
app = CommandLineInterface()
sys.exit( app.main() )
# vim: sw=4 et

175
check_source_in_factory.py Executable file
View File

@ -0,0 +1,175 @@
#!/usr/bin/python
# Copyright (c) 2014 SUSE Linux Products GmbH
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
from pprint import pprint
import os, sys, re
import logging
from optparse import OptionParser
import cmdln
try:
from xml.etree import cElementTree as ET
except ImportError:
import cElementTree as ET
import osc.conf
import osc.core
import urllib2
import ReviewBot
class FactorySourceChecker(ReviewBot.ReviewBot):
""" this review bot checks if the sources of a submission are
either in Factory or a request for Factory with the same sources
exist. If the latter a request is only accepted if the Factory
request is reviewed positive."""
def __init__(self, *args, **kwargs):
self.factory = None
if 'factory' in kwargs:
self.factory = kwargs['factory']
del kwargs['factory']
if self.factory is None:
self.factory = "openSUSE:Factory"
ReviewBot.ReviewBot.__init__(self, *args, **kwargs)
self.review_messages = { 'accepted' : 'ok', 'declined': 'the package needs to be accepted in Factory first' }
def check_action_maintenance_incident(self, req, a):
rev = self._get_verifymd5(a.src_project, a.src_package, a.src_rev)
return self._check_package(a.src_project, a.src_package, rev, a.tgt_releaseproject, a.src_package)
def check_action_maintenance_release(self, req, a):
pkgname = a.src_package
if pkgname == 'patchinfo':
return None
linkpkg = self._get_linktarget_self(a.src_project, pkgname)
if linkpkg is not None:
pkgname = linkpkg
# packages in maintenance have links to the target. Use that
# to find the real package name
(linkprj, linkpkg) = self._get_linktarget(a.src_project, pkgname)
if linkpkg is None or linkprj is None or linkprj != a.tgt_project:
self.logger.error("%s/%s is not a link to %s"%(a.src_project, pkgname, a.tgt_project))
return False
else:
pkgname = linkpkg
src_rev = self._get_verifymd5(a.src_project, a.src_package)
return self._check_package(a.src_project, a.src_package, src_rev, a.tgt_project, pkgname)
def check_action_submit(self, req, a):
rev = self._get_verifymd5(a.src_project, a.src_package, a.src_rev)
return self._check_package(a.src_project, a.src_package, rev, a.tgt_package, a.tgt_package)
def _check_package(self, src_project, src_package, src_rev, target_project, target_package):
self.logger.info("%s/%s@%s -> %s/%s"%(src_project, src_package, src_rev, target_project, target_package))
good = self._check_factory(src_rev, target_package)
if good:
self.logger.info("%s is in Factory"%target_package)
return good
good = self._check_requests(src_rev, target_package)
if good:
self.logger.info("%s already reviewed for Factory"%target_package)
return good
def _check_factory(self, rev, package):
"""check if factory sources contain the package and revision. check head and history"""
self.logger.debug("checking %s in %s"%(package, self.factory))
srcmd5 = self._get_verifymd5(self.factory, package)
if srcmd5 is None:
self.logger.debug("new package")
return None
elif rev == srcmd5:
self.logger.debug("srcmd5 matches")
return True
self.logger.debug("%s not the latest version, checking history", rev)
u = osc.core.makeurl(self.apiurl, [ 'source', self.factory, package, '_history' ], { 'limit': '5' })
try:
r = osc.core.http_GET(u)
except urllib2.HTTPError, e:
self.logger.debug("package has no history!?")
return None
root = ET.parse(r).getroot()
for revision in root.findall('revision'):
node = revision.find('srcmd5')
if node is None:
continue
self.logger.debug("checking %s"%node.text)
if node.text == rev:
self.logger.debug("got it, rev %s"%revision.get('rev'))
return True
self.logger.debug("srcmd5 not found in history either")
return False
def _check_requests(self, rev, package):
self.logger.debug("checking requests")
requests = osc.core.get_request_list(self.apiurl, self.factory, package, None, ['new', 'review'], 'submit')
for req in requests:
for a in req.actions:
rqrev = self._get_verifymd5(a.src_project, a.src_package, a.src_rev)
self.logger.debug("rq %s: %s/%s@%s"%(req.reqid, a.src_project, a.src_package, rqrev))
if rqrev == rev:
if req.state.name == 'new':
self.logger.debug("request ok")
return True
elif req.state.name == 'review':
self.logger.debug("request still in review")
return None
else:
self.logger.error("request in state %s not expected"%req.state.name)
return None
return False
class CommandLineInterface(ReviewBot.CommandLineInterface):
def __init__(self, *args, **kwargs):
ReviewBot.CommandLineInterface.__init__(self, args, kwargs)
def get_optparser(self):
parser = ReviewBot.CommandLineInterface.get_optparser(self)
parser.add_option("--factory", metavar="project", help="the openSUSE Factory project")
return parser
def setup_checker(self):
apiurl = osc.conf.config['apiurl']
if apiurl is None:
raise osc.oscerr.ConfigError("missing apiurl")
user = self.options.user
if user is None:
user = osc.conf.get_apiurl_usr(apiurl)
return FactorySourceChecker(apiurl = apiurl, \
factory = self.options.factory, \
dryrun = self.options.dry, \
user = user, \
logger = self.logger)
if __name__ == "__main__":
app = CommandLineInterface()
sys.exit( app.main() )
# vim: sw=4 et

42
docs/factory-source.asciidoc Normal file
View File

@ -0,0 +1,42 @@
Factory Source Check
====================
:author: Ludwig Nussel <ludwig.nussel@suse.de>
:toc:
Introduction
------------
[id="intro"]
A review bot that checks if the sources of a submission are either in Factory
or a request for Factory with the same sources exist. If the latter a request
is only accepted if the Factory request is reviewed positive.
It's based on the generic ReviewBot.py
Installation
------------
[id="install"]
No installation. The bot can run directly from git.
Command line
------------
[id="cli"]
Check all request that have "factory-source" as reviewer:
-------------------------------------------------------------------------------
./check_source_in_factory.py review
-------------------------------------------------------------------------------
Checks done
-----------
[id="checks"]
This bot accepts review requests if sources for a request are accepted in
factory. Either at top, in the history or due to a submit request with the same
sources in state new. If not the request is rejected unless a submission with
the same sources in state review exists. In that case the bot doesn't touch the
request.

40
docs/maintbot.asciidoc Normal file
View File

@ -0,0 +1,40 @@
Maintenance Bot
===============
:author: Ludwig Nussel <ludwig.nussel@suse.de>
:toc:
Introduction
------------
[id="intro"]
A review bot that handles maintenance incident requests.
It's based on the generic ReviewBot.py
Installation
------------
[id="install"]
No installation. The bot can run directly from git.
Command line
------------
[id="cli"]
Check all request that have "maintbot" as reviewer:
-------------------------------------------------------------------------------
./check_maintenance_incidents.py review
-------------------------------------------------------------------------------
Checks done
-----------
[id="checks"]
This bot accepts review requests if the author of the request is a known
maintainer of the package in Factory. If not the devel project/package is set
as reviewer.
Furthermore the bot checks if the submission is for the CPE project. In that
case it adds factory-source as reviewer.

1
requirements.txt
View File

@ -2,6 +2,7 @@ PyYAML
pycurl
urlgrabber
pyxdg
cmdln
git+https://github.com/openSUSE/osc
# Dependencies for testing

11
run-maintbot Executable file
View File

@ -0,0 +1,11 @@
#!/bin/bash
: ${interval:=1800}
while date; do
echo "maintbot ..."
sudo -u "maintbot" ./check_maintenance_incidents.py "$@" review
echo "factory-source ..."
sudo -u "factory-source" ./check_source_in_factory.py "$@" review
read -t "$interval" -p "done. sleeping. press enter to check immediately"
[ "$?" -eq 0 ] || echo
[ "$REPLY" != 'q' ] || break
done

270
tests/factory_source_tests.py Normal file
View File

@ -0,0 +1,270 @@
#!/usr/bin/python
import os
import unittest
import logging
import httpretty
import osc
import re
import urlparse
from check_source_in_factory import FactorySourceChecker
APIURL = 'https://testhost.example.com'
FIXTURES = os.path.join(os.getcwd(), 'tests/fixtures')
def rr(s):
return re.compile(re.escape(APIURL + s))
class TestFactorySourceAccept(unittest.TestCase):
def setUp(self):
"""
Initialize the configuration
"""
httpretty.reset()
httpretty.enable()
oscrc = os.path.join(FIXTURES, 'oscrc')
osc.core.conf.get_config(override_conffile=oscrc,
override_no_keyring=True,
override_no_gnome_keyring=True)
#osc.conf.config['debug'] = 1
logging.basicConfig()
self.logger = logging.getLogger(__file__)
self.logger.setLevel(logging.DEBUG)
self.checker = FactorySourceChecker(apiurl = APIURL, \
user = 'factory-source', \
logger = self.logger)
def test_accept_request(self):
httpretty.register_uri(httpretty.GET,
APIURL + "/request/770001",
body = """
<request id="770001">
<action type="submit">
<source project="Base:System" package="timezone" rev="481ecbe0dfc63ece3a1f1b5598f7d96c"/>
<target project="openSUSE:13.2" package="timezone"/>
</action>
<state name="new" who="factory-source" when="2014-10-08T12:06:07">
<comment>...</comment>
</state>
<review state="new" by_user="factory-source"/>
<description>...</description>
</request>
""")
httpretty.register_uri(httpretty.GET,
rr("/source/Base:System/timezone?rev=481ecbe0dfc63ece3a1f1b5598f7d96c&view=info"),
match_querystring = True,
body = """
<sourceinfo package="timezone"
rev="481ecbe0dfc63ece3a1f1b5598f7d96c"
srcmd5="481ecbe0dfc63ece3a1f1b5598f7d96c"
verifymd5="67bac34d29d70553239d33aaf92d2fdd">
<filename>timezone.spec</filename>
</sourceinfo>
""")
httpretty.register_uri(httpretty.GET,
rr("/source/openSUSE:Factory/timezone?view=info"),
match_querystring = True,
body = """
<sourceinfo package="timezone"
rev="89"
vrev="1"
srcmd5="a36605617cbeefa8168bf0ccf3058074"
verifymd5="a36605617cbeefa8168bf0ccf3058074">
<filename>timezone.spec</filename>
</sourceinfo>
""")
httpretty.register_uri(httpretty.GET,
rr("/source/openSUSE:Factory/timezone/_history?limit=5"),
match_querystring = True,
body = """
<sourceinfo package="timezone"
rev="89"
vrev="1"
srcmd5="a36605617cbeefa8168bf0ccf3058074"
verifymd5="a36605617cbeefa8168bf0ccf3058074">
<filename>timezone.spec</filename>
</sourceinfo>
""")
httpretty.register_uri(httpretty.GET,
rr("/search/request?match=%28state%2F%40name%3D%27new%27+or+state%2F%40name%3D%27review%27%29+and+%28action%2Ftarget%2F%40project%3D%27openSUSE%3AFactory%27+or+submit%2Ftarget%2F%40project%3D%27openSUSE%3AFactory%27+or+action%2Fsource%2F%40project%3D%27openSUSE%3AFactory%27+or+submit%2Fsource%2F%40project%3D%27openSUSE%3AFactory%27%29+and+%28action%2Ftarget%2F%40package%3D%27timezone%27+or+submit%2Ftarget%2F%40package%3D%27timezone%27+or+action%2Fsource%2F%40package%3D%27timezone%27+or+submit%2Fsource%2F%40package%3D%27timezone%27%29+and+action%2F%40type%3D%27submit%27"),
match_querystring = True,
responses = [
httpretty.Response( body = """
<collection matches="1">
<request id="254684">
<action type="submit">
<source project="Base:System" package="timezone" rev="481ecbe0dfc63ece3a1f1b5598f7d96c"/>
<target project="openSUSE:Factory" package="timezone"/>
</action>
<state name="review" who="factory-auto" when="2014-10-08T11:55:56">
<comment>...</comment>
</state>
<description> ... </description>
</request>
</collection>
"""),
httpretty.Response( body = """
<collection matches="1">
<request id="254684">
<action type="submit">
<source project="Base:System" package="timezone" rev="481ecbe0dfc63ece3a1f1b5598f7d96c"/>
<target project="openSUSE:Factory" package="timezone"/>
</action>
<state name="new" who="factory-auto" when="2014-10-08T11:55:56">
<comment>...</comment>
</state>
<description> ... </description>
</request>
</collection>
""")
])
result = { 'status' : None }
def change_request(result, method, uri, headers):
u = urlparse.urlparse(uri)
if u.query == 'newstate=accepted&cmd=changereviewstate&by_user=factory-source':
result['status'] = True
return (200, headers, '<status code="blah"/>')
httpretty.register_uri(httpretty.POST,
APIURL + "/request/770001",
body = lambda method, uri, headers: change_request(result, method, uri, headers))
# first time request is in in review
self.checker.set_request_ids(['770001'])
self.checker.check_requests()
self.assertEqual(result['status'], None)
# second time request is in state new so we can accept
self.checker.set_request_ids(['770001'])
self.checker.check_requests()
self.assertTrue(result['status'])
def test_source_not_in_factory(self):
httpretty.register_uri(httpretty.GET,
rr("/search/request?match=state/@name='review'+and+review[@by_user='factory-source'+and+@state='new']&withhistory=1"),
match_querystring = True,
body = """
<collection matches="1">
<request id="261411">
<action type="maintenance_incident">
<source project="home:lnussel:branches:openSUSE:CPE:SLE-12" package="plan" rev="71e76daf2c2e9ddb0b9208f54a14f608"/>
<target project="openSUSE:Maintenance" releaseproject="openSUSE:CPE:SLE-12"/>
</action>
<state name="review" who="maintbot" when="2014-11-13T13:22:02">
<comment></comment>
</state>
<review state="accepted" when="2014-11-13T13:22:02" who="maintbot" by_user="maintbot">
<comment>accepted</comment>
<history who="maintbot" when="2014-11-13T16:43:09">
<description>Review got accepted</description>
<comment>accepted</comment>
</history>
</review>
<review state="new" by_user="factory-source"/>
<history who="lnussel" when="2014-11-13T13:22:02">
<description>Request created</description>
<comment>test update</comment>
</history>
<history who="maintbot" when="2014-11-13T16:43:08">
<description>Request got a new review request</description>
</history>
<description>test update</description>
</request>
</collection>
""")
httpretty.register_uri(httpretty.GET,
APIURL + "/request/261411",
body = """
<request id="261411">
<action type="maintenance_incident">
<source project="home:lnussel:branches:openSUSE:CPE:SLE-12" package="plan" rev="71e76daf2c2e9ddb0b9208f54a14f608"/>
<target project="openSUSE:Maintenance" releaseproject="openSUSE:CPE:SLE-12"/>
</action>
<state name="review" who="maintbot" when="2014-11-13T13:22:02">
<comment></comment>
</state>
<review state="accepted" when="2014-11-13T13:22:02" who="maintbot" by_user="maintbot">
<comment>accepted</comment>
<history who="maintbot" when="2014-11-13T16:43:09">
<description>Review got accepted</description>
<comment>accepted</comment>
</history>
</review>
<review state="new" by_user="factory-source"/>
<history who="lnussel" when="2014-11-13T13:22:02">
<description>Request created</description>
<comment>test update</comment>
</history>
<history who="maintbot" when="2014-11-13T16:43:08">
<description>Request got a new review request</description>
</history>
<description>test update</description>
</request>
""")
httpretty.register_uri(httpretty.GET,
APIURL + "/source/home:lnussel:branches:openSUSE:CPE:SLE-12/plan",
body = """
<directory name="plan" rev="1" vrev="1" srcmd5="b4ed19dc30c1b328168bc62a81ec6998">
<linkinfo project="home:lnussel:plan" package="plan" srcmd5="7a2353f73b29dba970702053229542a0" baserev="7a2353f73b29dba970702053229542a0" xsrcmd5="71e76daf2c2e9ddb0b9208f54a14f608" lsrcmd5="b4ed19dc30c1b328168bc62a81ec6998" />
<entry name="_link" md5="91f81d88456818a18a7332999fb2da18" size="125" mtime="1415807350" />
<entry name="plan.spec" md5="b6814215f6d2e8559b43de9a214b2cbd" size="8103" mtime="1413627959" />
</directory>
""")
httpretty.register_uri(httpretty.GET,
rr("/source/openSUSE:Factory/plan?view=info"),
match_querystring = True,
status = 404,
body = """
<status code="unknown_package">
<summary>openSUSE:Factory/plan</summary>
</status>
""")
httpretty.register_uri(httpretty.GET,
rr("/search/request?match=%28state%2F%40name%3D%27new%27+or+state%2F%40name%3D%27review%27%29+and+%28action%2Ftarget%2F%40project%3D%27openSUSE%3AFactory%27+or+submit%2Ftarget%2F%40project%3D%27openSUSE%3AFactory%27+or+action%2Fsource%2F%40project%3D%27openSUSE%3AFactory%27+or+submit%2Fsource%2F%40project%3D%27openSUSE%3AFactory%27%29+and+%28action%2Ftarget%2F%40package%3D%27plan%27+or+submit%2Ftarget%2F%40package%3D%27plan%27+or+action%2Fsource%2F%40package%3D%27plan%27+or+submit%2Fsource%2F%40package%3D%27plan%27%29+and+action%2F%40type%3D%27submit%27"),
match_querystring = True,
body = """
<collection matches="0">
</collection>
""")
result = { 'factory_source_declined' : None }
def change_request(result, method, uri, headers):
u = urlparse.urlparse(uri)
if u.query == 'newstate=declined&cmd=changereviewstate&by_user=factory-source':
result['factory_source_declined'] = True
return (200, headers, '<status code="ok"/>')
httpretty.register_uri(httpretty.POST,
APIURL + "/request/261411",
body = lambda method, uri, headers: change_request(result, method, uri, headers))
self.checker.requests = []
self.checker.set_request_ids_search_review()
self.checker.check_requests()
self.assertTrue(result['factory_source_declined'])
if __name__ == '__main__':
unittest.main()
# vim: sw=4 et

233
tests/maintenance_tests.py Normal file
View File

@ -0,0 +1,233 @@
#!/usr/bin/python
import os
import unittest
import logging
import httpretty
import osc
import re
import urlparse
from check_maintenance_incidents import MaintenanceChecker
APIURL = 'https://maintenancetest.example.com'
FIXTURES = os.path.join(os.getcwd(), 'tests/fixtures')
def rr(s):
return re.compile(re.escape(APIURL + s))
class TestMaintenance(unittest.TestCase):
def setUp(self):
"""
Initialize the configuration
"""
httpretty.reset()
httpretty.enable()
oscrc = os.path.join(FIXTURES, 'oscrc')
osc.core.conf.get_config(override_conffile=oscrc,
override_no_keyring=True,
override_no_gnome_keyring=True)
#osc.conf.config['debug'] = 1
logging.basicConfig()
self.logger = logging.getLogger(__file__)
self.logger.setLevel(logging.DEBUG)
self.checker = MaintenanceChecker(apiurl = APIURL, \
user = 'maintbot', \
logger = self.logger)
def test_non_maintainer_submit(self):
httpretty.register_uri(httpretty.GET,
rr("/search/request?match=state/@name='review'+and+review[@by_user='maintbot'+and+@state='new']&withhistory=1"),
match_querystring = True,
body = """
<collection matches="1">
<request id="261355">
<action type="maintenance_incident">
<source project="home:brassh" package="mysql-workbench" rev="857c77d2ba1d347b6dc50a1e5bcb74e1"/>
<target project="openSUSE:Maintenance" releaseproject="openSUSE:13.2:Update"/>
</action>
<state name="review" who="lnussel_factory" when="2014-11-13T10:46:52">
<comment></comment>
</state>
<review state="new" by_user="maintbot">
<comment></comment>
</review>
<history who="brassh" when="2014-11-13T09:18:19">
<description>Request created</description>
<comment>...</comment>
</history>
<history who="lnussel_factory" when="2014-11-13T10:46:52">
<description>Request got a new review request</description>
</history>
<description>...</description>
</request>
</collection>
""")
httpretty.register_uri(httpretty.GET,
APIURL + "/request/261355",
match_querystring = True,
body = """
<request id="261355">
<action type="maintenance_incident">
<source project="home:brassh" package="mysql-workbench" rev="857c77d2ba1d347b6dc50a1e5bcb74e1"/>
<target project="openSUSE:Maintenance" releaseproject="openSUSE:13.2:Update"/>
</action>
<state name="review" who="lnussel_factory" when="2014-11-13T10:46:52">
<comment></comment>
</state>
<review state="new" by_user="maintbot">
<comment></comment>
</review>
<history who="brassh" when="2014-11-13T09:18:19">
<description>Request created</description>
<comment>...</comment>
</history>
<history who="lnussel_factory" when="2014-11-13T10:46:52">
<description>Request got a new review request</description>
</history>
<description>...</description>
</request>
""")
httpretty.register_uri(httpretty.GET,
APIURL + "/source/home:brassh/mysql-workbench",
match_querystring = True,
body = """
<directory name="mysql-workbench" rev="6" vrev="6" srcmd5="858204decf53f923d5574dbe6ae63b15">
<linkinfo project="openSUSE:13.2" package="mysql-workbench" srcmd5="ed9c3b12388cbd14868eb3faabe34685" baserev="ed9c3b12388cbd14868eb3faabe34685" xsrcmd5="08bfb4f40cb1e2de8f9cd4633bf02eb1" lsrcmd5="858204decf53f923d5574dbe6ae63b15" />
<serviceinfo code="succeeded" xsrcmd5="6ec4305a8e5363e26a7f4895a0ae12d2" />
<entry name="_link" md5="85ef5fb38ca1ec7c300311fda9f4b3d1" size="121" mtime="1414567341" />
<entry name="mysql-workbench-community-6.1.7-src.tar.gz" md5="ac059e239869fb77bf5d7a1f5845a8af" size="24750696" mtime="1404405925" />
<entry name="mysql-workbench-ctemplate.patch" md5="06ccba1f8275cd9408f515828ecede19" size="1322" mtime="1404658323" />
<entry name="mysql-workbench-glib.patch" md5="67fd7d8e3503ce0909381bde747c8a1e" size="1785" mtime="1415732509" />
<entry name="mysql-workbench-mysql_options4.patch" md5="9c07dfe1b94af95daf3e16bd6a161684" size="910" mtime="1404658324" />
<entry name="mysql-workbench-no-check-for-updates.patch" md5="1f0c9514ff8218d361ea46d3031b2b64" size="1139" mtime="1404658324" />
<entry name="mysql-workbench.changes" md5="26bc54777e6a261816b72f64c69630e4" size="13354" mtime="1415747835" />
<entry name="mysql-workbench.spec" md5="88b562a93f01b842a5798f809e3c8188" size="7489" mtime="1415745943" />
<entry name="openSUSE_(Vendor_Package).xml" md5="ab041af98d7748c216e7e5787ec36f65" size="743" mtime="1315923090" />
<entry name="patch-desktop-categories.patch" md5="c24b3283573c34a5e072be122388f8e1" size="391" mtime="1376991147" />
</directory>
""")
result = { 'devel_review_added' : None }
def change_request(result, method, uri, headers):
u = urlparse.urlparse(uri)
if u.query == 'by_package=mysql-workbench&cmd=addreview&by_project=server%3Adatabase':
result['devel_review_added'] = True
return (200, headers, '<status code="ok"/>')
httpretty.register_uri(httpretty.POST,
APIURL + "/request/261355",
body = lambda method, uri, headers: change_request(result, method, uri, headers))
httpretty.register_uri(httpretty.GET,
rr("/search/owner?binary=mysql-workbench"),
match_querystring = True,
body = """
<collection>
<owner rootproject="openSUSE" project="server:database" package="mysql-workbench">
<person name="Gankov" role="maintainer"/>
<person name="bruno_friedmann" role="maintainer"/>
</owner>
</collection>
""")
self.checker.requests = []
self.checker.set_request_ids_search_review()
self.checker.check_requests()
self.assertTrue(result['devel_review_added'])
def test_cpe_submit(self):
httpretty.register_uri(httpretty.GET,
rr("/search/request?match=state/@name='review'+and+review[@by_user='maintbot'+and+@state='new']&withhistory=1"),
match_querystring = True,
body = """
<collection matches="1">
<request id="261411">
<action type="maintenance_incident">
<source project="home:lnussel:branches:openSUSE:CPE:SLE-12" package="plan" rev="71e76daf2c2e9ddb0b9208f54a14f608"/>
<target project="openSUSE:Maintenance" releaseproject="openSUSE:CPE:SLE-12"/>
</action>
<state name="review" who="lnussel" when="2014-11-13T13:22:02">
<comment></comment>
</state>
<review state="new" by_user="maintbot"/>
<history who="lnussel" when="2014-11-13T13:22:02">
<description>Request created</description>
<comment>test update</comment>
</history>
<description>test update</description>
</request>
</collection>
""")
httpretty.register_uri(httpretty.GET,
APIURL + "/request/261411",
body = """
<request id="261411">
<action type="maintenance_incident">
<source project="home:lnussel:branches:openSUSE:CPE:SLE-12" package="plan" rev="71e76daf2c2e9ddb0b9208f54a14f608"/>
<target project="openSUSE:Maintenance" releaseproject="openSUSE:CPE:SLE-12"/>
</action>
<state name="review" who="lnussel" when="2014-11-13T13:22:02">
<comment></comment>
</state>
<review state="new" by_user="maintbot"/>
<history who="lnussel" when="2014-11-13T13:22:02">
<description>Request created</description>
<comment>test update</comment>
</history>
<description>test update</description>
</request>
""")
httpretty.register_uri(httpretty.GET,
APIURL + "/source/home:lnussel:branches:openSUSE:CPE:SLE-12/plan",
body = """
<directory name="plan" rev="1" vrev="1" srcmd5="b4ed19dc30c1b328168bc62a81ec6998">
<linkinfo project="home:lnussel:plan" package="plan" srcmd5="7a2353f73b29dba970702053229542a0" baserev="7a2353f73b29dba970702053229542a0" xsrcmd5="71e76daf2c2e9ddb0b9208f54a14f608" lsrcmd5="b4ed19dc30c1b328168bc62a81ec6998" />
<entry name="_link" md5="91f81d88456818a18a7332999fb2da18" size="125" mtime="1415807350" />
<entry name="plan.spec" md5="b6814215f6d2e8559b43de9a214b2cbd" size="8103" mtime="1413627959" />
</directory>
""")
httpretty.register_uri(httpretty.GET,
rr("/search/owner?binary=plan"),
match_querystring = True,
body = """
<collection/>
""")
result = { 'factory_review_added' : None }
def change_request(result, method, uri, headers):
u = urlparse.urlparse(uri)
if u.query == 'cmd=addreview&by_user=factory-source':
result['factory_review_added'] = True
return (200, headers, '<status code="ok"/>')
httpretty.register_uri(httpretty.POST,
APIURL + "/request/261411",
body = lambda method, uri, headers: change_request(result, method, uri, headers))
self.checker.requests = []
self.checker.set_request_ids_search_review()
self.checker.check_requests()
self.assertTrue(result['factory_review_added'])
if __name__ == '__main__':
unittest.main()
# vim: sw=4 et

21
tests/obs.py
View File

@ -57,12 +57,6 @@ def router_handler_DELETE(request, uri, headers):
return router_handler(_table[httpretty.DELETE], 'DELETE', request, uri, headers)
httpretty.register_uri(httpretty.GET, re.compile(r'.*'), body=router_handler_GET)
httpretty.register_uri(httpretty.POST, re.compile(r'.*'), body=router_handler_POST)
httpretty.register_uri(httpretty.PUT, re.compile(r'.*'), body=router_handler_PUT)
httpretty.register_uri(httpretty.DELETE, re.compile(r'.*'), body=router_handler_DELETE)
def method_decorator(method, path):
def _decorator(fn):
def _fn(*args, **kwargs):
@ -94,9 +88,18 @@ class OBS(object):
def __new__(cls, *args, **kwargs):
"""Class constructor."""
if not cls._self:
cls._self = super(OBS, cls).__new__(cls, *args, **kwargs)
return cls._self
if not OBS._self:
OBS._self = super(OBS, cls).__new__(cls, *args, **kwargs)
httpretty.reset()
httpretty.enable()
httpretty.register_uri(httpretty.GET, re.compile(r'.*'), body=router_handler_GET)
httpretty.register_uri(httpretty.POST, re.compile(r'.*'), body=router_handler_POST)
httpretty.register_uri(httpretty.PUT, re.compile(r'.*'), body=router_handler_PUT)
httpretty.register_uri(httpretty.DELETE, re.compile(r'.*'), body=router_handler_DELETE)
return OBS._self
def __init__(self, fixtures=FIXTURES):
"""Instance constructor."""