Differential D71

add koji_utils.getNEVR() and rpm_utils.cmpNEVR()
ClosedPublic
Actions

Authored by kparal on Apr 25 2014, 2:22 PM.

Details

Reviewers
tflink
mkrizek
pschindl
jskladan
Maniphest Tasks
T79: Port upgradepath from AutoQA to Taskotron
Commits
rLTRNab7377f4ebf1: add koji_utils.getNEVR() and rpm_utils.cmpNEVR()
rLTRNbd65c58461c1: add koji_utils.getNEVR() and rpm_utils.cmpNEVR()
Summary

These two methods are needed for upgradepath, but they can be generally
useful for any check or library call working with epochs.

Test Plan

Tests added.

Diff Detail

Lint
Lint Skipped
Unit
Unit Tests Skipped
tflink accepted this revision.Apr 25 2014, 2:39 PM

Looks good to me. Please merge into develop

kparal closed this revision.Apr 25 2014, 2:52 PM

Closed by commit rLTRNab7377f4ebf1 (authored by @kparal).

Revision Contents

Diff 200

View Options

libtaskotron/koji_utils.py

1​# -*- coding: utf-8 -*-
2​# Copyright 2014, Red Hat, Inc.
3​# License: GNU General Public License version 2 or later
4​
5​''' Utility methods related to Koji '''
6​
7​import collections
1import koji 8​import koji
9​import hawkey
2 10​
3from libtaskotron import file_utils 11​from libtaskotron import file_utils
4from libtaskotron.logger import log 12​from libtaskotron.logger import log
5from libtaskotron.exceptions import TaskotronRemoteError 13​from libtaskotron import exceptions as exc
14​from libtaskotron import rpm_utils
6 15​
7KOJI_URL = 'http://koji.fedoraproject.org/kojihub' 16​KOJI_URL = 'http://koji.fedoraproject.org/kojihub'
8PKG_URL = 'http://kojipkgs.fedoraproject.org/packages' 17​PKG_URL = 'http://kojipkgs.fedoraproject.org/packages'
9 18​
10class KojiClient(object): 19​class KojiClient(object):
11 def __init__(self, koji_session=None): 20​ def __init__(self, koji_session=None):
12 self.session = koji_session 21​ self.session = koji_session
13 22​
Show All 16 Lines29 def nvr_to_urls(self, nvr, arches=None, debuginfo=False, src=True):
30 ksession = self.get_koji_session() 39​ ksession = self.get_koji_session()
31 40​
32 # add i686 arch if i386 is present in arches 41​ # add i686 arch if i386 is present in arches
33 if arches and 'i386' in arches and 'i686' not in arches: 42​ if arches and 'i386' in arches and 'i686' not in arches:
34 arches.append('i686') 43​ arches.append('i686')
35 44​
36 info = ksession.getBuild(nvr) 45​ info = ksession.getBuild(nvr)
37 if info is None: 46​ if info is None:
38 raise TaskotronRemoteError( 47​ raise exc.TaskotronRemoteError(
39 "No build information found for %s" % nvr) 48​ "No build information found for %s" % nvr)
40 49​
41 baseurl = '/'.join((PKG_URL, info['package_name'], 50​ baseurl = '/'.join((PKG_URL, info['package_name'],
42 info['version'], info['release'])) 51​ info['version'], info['release']))
43 52​
44 rpms = ksession.listRPMs(buildID=info['id'], arches=arches) 53​ rpms = ksession.listRPMs(buildID=info['id'], arches=arches)
45 if not debuginfo: 54​ if not debuginfo:
46 rpms = [r for r in rpms if not r['name'].endswith('-debuginfo')] 55​ rpms = [r for r in rpms if not r['name'].endswith('-debuginfo')]
47 if not src: 56​ if not src:
48 rpms = [r for r in rpms if not r['arch'] == 'src'] 57​ rpms = [r for r in rpms if not r['arch'] == 'src']
49 58​
50 urls = ['%s/%s' % (baseurl, koji.pathinfo.rpm(r)) for r in rpms] 59​ urls = ['%s/%s' % (baseurl, koji.pathinfo.rpm(r)) for r in rpms]
51 if len(urls) == 0: 60​ if len(urls) == 0:
52 raise TaskotronRemoteError('No RPMs found for %s' % nvr) 61​ raise exc.TaskotronRemoteError('No RPMs found for %s' % nvr)
53 62​
54 return sorted(urls) 63​ return sorted(urls)
55 64​
56 def get_nvr_rpms(self, nvr, rpm_dir, arches=None, debuginfo=False, 65​ def get_nvr_rpms(self, nvr, rpm_dir, arches=None, debuginfo=False,
57 src=False): 66​ src=False):
58 '''Retrieve the RPMs associated with a build NVR into the specified 67​ '''Retrieve the RPMs associated with a build NVR into the specified
59 directory. 68​ directory.
60 69​
Show All 27 Lines90 def get_tagged_rpms(self, tag, dest, arches):
88 opts = {} 97​ opts = {}
89 tag_data = ksession.listTagged(tag, **opts) 98​ tag_data = ksession.listTagged(tag, **opts)
90 99​
91 nvrs = ["%(nvr)s" % x for x in tag_data] 100​ nvrs = ["%(nvr)s" % x for x in tag_data]
92 rpms = [] 101​ rpms = []
93 for nvr in nvrs: 102​ for nvr in nvrs:
94 rpms.append(self.get_nvr_rpms(nvr, dest, arches)) 103​ rpms.append(self.get_nvr_rpms(nvr, dest, arches))
95 return rpms 104​ return rpms
105​
106​
107​def getNEVR(build):
108​ '''Extract RPM version identifier in NEVR format from Koji build object
109​
110​ @param build Koji buildinfo dictionary (as returned from @module koji
111​ methods like getBuild())
112​ @return NEVR string; epoch is included when non-zero, otherwise omitted
113​ @raise TaskotronValueError if @param build is of incorrect type
114​ '''
115​ # validate input
116​ if (not isinstance(build, collections.Mapping) or 'nvr' not in build or
117​ 'epoch' not in build):
118​ raise exc.TaskotronValueError("Input argument doesn't look like "
119​ "a Koji build object: %s" % build)
120​
121​ rpmver = hawkey.split_nevra(build['nvr'] + '.noarch')
122​ rpmver.epoch = build['epoch'] or 0
123​ nevr = '%s-%s' % (rpmver.name, rpmver.evr())
124​ # supress 0 epoch (evr() method always includes epoch, even if 0)
125​ nevr = rpm_utils.rpmformat(nevr, fmt='nevr')
126​
127​ return nevr
View Options

libtaskotron/rpm_utils.py

1# -*- coding: utf-8 -*- 1​# -*- coding: utf-8 -*-
2# Copyright 2014, Red Hat, Inc. 2​# Copyright 2014, Red Hat, Inc.
3# License: GNU General Public License version 2 or later 3​# License: GNU General Public License version 2 or later
4 4​
5''' Utility methods related to RPM and YUM/DNF ''' 5​''' Utility methods related to RPM '''
6 6​
7import os 7​import os
8import hawkey 8​import hawkey
9 9​
10from . import exceptions as exc 10​from libtaskotron import exceptions as exc
11 11​
12 12​
13def basearch(arch=None): 13​def basearch(arch=None):
14 ''' 14​ '''
15 This returns the 'base' architecture identifier for a specified architecture 15​ This returns the 'base' architecture identifier for a specified architecture
16 (e.g. i386 for i[3-6]86), to be used by YUM etc. 16​ (e.g. i386 for i[3-6]86), to be used by YUM etc.
17 17​
18 @param arch an arch (string) to be converted to a basearch. If None, then 18​ @param arch an arch (string) to be converted to a basearch. If None, then
▲ Show 20 Lines • Show All 58 Lines • ▼ Show 20 Line(s)29def rpmformat(rpmstr, fmt='nvr', end_arch=False):
77 77​
78 result = '%s-%s' % (nevra.name, evr) 78​ result = '%s-%s' % (nevra.name, evr)
79 79​
80 # append arch if requested 80​ # append arch if requested
81 if 'a' in fmt: 81​ if 'a' in fmt:
82 result += '.' + nevra.arch 82​ result += '.' + nevra.arch
83 83​
84 return result 84​ return result
85​
86​
87​def cmpNEVR(nevr1, nevr2):
88​ '''Compare two RPM version identifiers in NEVR format.
89​
90​ @param nevr1 a string in N(E)VR format
91​ @param nevr2 a string in N(E)VR format
92​ @return -1/0/1 if nevr1 < nevr2 / nevr1 == nevr2 / nevr1 > nevr2
93​ @raise TaskotronValueError if name in nevr1 doesn't match name in nevr2
94​ '''
95​ rpmver1 = hawkey.split_nevra(nevr1 + '.noarch')
96​ rpmver2 = hawkey.split_nevra(nevr2 + '.noarch')
97​
98​ if rpmver1.name != rpmver2.name:
99​ raise exc.TaskotronValueError("Name in nevr1 doesn't match name in "
100​ "nevr2: %s, %s" % (nevr1, nevr2))
101​
102​ # sack is needed for the comparison, because it can be influence the
103​ # comparison (for example epoch can be set to be ignored). A default empty
104​ # sack should match Fedora customs
105​ sack = hawkey.Sack()
106​
107​ return rpmver1.evr_cmp(rpmver2, sack)
View Options

testing/test_koji_utils.py

1​# -*- coding: utf-8 -*-
2​# Copyright 2014, Red Hat, Inc.
3​# License: GNU General Public License version 2 or later
4​
5​'''Unit tests for libtaskotron/koji_utils.py'''
6​
1import pytest 7​import pytest
2from dingus import Dingus 8​from dingus import Dingus
3 9​
4from libtaskotron import koji_utils 10​from libtaskotron import koji_utils
11​from libtaskotron import exceptions as exc
5 12​
6class TestKojiClient(): 13​class TestKojiClient():
7 def setup_method(self, method): 14​ def setup_method(self, method):
8 self.ref_nvr = 'foo-1.2-3.fc99' 15​ self.ref_nvr = 'foo-1.2-3.fc99'
9 self.ref_arch = 'noarch' 16​ self.ref_arch = 'noarch'
10 self.ref_name = 'foo' 17​ self.ref_name = 'foo'
11 self.ref_version = '1.2' 18​ self.ref_version = '1.2'
12 self.ref_release = '3.fc99' 19​ self.ref_release = '3.fc99'
▲ Show 20 Lines • Show All 57 Lines • ▼ Show 20 Line(s)
70 def should_throw_exception_norpms(self): 77​ def should_throw_exception_norpms(self):
71 stub_koji = Dingus(getBuild__returns = self.ref_build) 78​ stub_koji = Dingus(getBuild__returns = self.ref_build)
72 79​
73 test_koji = koji_utils.KojiClient(stub_koji) 80​ test_koji = koji_utils.KojiClient(stub_koji)
74 81​
75 with pytest.raises(Exception): 82​ with pytest.raises(Exception):
76 test_koji.nvr_to_urls(self.ref_nvr, arches = [self.ref_arch]) 83​ test_koji.nvr_to_urls(self.ref_nvr, arches = [self.ref_arch])
77 84​
85​
86​class TestGetENVR(object):
87​
88​ def test_epoch(self):
89​ '''Epoch included in build'''
90​ build = {'nvr': 'foo-1.2-3.fc20',
91​ 'epoch': 1,
92​ }
93​ assert koji_utils.getNEVR(build) == 'foo-1:1.2-3.fc20'
94​
95​ def test_no_epoch(self):
96​ '''Epoch not included in build'''
97​ build = {'nvr': 'foo-1.2-3.fc20',
98​ 'epoch': None,
99​ }
100​ assert koji_utils.getNEVR(build) == 'foo-1.2-3.fc20'
101​
102​ def test_raise(self):
103​ '''Invalid input param'''
104​ with pytest.raises(exc.TaskotronValueError):
105​ koji_utils.getNEVR('foo-1.2-3.fc20')
View Options

testing/test_rpm_utils.py

1# -*- coding: utf-8 -*- 1​# -*- coding: utf-8 -*-
2# Copyright 2014, Red Hat, Inc. 2​# Copyright 2014, Red Hat, Inc.
3# License: GNU General Public License version 2 or later 3​# License: GNU General Public License version 2 or later
4 4​
5'''Unit tests for libtaskotron/rpm_utils.py''' 5​'''Unit tests for libtaskotron/rpm_utils.py'''
6 6​
7import pytest 7​import pytest
8 8​
9from libtaskotron.rpm_utils import basearch, rpmformat 9​from libtaskotron.rpm_utils import basearch, rpmformat, cmpNEVR
10from libtaskotron import exceptions as exc 10​from libtaskotron import exceptions as exc
11 11​
12 12​
13class TestBasearch: 13​class TestBasearch:
14 14​
15 def test_basearch_i386(self): 15​ def test_basearch_i386(self):
16 '''i386-like archs''' 16​ '''i386-like archs'''
17 assert basearch('i386') == 'i386' 17​ assert basearch('i386') == 'i386'
▲ Show 20 Lines • Show All 84 Lines • ▼ Show 20 Line(s)100 def test_raise(self):
102 with pytest.raises(exc.TaskotronValueError): 102​ with pytest.raises(exc.TaskotronValueError):
103 rpmformat('foo-1.2-3.fc20', 'x') 103​ rpmformat('foo-1.2-3.fc20', 'x')
104 104​
105 with pytest.raises(exc.TaskotronValueError): 105​ with pytest.raises(exc.TaskotronValueError):
106 rpmformat('foo-1.2-3.fc20', 'envra') 106​ rpmformat('foo-1.2-3.fc20', 'envra')
107 107​
108 with pytest.raises(exc.TaskotronValueError): 108​ with pytest.raises(exc.TaskotronValueError):
109 rpmformat('foo-1.2-3.fc20', 'n-v-r') 109​ rpmformat('foo-1.2-3.fc20', 'n-v-r')
110​
111​
112​class TestCmpNEVR(object):
113​
114​ def test_no_epoch(self):
115​ '''Both params without epoch'''
116​ assert cmpNEVR('foo-1.2-3.fc20', 'foo-1.2-2.fc20') == 1
117​ assert cmpNEVR('foo-1.2-3.fc20', 'foo-1.2-3.fc20') == 0
118​ assert cmpNEVR('foo-1.2-3.fc20', 'foo-1.2-4.fc20') == -1
119​
120​ assert cmpNEVR('foo-1.2-3.fc20', 'foo-1.1-4.fc20') == 1
121​ assert cmpNEVR('foo-1.2-3.fc20', 'foo-2.1-1.fc20') == -1
122​ assert cmpNEVR('foo-1.2-3.fc20', 'foo-1.2-3.fc19') == 1
123​
124​ def test_epoch(self):
125​ '''Both params with epoch'''
126​ assert cmpNEVR('foo-1:1.2-3.fc20', 'foo-1:1.2-2.fc20') == 1
127​ assert cmpNEVR('foo-1:1.2-3.fc20', 'foo-1:1.2-3.fc20') == 0
128​ assert cmpNEVR('foo-0:1.2-3.fc20', 'foo-0:1.2-3.fc20') == 0
129​ assert cmpNEVR('foo-3:1.2-3.fc20', 'foo-3:1.2-4.fc20') == -1
130​
131​ assert cmpNEVR('foo-2:1.2-3.fc20', 'foo-2:1.1-4.fc20') == 1
132​ assert cmpNEVR('foo-2:1.2-3.fc20', 'foo-2:2.1-1.fc20') == -1
133​ assert cmpNEVR('foo-2:1.2-3.fc20', 'foo-2:1.2-3.fc19') == 1
134​
135​ assert cmpNEVR('foo-1:1.2-3.fc20', 'foo-2:1.2-3.fc20') == -1
136​ assert cmpNEVR('foo-1:1.2-3.fc20', 'foo-0:1.2-3.fc20') == 1
137​ assert cmpNEVR('foo-1:1.2-3.fc20', 'foo-0:2.2-3.fc20') == 1
138​
139​ def test_some_epoch(self):
140​ '''One param with epoch'''
141​ assert cmpNEVR('foo-1:1.2-3.fc20', 'foo-1.2-3.fc20') == 1
142​ assert cmpNEVR('foo-1:1.2-3.fc20', 'foo-2.2-3.fc20') == 1
143​ assert cmpNEVR('foo-0:1.2-3.fc20', 'foo-1.2-3.fc20') == 0
144​ assert cmpNEVR('foo-1.2-3.fc20', 'foo-2:0.1-1.fc19') == -1
145​
146​ def test_raise(self):
147​ '''Invalid input param'''
148​ with pytest.raises(exc.TaskotronValueError):
149​ cmpNEVR('foo-1.2-3.fc20', 'bar-1.2-3.fc20')