Differential D137

Depcheck will run on taskotron if changes from D136 are applied to taskotron
AbandonedPublic
Actions

Authored by jskladan on Jun 12 2014, 12:31 PM.

Details

Reviewers
kparal
tflink
mkrizek
Summary

These changes make depcheck runable in Taskotron.
Note that further changes are required, since squash_results needs to squash per-rpm results to per-builds updates before squashing the per-build results to per-update results.
It did not occur to me, at the beginning, that list of RPMs is not list of builds, so at the moment, depcheck can false-positive when one (or more) rpms from the build fails.

To be specific, let's have a build XYZ containing rpms XYZ.rpm XYZ_doc.rpm and XYZ_lib.rpm. The way it is currently (wrong) dealt with is, that squash_results gets list of per-rpm results, where the rpm name is converted to NVR.
This means that the overall build-outcome is not determined by aggregating all three rpms's results, but just on the XYZ.rpms outcome.

This needs to be dealt with, but the patch will not be a small one, and I will provide it once it is done.

Test Plan

unittests work, running through taskotron's runner works

Diff Detail

Repository
rDEPCK task-depcheck
Branch
feature/t187
Lint
No Linters Available
Unit
No Unit Test Coverage
jskladan retitled this revision from to Depcheck will run on taskotron if changes from D136 are applied to taskotron.Jun 12 2014, 12:31 PM
jskladan updated this object.
jskladan edited the test plan for this revision. (Show Details)
jskladan added reviewers: tflink, kparal, mkrizek.
jskladan abandoned this revision.Jun 12 2014, 12:52 PM

Revision Contents

PathPackages
Mdepcheck.yml (44 lines)
Mdepcheck/__init__.py (35 lines)
Mdepcheck/depcheck.py (36 lines)
AMdepcheck/exc.py (4 lines)
Mdepcheck/squash_results.py (70 lines)
Metc/depcheck.conf (2 lines)
Mrun_depcheck.py (39 lines)
Mtests/test_depcheck.py (9 lines)
Mtests/test_squash_results.py (56 lines)

Diff 396

View Options

depcheck.yml

1--- 1​---
2name: depcheck 2​name: depcheck
3desc: "Check dependency trees" 3​desc: "Check dependency trees"
4maintainer: jdulaney 4​maintainer: jdulaney
5 5​
6input: 6​input:
7 args: arch 7​ args:
8​ - arch
9​ - koji_tag
8 10​
9environment: 11​environment:
10 rpm: 12​ rpm:
11 - depcheck 13​ - depcheck
12 - libtaskotron 14​ - libtaskotron
13 15​
14task: 16​task:
17​ - name: download koji tag
18​ koji:
19​ action: download_tag
20​ koji_tag: ${koji_tag}
21​ arch: ${arch}
22​ target_dir: "${workdir}/downloaded_tag/"
23​
24​ - name: mash downloaded rpms
25​ mash:
26​ rpmdir: "${workdir}/downloaded_tag/"
27​ dodelta: "false"
28​ arch: ${arch}
29​
30​ - name: get yum repos
31​ yumrepoinfo:
32​ koji_tag: ${koji_tag}
33​ arch: ${arch}
34​ export: yumrepos
35​
36​ # get repos
15 - name: run depcheck 37​ - name: run depcheck
16 python: 38​ python:
17 file: run_depcheck.py 39​ file: run_depcheck.py
18 callable: run 40​ callable: taskotron_run
19 arch: "{{ arch }}" 41​ arch: ${arch}
20 rpms: "{{ workdir }}" 42​ #rpms: ["/var/tmp/taskotron/task-XfvSng/downloaded_tag/"]
21 repos: "{{ repos }}" 43​ rpms: ["${workdir}/downloaded_tag/"]
22 report_format: "update" 44​ repos:
45​ - ${yumrepos}
46​ #- "workdir_repo": "/var/tmp/taskotron/task-XfvSng/downloaded_tag/"
47​ - "workdir_repo": "${workdir}/downloaded_tag/"
48​ report_format: "updates"
23 export: depcheck_output 49​ export: depcheck_output
24 50​
25 - name: report results to resultsdb 51​# - name: report results to resultsdb
26 resultsdb: 52​# resultsdb:
27 results: "{{ depcheck_output }}" 53​# results: "{{ depcheck_output }}"
28 54​
View Options

depcheck/__init__.py

Show All 29 Lines
30Pass = 'PASSED' 30​Pass = 'PASSED'
31Fail = 'FAILED' 31​Fail = 'FAILED'
32 32​
33import os.path 33​import os.path
34import sys 34​import sys
35import argparse 35​import argparse
36import tempfile 36​import tempfile
37import platform 37​import platform
38​import json
38 39​
39from . import depcheck 40​from . import depcheck
40from . import metadata 41​from . import metadata
42​from . import exc
43​from . import squash_results
44​
45​from libtaskotron.python_utils import iterable
41 46​
42 47​
43def dir_to_rpms(dirname): 48​def dir_to_rpms(dirname):
44 """Find all rpms in the specified directory 49​ """Find all rpms in the specified directory
45 :param dirname: local directory in which to look for rpms 50​ :param dirname: local directory in which to look for rpms
46 :type dirnema: str 51​ :type dirnema: str
47 :returns: list of rpm files with absolute, local path 52​ :returns: list of rpm files with absolute, local path
48 """ 53​ """
49 rpm_list = [] 54​ rpm_list = []
50 for filename in os.listdir(dirname): 55​ for filename in os.listdir(dirname):
51 if filename.endswith('.rpm'): 56​ if filename.endswith('.rpm'):
52 rpm_list.append(os.path.join(dirname, filename)) 57​ rpm_list.append(os.path.join(dirname, filename))
53 return rpm_list 58​ return rpm_list
54 59​
55 60​
56def extract_named_repo(namedrepo): 61​def extract_named_repo(namedrepo):
57 return namedrepo.split(':', 1) 62​ return namedrepo.split(':', 1)
58 63​
59 64​
60def run(rpms, arch, repos, report_format="builds"): 65​def run(rpms, arch, repos, report_format="rpms"):
66​ """
67​ report_format: rpms, updates
68​ """
61 69​
62 rpmlist = [] ​
63 # handle directory args, expanding them to a list of rpm files with absolute 70​ # handle directory args, expanding them to a list of rpm files with absolute
64 # paths 71​ # paths
72​ if isinstance(rpms, basestring):
73​ rpms = [rpms]
74​ if not iterable(rpms, basestring):
75​ raise exc.DepcheckException("rpms argument must be either string"
76​ "or list of strings")
77​
78​ rpmlist = []
65 for filename in rpms: 79​ for filename in rpms:
66 if os.path.isdir(filename): 80​ if os.path.isdir(filename):
67 rpmlist.extend(dir_to_rpms(filename)) 81​ rpmlist.extend(dir_to_rpms(filename))
68 else: 82​ else:
69 rpmlist.append(filename) 83​ rpmlist.append(filename)
70 84​
71 # arm has some special case requirements for arch handling 85​ # arm has some special case requirements for arch handling
72 # Fixup for arm; the yum repo is labled as armhfp but the nvr is armv7hl 86​ # Fixup for arm; the yum repo is labled as armhfp but the nvr is armv7hl
73 if arch == 'armv7l': 87​ if arch == 'armv7l':
74 arch = 'armhfp' 88​ arch = 'armhfp'
75 89​
76 # handle data formatting for repo metadata (list of dicts) 90​ # handle data formatting for repo metadata (list of dicts)
77 repos = {} 91​ handled_repos = {}
78 for repo in repos: 92​ for repo, path in repos.items():
79 namedrepo = extract_named_repo(repo) 93​ handled_repos[repo] = {'arch': arch, 'path': path}
80 repos[namedrepo[0]] = {'arch': arch, 'path': namedrepo[1]} ​
81 94​
82 workdir = tempfile.mkdtemp('', 'depcheck-', '/var/tmp/') 95​ workdir = tempfile.mkdtemp('', 'depcheck-', '/var/tmp/')
83 md_handle = metadata.DepcheckMetadata(repos, targetdir=workdir) 96​ md_handle = metadata.DepcheckMetadata(handled_repos, targetdir=workdir)
84 97​
85 repomd_files = md_handle.repodata 98​ repomd_files = md_handle.repodata
86 99​
87 output = None 100​ output = None
88 101​
89 depck = Depcheck(arch, repomd_files, workdir=workdir) 102​ depck = depcheck.Depcheck(str(arch), repomd_files, workdir=workdir)
90 run_result = depck.check_rpm_deps(rpmlist) 103​ run_result = depck.check_rpm_deps(rpmlist)
91 104​
92 if report_format == 'updates': 105​ if report_format == 'updates':
93 results = squash_results.squash_build_results_to_updates(run_result) 106​ results = squash_results.squash_rpm_results_to_updates(run_result)
94 output = squash_results.check.export_TAP(results) 107​ output = squash_results.check.export_TAP(*results)
95 else: 108​ else:
96 output = json.dumps(run_result, indent=4, separators=(',', ': ')) 109​ output = json.dumps(run_result, indent=4, separators=(',', ': '))
97 110​
98 return output 111​ return output
99 112​
100def depcheck_cli(): 113​def depcheck_cli():
101 parser = argparse.ArgumentParser() 114​ parser = argparse.ArgumentParser()
102 parser.add_argument("rpm", nargs='+', help="RPM to be tested") 115​ parser.add_argument("rpm", nargs='+', help="RPM to be tested")
103 parser.add_argument("-a", "--arch", choices=["i386", "x86_64", "armhfp"], help=""" 116​ parser.add_argument("-a", "--arch", choices=["i386", "x86_64", "armhfp"], help="""
104 Architecture to be tested; support arches are x86, x86_64, and armhfp. 117​ Architecture to be tested; support arches are x86, x86_64, and armhfp.
105 If omitted, defaults to current machine architecture. 118​ If omitted, defaults to current machine architecture.
106 """) 119​ """)
107 parser.add_argument("-r", "--repos", nargs='+', action='store', help=""" 120​ parser.add_argument("-r", "--repos", nargs='+', action='store', help="""
108 Repo to be tested against, in the form of reponame:repolocation where 121​ Repo to be tested against, in the form of reponame:repolocation where
109 repolocation can be local (in the form /dir/of/repo) or 122​ repolocation can be local (in the form /dir/of/repo) or
110 remote (direct link or standard metalink), to be tested against. 123​ remote (direct link or standard metalink), to be tested against.
111 """) 124​ """)
112 parser.add_argument("-f", "--format", choices=["builds", "updates"], help=""" 125​ parser.add_argument("-f", "--format", choices=["rpms", "updates"], help="""
113 Specify output format to override configuration 126​ Specify output format to override configuration
114 """) 127​ """)
115 128​
116 129​
117 args = parser.parse_args() 130​ args = parser.parse_args()
118 # autodetect arch if none is passed in 131​ # autodetect arch if none is passed in
119 arch = args.arch 132​ arch = args.arch
120 if arch is None: 133​ if arch is None:
Show All 10 Lines
View Options

depcheck/depcheck.py

Show All 21 Lines
22# Authors: 22​# Authors:
23# John H. Dulaney <jdulaney@fedoraproject.org> 23​# John H. Dulaney <jdulaney@fedoraproject.org>
24# Tim Flink <tflink@redhat.com> 24​# Tim Flink <tflink@redhat.com>
25# 25​#
26# EPEL dep: python-argparse 26​# EPEL dep: python-argparse
27 27​
28 28​
29import solv 29​import solv
30import os ​
31import tempfile 30​import tempfile
32 31​
33from . import Pass, Fail 32​from . import Pass, Fail
33​from . import exc
34 34​
35from libtaskotron.rpm_utils import rpmformat ​
36 35​
37config_file = "/etc/depcheck.conf" 36​config_file = "/etc/depcheck.conf"
38 37​
39# TODO: handle configuration file 38​# TODO: handle configuration file
40 39​
41class DepcheckException(Exception): ​
42 pass ​
43 ​
44 ​
45class Depcheck(object): 40​class Depcheck(object):
46 def __init__(self, arch, repodata, workdir=None): 41​ def __init__(self, arch, repodata, workdir=None):
47 """Depcheck uses existing repodata to check a set of rpm files for 42​ """Depcheck uses existing repodata to check a set of rpm files for
48 depsolving issues. 43​ depsolving issues.
49 44​
50 This class itself doesn't handle much other than checking rpms and all 45​ This class itself doesn't handle much other than checking rpms and all
51 data prep needs to be handled externally 46​ data prep needs to be handled externally
52 47​
53 :param arch: The repository arch to use for dependency solving 48​ :param arch: The repository arch to use for dependency solving
54 :type arch: str 49​ :type arch: str (not unicode)
55 :param repodata: local paths to downloaded and verified repodata 50​ :param repodata: local paths to downloaded and verified repodata
56 :type repodata: dir of dir {'reponame': {'primary': primarygz_filename, 'filelists': filelists_filename}...} 51​ :type repodata: dir of dir {'reponame': {'primary': primarygz_filename, 'filelists': filelists_filename}...}
57 :param workdir: The directory to use for depcheck work, a temp directory will be created if workdir is None 52​ :param workdir: The directory to use for depcheck work, a temp directory will be created if workdir is None
58 :type workdir: str 53​ :type workdir: str
59 """ 54​ """
60 55​
61 self.arch = arch 56​ self.arch = arch
62 self.repos = repodata 57​ self.repos = repodata
▲ Show 20 Lines • Show All 62 Lines • ▼ Show 20 Line(s)111 else:
125 120​
126 return Fail, problem_strs 121​ return Fail, problem_strs
127 122​
128 def check_rpm_deps(self, rpmfiles): 123​ def check_rpm_deps(self, rpmfiles):
129 """Check dependencies of the given rpms against supplied repodata 124​ """Check dependencies of the given rpms against supplied repodata
130 125​
131 :param rpmfiles: local absolute paths to rpms to check 126​ :param rpmfiles: local absolute paths to rpms to check
132 :type rpmfiles: list of str 127​ :type rpmfiles: list of str
133 :returns: dictionary of results {NVR: {'results': Pass/Fail, 'details': []}} 128​ :returns: dictionary of results {rpm: {'results': Pass/Fail, 'details': []}}
134 """ 129​ """
135 130​
136 self._prepare_libsolv() 131​ self._prepare_libsolv()
137 132​
138 run_result = {} 133​ run_result = {}
139 134​
140 for rpmfile in rpmfiles: 135​ for rpmfile in rpmfiles:
141 print 'running depcheck for %s' % rpmfile 136​ print 'running depcheck for %s' % rpmfile
142 result_state, result_details = self._check_rpm(rpmfile) 137​ result_state, result_details = self._check_rpm(rpmfile)
143 138​
144 nvr = self.rpm_filename_to_nvr(rpmfile) 139​ if rpmfile in run_result:
145 if nvr in run_result: 140​ if result_state != run_result[rpmfile]['result']:
146 if result_state != run_result[nvr]['result']: ​
147 #TODO maybe do something saner, ideas welcomed 141​ #TODO maybe do something saner, ideas welcomed
148 raise DepcheckException("NVR %r has multiple conflicting results" % nvr) 142​ raise exc.DepcheckException("RPM %r has multiple conflicting results" % rpmfile)
149 else: 143​ else:
150 run_result[nvr]['details'].append(result_details) 144​ run_result[rpmfile]['details'].append(result_details)
151 145​
152 else: 146​ else:
153 run_result[nvr] = {'result': result_state, 147​ run_result[rpmfile] = {'result': result_state,
154 'details': result_details} 148​ 'details': result_details}
155 149​
156 return run_result 150​ return run_result
157 151​
158 152​
159 # FIXME: this will eventually be a util in libtaskotron ​
160 def rpm_filename_to_nvr(self, filename): ​
161 """ ​
162 Transforms rpm filename to NVR. ​
163 We assume the Fedora environment, where all the rpm filenames ​
164 are in the NVRA.rpm format. ​
165 ​
166 :param filename: name of the rpm file ​
167 :type filename: str ​
168 :returns: NVR ​
169 """ ​
170 name = os.path.basename(filename)[:-len('.rpm')] ​
171 return rpmformat(name, 'nvr') ​
172 ​
173 153​
View Options

depcheck/exc.py

  • This file was added.
1​class DepcheckException(Exception):
2​ pass
3​
4​
View Options

depcheck/squash_results.py

1​import os
2​
3​from . import Pass, Fail
4​
1from libtaskotron.bodhi_utils import BodhiUtils 5​from libtaskotron.bodhi_utils import BodhiUtils
2from libtaskotron import check 6​from libtaskotron import check
3from . import Pass, Fail 7​from libtaskotron.rpm_utils import rpmformat
8​
9​# FIXME: this will eventually be a util in libtaskotron
10​def rpm_filename_to_nvr(filename, format='nvr'):
11​ """
12​ Transforms rpm filename to NVR.
13​ We assume the Fedora environment, where all the rpm filenames
14​ are in the NVRA.rpm format.
15​
16​ :param filename: name of the rpm file
17​ :type filename: str
18​ :returns: NVR
19​ """
20​ if filename.endswith('.rpm'):
21​ filename = os.path.basename(filename)[:-len('.rpm')]
22​ return rpmformat(filename, format, end_arch = True)
4 23​
5def squash_build_results_to_updates(run_result, bodhi = None): 24​
25​def squash_rpms_to_builds(run_result):
26​ results_nvr = {}
27​ for rpm, result in run_result.items():
28​ nvr = rpm_filename_to_nvr(rpm)
29​
30​ if nvr in results_nvr:
31​ if results_nvr['nvr']['result'] != result['result']:
32​ #TODO maybe do something saner, ideas welcomed
33​ raise exc.DepcheckException("NVR %r has multiple conflicting results" % nvr)
34​ else:
35​ results_nvr[nvr] = result
36​ return results_nvr
37​
38​
39​def squash_rpm_results_to_updates(run_result, bodhi = None):
6 """ 40​ """
7 Converts the per-build results created by check_rpm_deps into per-update 41​ Converts the per-rpm results created by check_rpm_deps into per-update
8 results. 42​ results.
9 :param run_result: output of the check_rpm_deps method 43​ :param run_result: output of the check_rpm_deps method
10 :type run_result: dict 44​ :type run_result: dict
11 :returns: list of CheckDetail 45​ :returns: list of CheckDetail
12 """ 46​ """
13 if bodhi is None: 47​ if bodhi is None:
14 bodhi = BodhiUtils() 48​ bodhi = BodhiUtils()
15 49​
16 pending_ok, pending_fail = bodhi.build2update(run_result.keys(), strict = True) 50​ results_nvr = squash_rpms_to_builds(run_result)
51​
52​ pending_ok, pending_fail = bodhi.build2update(results_nvr.keys(), strict = True)
53​ import pprint
17 54​
18 updates = {} 55​ updates = {}
19 for _, update in pending_ok.items(): 56​ for _, update in pending_ok.items():
20 updates[update['title']] = update 57​ updates[update['title']] = update
21 58​
22 for _, update in pending_fail.items(): 59​ for build, update in pending_fail.items():
60​ if update is None:
61​ #FIXME change print to log
62​ print "ERR: build without an update: %r" % build
63​ continue
23 if update['title'] not in updates: 64​ if update['title'] not in updates:
24 updates[update['title']] = update 65​ updates[update['title']] = update
25 66​
26 updates_results = [] 67​ updates_results = []
27 for update_title, update in updates.items(): 68​ for update_title, update in updates.items():
28 cd = check.CheckDetail(update_title, check.ReportType.BODHI_UPDATE, outcome = 'PASSED') 69​ cd = check.CheckDetail(update_title, check.ReportType.BODHI_UPDATE, outcome = 'PASSED')
29 updates_results.append(cd) 70​ updates_results.append(cd)
30 71​
72​ untested_builds = []
31 for build in update['builds']: 73​ for build in update['builds']:
74​ build = build['nvr']
75​
32 # If build is in pending_fail either when it has no assigned 76​ # If build is in pending_fail either when it has no assigned
33 # update (that can not happen, as we traverse builds of 77​ # update (that can not happen, as we traverse builds of
34 # a specific update), or when an update was incomplete 78​ # a specific update), or when an update was incomplete
35 # (e.g. missing some builds). 79​ # (e.g. missing some builds).
36 # If an update is incomplete, all it's builds are put 80​ # If an update is incomplete, all it's builds are put
37 # into the pending_fail, so this should fail early, 81​ # into the pending_fail, so this should fail early,
38 if build in pending_fail: 82​ if build in pending_fail.keys():
39 cd.update_outcome('ABORTED') 83​ cd.update_outcome('ABORTED')
40 cd.store('Update was incomplete.') 84​ cd.store('Update was incomplete.')
41 break 85​ break
42 86​
43 if build not in pending_ok: 87​ if build not in pending_ok.keys():
44 cd.update_outcome('ABORTED') 88​ cd.update_outcome('ABORTED')
45 cd.store('Update contains build that was not tested.') 89​ untested_builds.append(build)
46 break 90​ print "Untested build %s for update %s" % (build, update_title)
91​ continue
47 92​
48 if run_result[build]['result'] is not Pass: 93​ if results_nvr[build]['result'] is not Pass:
49 cd.update_outcome('FAILED') 94​ cd.update_outcome('FAILED')
50 cd.store('Build %s failed depcheck' % build) 95​ cd.store('Build %s failed depcheck' % build)
51 cd.output.extend(run_result[build]['details']) 96​ cd.output.extend(results_nvr[build]['details'])
97​
98​ if untested_builds:
99​ cd.store('Update containted untested builds: %s' % untested_builds)
52 100​
53 return updates_results 101​ return updates_results
54 102​
View Options

etc/depcheck.conf

1# Configuration file for depcheck mk 2 1​# Configuration file for depcheck mk 2
2 2​
3[configuration] 3​[configuration]
4# output format; build or update 4​# output format; build or update
5format=builds 5​format=rpms
6 6​
7 7​
8[released] 8​[released]
9metalink=https://mirrors.fedoraproject.org/metalink?repo=fedora-$release&arch=$arch 9​metalink=https://mirrors.fedoraproject.org/metalink?repo=fedora-$release&arch=$arch
10 10​
11 11​
12[updates] 12​[updates]
13metalink=https://mirrors.fedoraproject.org/metalink?repo=updates-released-f$release&arch=$arch 13​metalink=https://mirrors.fedoraproject.org/metalink?repo=updates-released-f$release&arch=$arch
14 14​
View Options

run_depcheck.py

1#!/usr/bin/python 1​#!/usr/bin/python
2 2​
3​import sys
4​import os
5​
6​sys.path.insert(0, os.path.realpath(os.path.dirname(__file__)))
7​
3import depcheck 8​import depcheck
9​from depcheck import exc
10​
11​def taskotron_run(rpms, arch, repos, report_format="rpms"):
12​ """
13​ rpms: list of rpm files and/or directories containing rpm files
14​ arch: list of arches
15​ repos: dictionary or list of dictionaries
16​ report_format: "rpms" or "updates"
17​ """
18​
19​ # FIXME: do something saner in the future - nasty fix for the moment
20​ filtered_arches = [a for a in arch if a not in ['noarch', 'all', 'src']]
21​ if len(filtered_arches) != 1:
22​ raise exc.DepcheckException("Invalid content of arches after filtering:"
23​ " %r" % filtered_arches)
24​ arch = filtered_arches[0]
25​
26​ repositories = {}
27​
28​ if isinstance(repos, list):
29​ for repo in repos:
30​ for name, url in repo.items():
31​ while name in repositories:
32​ name += "_"
33​ repositories[name] = url
34​
35​ elif isinstance(repos, dict):
36​ repositories = repos
37​
38​ else:
39​ raise exc.DepcheckException('Unsupported type of repos arg')
4 40​
5def run(rpms, arch, repos, report_format="builds"): 41​ return depcheck.run(rpms, arch, repositories, report_format)
6 return depcheck.run(rpms, arch, repos, report_format) ​
7 42​
8if __name__ == '__main__': 43​if __name__ == '__main__':
9 depcheck.depcheck_cli() 44​ depcheck.depcheck_cli()
View Options

tests/test_depcheck.py

1import pytest 1​import pytest
2from dingus import Dingus 2​from dingus import Dingus
3import __builtin__ 3​import __builtin__
4 4​
5from depcheck import depcheck 5​from depcheck import depcheck
6from depcheck import Pass, Fail 6​from depcheck import Pass, Fail
7​from depcheck import exc
7 8​
8 9​
9class TestDepcheck(object): 10​class TestDepcheck(object):
10 def test_libsolv_setup_adds_repos(self, monkeypatch): 11​ def test_libsolv_setup_adds_repos(self, monkeypatch):
11 ref_arch = 'x86_64' 12​ ref_arch = 'x86_64'
12 ref_basepath = '/path/to/testing-repo/' 13​ ref_basepath = '/path/to/testing-repo/'
13 ref_repomdfiles = {'reponame': {'primary': '%sprimary.xml.gz' % ref_basepath, 14​ ref_repomdfiles = {'reponame': {'primary': '%sprimary.xml.gz' % ref_basepath,
14 'filelists': '%sfilielists.xml.gz' % ref_basepath}} 15​ 'filelists': '%sfilielists.xml.gz' % ref_basepath}}
Show All 10 Lines
25 test_depcheck._prepare_libsolv() 26​ test_depcheck._prepare_libsolv()
26 27​
27 assert len(stub_solvrepo.calls()) == 2 28​ assert len(stub_solvrepo.calls()) == 2
28 29​
29 def test_check_rpm_deps_output(self, monkeypatch): 30​ def test_check_rpm_deps_output(self, monkeypatch):
30 """ 31​ """
31 Tests whether rpm_deps provides correct output for a given input. 32​ Tests whether rpm_deps provides correct output for a given input.
32 """ 33​ """
33 return_input = lambda x: x ​
34 34​
35 test_depcheck = depcheck.Depcheck(None, None, "fakedir") 35​ test_depcheck = depcheck.Depcheck(None, None, "fakedir")
36 monkeypatch.setattr(test_depcheck, '_prepare_libsolv', Dingus()) 36​ monkeypatch.setattr(test_depcheck, '_prepare_libsolv', Dingus())
37 monkeypatch.setattr(test_depcheck, 'rpm_filename_to_nvr', return_input) ​
38 37​
39 rpmfiles = ['rpm_1', 'rpm_2'] 38​ rpmfiles = ['rpm_1', 'rpm_2']
40 def fake_check_rpm(rpmfile): 39​ def fake_check_rpm(rpmfile):
41 results = { 40​ results = {
42 rpmfiles[0]: (Pass, []), 41​ rpmfiles[0]: (Pass, []),
43 rpmfiles[1]: (Fail, ['message']), 42​ rpmfiles[1]: (Fail, ['message']),
44 } 43​ }
45 return results.get(rpmfile, (Pass, [])) 44​ return results.get(rpmfile, (Pass, []))
46 45​
47 monkeypatch.setattr(test_depcheck, '_check_rpm', fake_check_rpm) 46​ monkeypatch.setattr(test_depcheck, '_check_rpm', fake_check_rpm)
48 47​
49 outcome = test_depcheck.check_rpm_deps(rpmfiles) 48​ outcome = test_depcheck.check_rpm_deps(rpmfiles)
50 49​
51 assert len(outcome) == 2 50​ assert len(outcome) == 2
52 assert 'rpm_1' in outcome 51​ assert 'rpm_1' in outcome
53 assert 'rpm_2' in outcome 52​ assert 'rpm_2' in outcome
54 assert outcome['rpm_1']['result'] == Pass 53​ assert outcome['rpm_1']['result'] == Pass
55 assert outcome['rpm_2']['result'] == Fail 54​ assert outcome['rpm_2']['result'] == Fail
56 assert outcome['rpm_2']['details'] == ['message'] 55​ assert outcome['rpm_2']['details'] == ['message']
57 56​
58 def test_check_rpm_deps_conflicting_results(self, monkeypatch): 57​ def test_check_rpm_deps_conflicting_results(self, monkeypatch):
59 """ 58​ """
60 Checks whether rpm_deps raises Exceptions when conflicting 59​ Checks whether rpm_deps raises Exceptions when conflicting
61 results are provided for one NVR. 60​ results are provided for one RPM.
62 """ 61​ """
63 return_input = lambda x: x ​
64 62​
65 test_depcheck = depcheck.Depcheck(None, None, "fakedir") 63​ test_depcheck = depcheck.Depcheck(None, None, "fakedir")
66 monkeypatch.setattr(test_depcheck, '_prepare_libsolv', Dingus()) 64​ monkeypatch.setattr(test_depcheck, '_prepare_libsolv', Dingus())
67 monkeypatch.setattr(test_depcheck, 'rpm_filename_to_nvr', return_input) ​
68 65​
69 self.tmp_result = Pass 66​ self.tmp_result = Pass
70 def fake_check_rpm(rpmfile): 67​ def fake_check_rpm(rpmfile):
71 if self.tmp_result == Pass: 68​ if self.tmp_result == Pass:
72 self.tmp_result = Fail 69​ self.tmp_result = Fail
73 else: 70​ else:
74 self.tmp_result = Pass 71​ self.tmp_result = Pass
75 return (self.tmp_result, []) 72​ return (self.tmp_result, [])
76 73​
77 monkeypatch.setattr(test_depcheck, '_check_rpm', fake_check_rpm) 74​ monkeypatch.setattr(test_depcheck, '_check_rpm', fake_check_rpm)
78 75​
79 with pytest.raises(depcheck.DepcheckException): 76​ with pytest.raises(exc.DepcheckException):
80 outcome = test_depcheck.check_rpm_deps(['rpm1', 'rpm1']) 77​ outcome = test_depcheck.check_rpm_deps(['rpm1', 'rpm1'])
81 78​
View Options

tests/test_squash_results.py

1import pytest 1​import pytest
2from dingus import Dingus 2​from dingus import Dingus
3import __builtin__ 3​import __builtin__
4 4​
5from depcheck.squash_results import squash_build_results_to_updates 5​from depcheck import squash_results
6​from depcheck.squash_results import squash_rpm_results_to_updates
6from depcheck import Pass, Fail 7​from depcheck import Pass, Fail
7 8​
8 9​
9class TestSquashResults(object): 10​class TestSquashResults(object):
10 def test_squash_build_results_to_updates_all_ok(self): 11​ def test_squash_rpm_results_to_updates_all_ok(self, monkeypatch):
11 """ 12​ """
12 Checks whether _squash_build_results_to_updates provides correct output 13​ Checks whether _squash_rpm_results_to_updates provides correct output
13 when all the builds in an update passed depchecking. 14​ when all the builds in an update passed depchecking.
14 """ 15​ """
16​
17​ return_input = lambda x: x
18​ monkeypatch.setattr(squash_results, 'rpm_filename_to_nvr', return_input)
19​ monkeypatch.setattr(squash_results, 'squash_rpms_to_builds', return_input)
20​
15 run_results = { 21​ run_results = {
16 'nvr1': {'result': Pass, 'details': []}, 22​ 'nvr1': {'result': Pass, 'details': []},
17 'nvr2': {'result': Pass, 'details': []}, 23​ 'nvr2': {'result': Pass, 'details': []},
18 } 24​ }
19 update = {'title': 'update_name', 'builds': run_results.keys()} 25​ update = {'title': 'update_name', 'builds': [{'nvr': k} for k in run_results.keys()]}
20 build_2_update = ( 26​ build_2_update = (
21 { 27​ {
22 'nvr1': update, 28​ 'nvr1': update,
23 'nvr2': update, 29​ 'nvr2': update,
24 }, {}) 30​ }, {})
25 31​
26 fake_bodhi = Dingus(build2update__returns = build_2_update) 32​ fake_bodhi = Dingus(build2update__returns = build_2_update)
27 33​
28 outcome = squash_build_results_to_updates(run_results, fake_bodhi) 34​ outcome = squash_rpm_results_to_updates(run_results, fake_bodhi)
29 assert len(outcome) == 1 35​ assert len(outcome) == 1
30 36​
31 detail = outcome[0] 37​ detail = outcome[0]
32 assert detail.item == update['title'] 38​ assert detail.item == update['title']
33 assert detail.outcome == 'PASSED' 39​ assert detail.outcome == 'PASSED'
34 40​
35 def test_squash_build_results_to_updates_build_failed(self): 41​ def test_squash_rpm_results_to_updates_build_failed(self, monkeypatch):
36 """ 42​ """
37 Checks whether _squash_build_results_to_updates provides correct output 43​ Checks whether _squash_rpm_results_to_updates provides correct output
38 when some of the builds in an update failed depchecking. 44​ when some of the builds in an update failed depchecking.
39 """ 45​ """
46​ return_input = lambda x: x
47​ monkeypatch.setattr(squash_results, 'rpm_filename_to_nvr', return_input)
48​ monkeypatch.setattr(squash_results, 'squash_rpms_to_builds', return_input)
49​
40 run_results = { 50​ run_results = {
41 'nvr1': {'result': Pass, 'details': []}, 51​ 'nvr1': {'result': Pass, 'details': []},
42 'nvr2': {'result': Fail, 'details': ['### TEST @@@']}, 52​ 'nvr2': {'result': Fail, 'details': ['### TEST @@@']},
43 } 53​ }
44 update = {'title': 'update_name', 'builds': run_results.keys()} 54​ update = {'title': 'update_name', 'builds': [{'nvr': k} for k in run_results.keys()]}
45 build_2_update = ( 55​ build_2_update = (
46 { 56​ {
47 'nvr1': update, 57​ 'nvr1': update,
48 'nvr2': update, 58​ 'nvr2': update,
49 }, {}) 59​ }, {})
50 60​
51 fake_bodhi = Dingus(build2update__returns = build_2_update) 61​ fake_bodhi = Dingus(build2update__returns = build_2_update)
52 62​
53 outcome = squash_build_results_to_updates(run_results, fake_bodhi) 63​ outcome = squash_rpm_results_to_updates(run_results, fake_bodhi)
54 assert len(outcome) == 1 64​ assert len(outcome) == 1
55 65​
56 detail = outcome[0] 66​ detail = outcome[0]
57 assert detail.item == update['title'] 67​ assert detail.item == update['title']
58 assert detail.outcome == 'FAILED' 68​ assert detail.outcome == 'FAILED'
59 assert '### TEST @@@' in detail.output 69​ assert '### TEST @@@' in detail.output
60 70​
61 def test_squash_build_results_to_updates_incomplete_update(self): 71​ def test_squash_rpm_results_to_updates_incomplete_update(self, monkeypatch):
62 """ 72​ """
63 Checks whether _squash_build_results_to_updates provides correct output 73​ Checks whether _squash_rpm_results_to_updates provides correct output
64 when some update is incomplete (i.e. present in the second return-value 74​ when some update is incomplete (i.e. present in the second return-value
65 from the build2update method). 75​ from the build2update method).
66 """ 76​ """
77​ return_input = lambda x: x
78​ monkeypatch.setattr(squash_results, 'rpm_filename_to_nvr', return_input)
79​ monkeypatch.setattr(squash_results, 'squash_rpms_to_builds', return_input)
80​
67 run_results = { 81​ run_results = {
68 'nvr1': {'result': Pass, 'details': []}, 82​ 'nvr1': {'result': Pass, 'details': []},
69 'nvr2': {'result': Pass, 'details': []}, 83​ 'nvr2': {'result': Pass, 'details': []},
70 } 84​ }
71 update = {'title': 'update_name', 'builds': run_results.keys()} 85​ update = {'title': 'update_name', 'builds': [{'nvr': k} for k in run_results.keys()]}
72 build_2_update = ({}, 86​ build_2_update = ({},
73 { 87​ {
74 'nvr1': update, 88​ 'nvr1': update,
75 'nvr2': update, 89​ 'nvr2': update,
76 }) 90​ })
77 91​
78 fake_bodhi = Dingus(build2update__returns = build_2_update) 92​ fake_bodhi = Dingus(build2update__returns = build_2_update)
79 93​
80 outcome = squash_build_results_to_updates(run_results, fake_bodhi) 94​ outcome = squash_rpm_results_to_updates(run_results, fake_bodhi)
81 assert len(outcome) == 1 95​ assert len(outcome) == 1
82 96​
83 detail = outcome[0] 97​ detail = outcome[0]
84 assert detail.item == update['title'] 98​ assert detail.item == update['title']
85 assert detail.outcome == 'ABORTED' 99​ assert detail.outcome == 'ABORTED'
86 assert 'Update was incomplete.' in detail.output 100​ assert 'Update was incomplete.' in detail.output
87 101​
88 def test_squash_build_results_to_updates_missing_build(self): 102​ def test_squash_rpm_results_to_updates_missing_build(self, monkeypatch):
89 """ 103​ """
90 Checks whether _squash_build_results_to_updates provides correct output 104​ Checks whether _squash_rpm_results_to_updates provides correct output
91 when update is missing a result for one of its builds. 105​ when update is missing a result for one of its builds.
92 """ 106​ """
107​ return_input = lambda x: x
108​ monkeypatch.setattr(squash_results, 'rpm_filename_to_nvr', return_input)
109​ monkeypatch.setattr(squash_results, 'squash_rpms_to_builds', return_input)
110​
93 run_results = { 111​ run_results = {
94 'nvr1': {'result': Pass, 'details': []}, 112​ 'nvr1': {'result': Pass, 'details': []},
95 'nvr2': {'result': Pass, 'details': []}, 113​ 'nvr2': {'result': Pass, 'details': []},
96 } 114​ }
97 update = {'title': 'update_name', 'builds': run_results.keys()} 115​ update = {'title': 'update_name', 'builds': [{'nvr': k} for k in run_results.keys()]}
98 update['builds'].append('nvr3') 116​ update['builds'].append({'nvr':'nvr3'})
99 build_2_update = ( 117​ build_2_update = (
100 { 118​ {
101 'nvr1': update, 119​ 'nvr1': update,
102 'nvr2': update, 120​ 'nvr2': update,
103 }, {}) 121​ }, {})
104 122​
105 fake_bodhi = Dingus(build2update__returns = build_2_update) 123​ fake_bodhi = Dingus(build2update__returns = build_2_update)
106 124​
107 outcome = squash_build_results_to_updates(run_results, fake_bodhi) 125​ outcome = squash_rpm_results_to_updates(run_results, fake_bodhi)
108 assert len(outcome) == 1 126​ assert len(outcome) == 1
109 127​
110 detail = outcome[0] 128​ detail = outcome[0]
111 assert detail.item == update['title'] 129​ assert detail.item == update['title']
112 assert detail.outcome == 'ABORTED' 130​ assert detail.outcome == 'ABORTED'
113 assert 'Update contains build that was not tested.' in detail.output 131​ assert "Update containted untested builds: ['nvr3']" in detail.output