21511583274

Committed 30 Jan 2026 09:50AM UTC coverage: 88.271% (+1.3%) from 86.948%

Build # 21511583274

Build Type

push

github

Run Details

6013 of 6812 relevant lines covered (88.27%)

4.31 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.5

/rpmlint/rpmdiff.py

import contextlib
from itertools import chain
import pathlib
import sys
import tempfile

import rpm
from rpmlint.helpers import byte_to_string, print_warning
from rpmlint.pkg import get_installed_pkgs, Pkg


class Rpmdiff:
    # constants
    TAGS = (rpm.RPMTAG_NAME, rpm.RPMTAG_SUMMARY,
            rpm.RPMTAG_DESCRIPTION, rpm.RPMTAG_GROUP,
            rpm.RPMTAG_LICENSE, rpm.RPMTAG_URL,
            rpm.RPMTAG_PREIN, rpm.RPMTAG_POSTIN,
            rpm.RPMTAG_PREUN, rpm.RPMTAG_POSTUN,
            rpm.RPMTAG_PRETRANS, rpm.RPMTAG_POSTTRANS)

    PRCO = ('REQUIRES', 'PROVIDES', 'CONFLICTS', 'OBSOLETES',
            'RECOMMENDS', 'SUGGESTS', 'ENHANCES', 'SUPPLEMENTS')

    __FILEIDX = [['S', 'size'],
                 ['M', 'mode'],
                 ['5', 'digest'],
                 ['D', 'rdev'],
                 ['N', 'nlink'],
                 ['L', 'state'],
                 ['V', 'vflags'],
                 ['U', 'user'],
                 ['G', 'group'],
                 ['F', 'fflags'],
                 ['T', 'mtime']]

    DEPFORMAT = '%-12s%s %s %s %s'
    FORMAT = '%-12s%s'

    ADDED = 'added'
    REMOVED = 'removed'

    def __init__(self, old, new, ignore=None, exclude=None):
        self.result = []
        self.ignore = ignore or []
        self.exclude = exclude or []

        FILEIDX = self.__FILEIDX
        for tag in self.ignore:
            for entry in FILEIDX:
                if tag == entry[0]:
                    entry[1] = None
                    break

        try:
            old = self.__load_pkg(old).header
            new = self.__load_pkg(new).header
        except KeyError as e:
            print_warning(str(e))
            sys.exit(2)

        # Compare single tags
        for tag in self.TAGS:
            old_tag = old[tag]
            new_tag = new[tag]
            if old_tag != new_tag:
                tagname = rpm.tagnames[tag]
                if old_tag is None:
                    self.__add(self.FORMAT, (self.ADDED, tagname))
                elif new_tag is None:
                    self.__add(self.FORMAT, (self.REMOVED, tagname))
                else:
                    self.__add(self.FORMAT, ('S.5.....', tagname))

        # compare Provides, Requires, ...
        for tag in self.PRCO:
            self.__comparePRCOs(old, new, tag)

        # compare the files
        old_files_dict = self.__fileIteratorToDict(rpm.files(old))
        new_files_dict = self.__fileIteratorToDict(rpm.files(new))
        files = sorted(set(chain(iter(old_files_dict), iter(new_files_dict))))

        for f in files:
            if self._excluded(f):
                continue

            diff = False

            old_file = old_files_dict.get(f)
            new_file = new_files_dict.get(f)

            if not old_file:
                self.__add(self.FORMAT, (self.ADDED, f))
            elif not new_file:
                self.__add(self.FORMAT, (self.REMOVED, f))
            else:
                fmt = ''
                for entry in FILEIDX:
                    if entry[1] is not None and \
                            getattr(old_file, entry[1]) != getattr(new_file, entry[1]):
                        fmt += entry[0]
                        diff = True
                    else:
                        fmt += '.'
                if diff:
                    self.__add(self.FORMAT, (fmt, f))

    def _excluded(self, f):
        f = pathlib.PurePath(f)
        for glob in self.exclude:
            if f.match(glob):
                return True
            if glob.startswith('/'):
                for parent in f.parents:
                    if parent.match(glob):
                        return True
        return False

    # return a report of the differences
    def textdiff(self):
        return '\n'.join((fmt % data for fmt, data in self.result))

    # do the two rpms differ
    def differs(self):
        return bool(self.result)

    # add one differing item
    def __add(self, fmt, data):
        self.result.append((fmt, data))

    # load a package from a file or from the installed ones
    def __load_pkg(self, name):
        # FIXME: redo to try file/installed and proceed based on that, or pick
        # one of the selected first
        tmpdir = tempfile.gettempdir()
        with contextlib.suppress(TypeError):
            if name.is_file():
                return Pkg(name, tmpdir)
        inst = get_installed_pkgs(str(name))
        if not inst:
            raise KeyError(f'No installed packages by name {name}')
        if len(inst) > 1:
            raise KeyError(f'More than one installed packages by name {name}')
        return inst[0]

    # output the right string according to RPMSENSE_* const
    def sense2str(self, sense):
        s = ''
        for tag, char in ((rpm.RPMSENSE_LESS, '<'),
                          (rpm.RPMSENSE_GREATER, '>'),
                          (rpm.RPMSENSE_EQUAL, '=')):
            if sense & tag:
                s += char
        return s

    # output the right requires string according to RPMSENSE_* const
    def req2str(self, req):
        s = 'REQUIRES'
        # we want to use 64 even with rpm versions that define RPMSENSE_PREREQ
        # as 0 to get sane results when comparing packages built with an old
        # (64) version and a new (0) one
        if req & (rpm.RPMSENSE_PREREQ or 64):
            s = 'PREREQ'

        ss = []
        if req & rpm.RPMSENSE_SCRIPT_PRE:
            ss.append('pre')
        if req & rpm.RPMSENSE_SCRIPT_POST:
            ss.append('post')
        if req & rpm.RPMSENSE_SCRIPT_PREUN:
            ss.append('preun')
        if req & rpm.RPMSENSE_SCRIPT_POSTUN:
            ss.append('postun')
        if req & getattr(rpm, 'RPMSENSE_PRETRANS', 1 << 7):  # rpm >= 4.9.0
            ss.append('pretrans')
        if req & getattr(rpm, 'RPMSENSE_POSTTRANS', 1 << 5):  # rpm >= 4.9.0
            ss.append('posttrans')
        if ss:
            s += '(%s)' % ','.join(ss)

        return s

    # compare Provides, Requires, Conflicts, Obsoletes
    def __comparePRCOs(self, old, new, name):
        try:
            oldflags = old[name[:-1] + 'FLAGS']
        except ValueError:
            # assume tag not supported, e.g. Recommends with older rpm
            return
        newflags = new[name[:-1] + 'FLAGS']
        # fix buggy rpm binding not returning list for single entries
        if not isinstance(oldflags, list):
            oldflags = [oldflags]
        if not isinstance(newflags, list):
            newflags = [newflags]

        o = zip(old[name], oldflags, old[name[:-1] + 'VERSION'])
        if not isinstance(o, list):
            o = list(o)
        n = zip(new[name], newflags, new[name[:-1] + 'VERSION'])
        if not isinstance(n, list):
            n = list(n)

        # filter self provides, TODO: self %name(%_isa) as well
        if name == 'PROVIDES':
            oldE = old['epoch'] is not None and str(old['epoch']) + ':' or ''
            oldV = '{}{}'.format(oldE, old.format('%{VERSION}-%{RELEASE}'))
            oldNV = (old['name'], rpm.RPMSENSE_EQUAL, oldV.encode())
            newE = new['epoch'] is not None and str(new['epoch']) + ':' or ''
            newV = '{}{}'.format(newE, new.format('%{VERSION}-%{RELEASE}'))
            newNV = (new['name'], rpm.RPMSENSE_EQUAL, newV.encode())
            o = [entry for entry in o if entry != oldNV]
            n = [entry for entry in n if entry != newNV]

        for oldentry in o:
            if oldentry not in n:
                namestr = name
                if namestr == 'REQUIRES':
                    namestr = self.req2str(oldentry[1])
                self.__add(self.DEPFORMAT,
                           (self.REMOVED, namestr, byte_to_string(oldentry[0]),
                            self.sense2str(oldentry[1]), byte_to_string(oldentry[2])))
        for newentry in n:
            if newentry not in o:
                namestr = name
                if namestr == 'REQUIRES':
                    namestr = self.req2str(newentry[1])
                self.__add(self.DEPFORMAT,
                           (self.ADDED, namestr, byte_to_string(newentry[0]),
                            self.sense2str(newentry[1]), byte_to_string(newentry[2])))

    def __fileIteratorToDict(self, fi):
        result = {}
        for filedata in fi:
            result[filedata.name] = filedata
        return result

1	import contextlib	5✔
2	from itertools import chain	5✔
3	import pathlib	5✔
4	import sys	5✔
5	import tempfile	5✔
6
7	import rpm	5✔
8	from rpmlint.helpers import byte_to_string, print_warning	5✔
9	from rpmlint.pkg import get_installed_pkgs, Pkg	5✔
10
11
12	class Rpmdiff:	5✔
13	# constants
14	TAGS = (rpm.RPMTAG_NAME, rpm.RPMTAG_SUMMARY,	5✔
15	rpm.RPMTAG_DESCRIPTION, rpm.RPMTAG_GROUP,
16	rpm.RPMTAG_LICENSE, rpm.RPMTAG_URL,
17	rpm.RPMTAG_PREIN, rpm.RPMTAG_POSTIN,
18	rpm.RPMTAG_PREUN, rpm.RPMTAG_POSTUN,
19	rpm.RPMTAG_PRETRANS, rpm.RPMTAG_POSTTRANS)
20
21	PRCO = ('REQUIRES', 'PROVIDES', 'CONFLICTS', 'OBSOLETES',	5✔
22	'RECOMMENDS', 'SUGGESTS', 'ENHANCES', 'SUPPLEMENTS')
23
24	__FILEIDX = [['S', 'size'],	5✔
25	['M', 'mode'],
26	['5', 'digest'],
27	['D', 'rdev'],
28	['N', 'nlink'],
29	['L', 'state'],
30	['V', 'vflags'],
31	['U', 'user'],
32	['G', 'group'],
33	['F', 'fflags'],
34	['T', 'mtime']]
35
36	DEPFORMAT = '%-12s%s %s %s %s'	5✔
37	FORMAT = '%-12s%s'	5✔
38
39	ADDED = 'added'	5✔
40	REMOVED = 'removed'	5✔
41
42	def __init__(self, old, new, ignore=None, exclude=None):	5✔
43	self.result = []	5✔
44	self.ignore = ignore or []	5✔
45	self.exclude = exclude or []	5✔
46
47	FILEIDX = self.__FILEIDX	5✔
48	for tag in self.ignore:	5✔
49	for entry in FILEIDX:	5✔
50	if tag == entry[0]:	5✔
51	entry[1] = None	5✔
52	break	5✔
53
54	try:	5✔
55	old = self.__load_pkg(old).header	5✔
56	new = self.__load_pkg(new).header	5✔
57	except KeyError as e:	×
58	print_warning(str(e))	×
59	sys.exit(2)	×
60
61	# Compare single tags
62	for tag in self.TAGS:	5✔
63	old_tag = old[tag]	5✔
64	new_tag = new[tag]	5✔
65	if old_tag != new_tag:	5✔
66	tagname = rpm.tagnames[tag]	×
67	if old_tag is None:	×
68	self.__add(self.FORMAT, (self.ADDED, tagname))	×
69	elif new_tag is None:	×
70	self.__add(self.FORMAT, (self.REMOVED, tagname))	×
71	else:
72	self.__add(self.FORMAT, ('S.5.....', tagname))	×
73
74	# compare Provides, Requires, ...
75	for tag in self.PRCO:	5✔
76	self.__comparePRCOs(old, new, tag)	5✔
77
78	# compare the files
79	old_files_dict = self.__fileIteratorToDict(rpm.files(old))	5✔
80	new_files_dict = self.__fileIteratorToDict(rpm.files(new))	5✔
81	files = sorted(set(chain(iter(old_files_dict), iter(new_files_dict))))	5✔
82
83	for f in files:	5✔
84	if self._excluded(f):	5✔
85	continue	5✔
86
87	diff = False	5✔
88
89	old_file = old_files_dict.get(f)	5✔
90	new_file = new_files_dict.get(f)	5✔
91
92	if not old_file:	5✔
93	self.__add(self.FORMAT, (self.ADDED, f))	5✔
94	elif not new_file:	5✔
95	self.__add(self.FORMAT, (self.REMOVED, f))	5✔
96	else:
97	fmt = ''	5✔
98	for entry in FILEIDX:	5✔
99	if entry[1] is not None and \	5✔
100	getattr(old_file, entry[1]) != getattr(new_file, entry[1]):
101	fmt += entry[0]	5✔
102	diff = True	5✔
103	else:
104	fmt += '.'	5✔
105	if diff:	5✔
106	self.__add(self.FORMAT, (fmt, f))	5✔
107
108	def _excluded(self, f):	5✔
109	f = pathlib.PurePath(f)	5✔
110	for glob in self.exclude:	5✔
111	if f.match(glob):	5✔
112	return True	5✔
113	if glob.startswith('/'):	5✔
114	for parent in f.parents:	5✔
115	if parent.match(glob):	5✔
116	return True	5✔
117	return False	5✔
118
119	# return a report of the differences
120	def textdiff(self):	5✔
121	return '\n'.join((fmt % data for fmt, data in self.result))	5✔
122
123	# do the two rpms differ
124	def differs(self):	5✔
125	return bool(self.result)	×
126
127	# add one differing item
128	def __add(self, fmt, data):	5✔
129	self.result.append((fmt, data))	5✔
130
131	# load a package from a file or from the installed ones
132	def __load_pkg(self, name):	5✔
133	# FIXME: redo to try file/installed and proceed based on that, or pick
134	# one of the selected first
135	tmpdir = tempfile.gettempdir()	5✔
136	with contextlib.suppress(TypeError):	5✔
137	if name.is_file():	5✔
138	return Pkg(name, tmpdir)	5✔
139	inst = get_installed_pkgs(str(name))	×
140	if not inst:	×
141	raise KeyError(f'No installed packages by name {name}')	×
142	if len(inst) > 1:	×
143	raise KeyError(f'More than one installed packages by name {name}')	×
144	return inst[0]	×
145
146	# output the right string according to RPMSENSE_* const
147	def sense2str(self, sense):	5✔
148	s = ''	5✔
149	for tag, char in ((rpm.RPMSENSE_LESS, '<'),	5✔
150	(rpm.RPMSENSE_GREATER, '>'),
151	(rpm.RPMSENSE_EQUAL, '=')):
152	if sense & tag:	5✔
153	s += char	5✔
154	return s	5✔
155
156	# output the right requires string according to RPMSENSE_* const
157	def req2str(self, req):	5✔
158	s = 'REQUIRES'	5✔
159	# we want to use 64 even with rpm versions that define RPMSENSE_PREREQ
160	# as 0 to get sane results when comparing packages built with an old
161	# (64) version and a new (0) one
162	if req & (rpm.RPMSENSE_PREREQ or 64):	5✔
163	s = 'PREREQ'	×
164
165	ss = []	5✔
166	if req & rpm.RPMSENSE_SCRIPT_PRE:	5✔
167	ss.append('pre')	×
168	if req & rpm.RPMSENSE_SCRIPT_POST:	5✔
169	ss.append('post')	×
170	if req & rpm.RPMSENSE_SCRIPT_PREUN:	5✔
171	ss.append('preun')	×
172	if req & rpm.RPMSENSE_SCRIPT_POSTUN:	5✔
173	ss.append('postun')	×
174	if req & getattr(rpm, 'RPMSENSE_PRETRANS', 1 << 7): # rpm >= 4.9.0	5✔
175	ss.append('pretrans')	×
176	if req & getattr(rpm, 'RPMSENSE_POSTTRANS', 1 << 5): # rpm >= 4.9.0	5✔
177	ss.append('posttrans')	×
178	if ss:	5✔
179	s += '(%s)' % ','.join(ss)	×
180
181	return s	5✔
182
183	# compare Provides, Requires, Conflicts, Obsoletes
184	def __comparePRCOs(self, old, new, name):	5✔
185	try:	5✔
186	oldflags = old[name[:-1] + 'FLAGS']	5✔
187	except ValueError:	×
188	# assume tag not supported, e.g. Recommends with older rpm
189	return	×
190	newflags = new[name[:-1] + 'FLAGS']	5✔
191	# fix buggy rpm binding not returning list for single entries
192	if not isinstance(oldflags, list):	5✔
193	oldflags = [oldflags]	×
194	if not isinstance(newflags, list):	5✔
195	newflags = [newflags]	×
196
197	o = zip(old[name], oldflags, old[name[:-1] + 'VERSION'])	5✔
198	if not isinstance(o, list):	5✔
199	o = list(o)	5✔
200	n = zip(new[name], newflags, new[name[:-1] + 'VERSION'])	5✔
201	if not isinstance(n, list):	5✔
202	n = list(n)	5✔
203
204	# filter self provides, TODO: self %name(%_isa) as well
205	if name == 'PROVIDES':	5✔
206	oldE = old['epoch'] is not None and str(old['epoch']) + ':' or ''	5✔
207	oldV = '{}{}'.format(oldE, old.format('%{VERSION}-%{RELEASE}'))	5✔
208	oldNV = (old['name'], rpm.RPMSENSE_EQUAL, oldV.encode())	5✔
209	newE = new['epoch'] is not None and str(new['epoch']) + ':' or ''	5✔
210	newV = '{}{}'.format(newE, new.format('%{VERSION}-%{RELEASE}'))	5✔
211	newNV = (new['name'], rpm.RPMSENSE_EQUAL, newV.encode())	5✔
212	o = [entry for entry in o if entry != oldNV]	5✔
213	n = [entry for entry in n if entry != newNV]	5✔
214
215	for oldentry in o:	5✔
216	if oldentry not in n:	5✔
217	namestr = name	5✔
218	if namestr == 'REQUIRES':	5✔
219	namestr = self.req2str(oldentry[1])	5✔
220	self.__add(self.DEPFORMAT,	5✔
221	(self.REMOVED, namestr, byte_to_string(oldentry[0]),
222	self.sense2str(oldentry[1]), byte_to_string(oldentry[2])))
223	for newentry in n:	5✔
224	if newentry not in o:	5✔
225	namestr = name	5✔
226	if namestr == 'REQUIRES':	5✔
227	namestr = self.req2str(newentry[1])	5✔
228	self.__add(self.DEPFORMAT,	5✔
229	(self.ADDED, namestr, byte_to_string(newentry[0]),
230	self.sense2str(newentry[1]), byte_to_string(newentry[2])))
231
232	def __fileIteratorToDict(self, fi):	5✔
233	result = {}	5✔
234	for filedata in fi:	5✔
235	result[filedata.name] = filedata	5✔
236	return result	5✔

rpm-software-management / rpmlint / 21511583274

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous