7263519030

Committed 18 Dec 2023 07:10PM UTC coverage: 43.472% (+0.1%) from 43.325%

Build # 7263519030

Build Type

push

github

Committed by

bcl

Commit Message

ltmpl: Check for errors after running the transaction

Thanks to the discussion in
https://github.com/rpm-software-management/dnf5/issues/1074 lorax will
now correctly log installation size errors like:

The transaction process has ended abruptly:
installing package llvm-libs-17.0.6-2.fc40.x86_64 needs 107MB more space on the / filesystem
installing package libXv-1.0.12-1.fc40.x86_64 needs 107MB more space on the / filesystem
installing package libXcomposite-0.4.6-1.fc40.x86_64 needs 107MB more space on the / filesystem

Run Details

595 of 1582 branches covered (0.0%)

Branch coverage included in aggregate %.

0 of 4 new or added lines in 1 file covered. (0.0%)

448 existing lines in 6 files now uncovered.

1616 of 3504 relevant lines covered (46.12%)

0.46 hits per line

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

33.8

/src/pylorax/dnfhelper.py

#
# dnfhelper.py
#
# Copyright (C) 2010-2015 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
# Red Hat Author(s):  Martin Gracik <mgracik@redhat.com>
#                     Brian C. Lane <bcl@redhat.com>
#

import logging
logger = logging.getLogger("pylorax.dnfhelper")
import time
import pylorax.output as output

import libdnf5 as dnf5
SUCCESSFUL = dnf5.repo.DownloadCallbacks.TransferStatus_SUCCESSFUL

__all__ = ['LoraxDownloadCallback', 'LoraxRpmCallback']

def _paced(fn):
    """Execute `fn` no more often then every 2 seconds."""
    def paced_fn(self, *args):
        now = time.time()
        if now - self.last_time < 2:
            return
        self.last_time = now
        return fn(self, *args)
    return paced_fn


class LoraxDownloadCallback(dnf5.repo.DownloadCallbacks):
    def __init__(self, total_files):
        super(LoraxDownloadCallback, self).__init__()
        self.last_time = time.time()
        self.total_files = total_files
        self.pkgno = 0

        self.output = output.LoraxOutput()
        self.nevra = "unknown"

    def add_new_download(self, user_data, description, total_to_download):
        self.nevra = description or "unknown"

        # Returning anything here makes it crash
        return None

    @_paced
    def _update(self):
        msg = "Downloading %(pkgno)s / %(total_files)s RPMs\n"
        vals = {
            'pkgno'       : self.pkgno,
            'total_files' : self.total_files,
        }
        self.output.write(msg % vals)

    def end(self, user_cb_data, status, msg):
        if status == SUCCESSFUL:
            self.pkgno += 1
            self._update()
        else:
            logger.critical("Failed to download '%s': %d - %s", self.nevra, status, msg)
        return 0

    def progress(self, user_cb_data, total_to_download, downloaded):
        self._update()
        return 0

    def mirror_failure(self, user_cb_data, msg, url, metadata):
        message = f"{url} - {msg}"
        logger.critical("Mirror failure on '%s': %s (%s)", self.nevra, message, metadata)
        return 0


class LoraxRpmCallback(dnf5.rpm.TransactionCallbacks):
    def install_start(self, item, total):
        action = dnf5.base.transaction.transaction_item_action_to_string(item.get_action())
        package = item.get_package().get_nevra()
        logger.info("%s %s", action, package)

    # pylint: disable=redefined-builtin
    def script_start(self, item, nevra, type):
        if not item or not type:
            return

        package = item.get_package().get_nevra()
        script_type = self.script_type_to_string(type)
        logger.info("Running %s for %s", script_type, package)

    ## NOTE: These likely will not work right, SWIG seems to crash when raising errors
    ##       from callbacks.
    def unpack_error(self, item):
        package = item.get_package().get_nevra()
        raise RuntimeError(f"unpack_error on {package}")

    def cpio_error(self, item):
        package = item.get_package().get_nevra()
        raise RuntimeError(f"cpio_error on {package}")

    # pylint: disable=redefined-builtin
    def script_error(self, item, nevra, type, return_code):
        package = item.get_package().get_nevra()
        script_type = self.script_type_to_string(type)
        raise RuntimeError(f"script_error on {package}: {script_type} rc={return_code}")

1	#
2	# dnfhelper.py
3	#
4	# Copyright (C) 2010-2015 Red Hat, Inc.
5	#
6	# This program is free software; you can redistribute it and/or modify
7	# it under the terms of the GNU General Public License as published by
8	# the Free Software Foundation; either version 2 of the License, or
9	# (at your option) any later version.
10	#
11	# This program is distributed in the hope that it will be useful,
12	# but WITHOUT ANY WARRANTY; without even the implied warranty of
13	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14	# GNU General Public License for more details.
15	#
16	# You should have received a copy of the GNU General Public License
17	# along with this program. If not, see <http://www.gnu.org/licenses/>.
18	#
19	# Red Hat Author(s): Martin Gracik <mgracik@redhat.com>
20	# Brian C. Lane <bcl@redhat.com>
21	#
22
23	import logging	1✔
24	logger = logging.getLogger("pylorax.dnfhelper")	1✔
25	import time	1✔
26	import pylorax.output as output	1✔
27
28	import libdnf5 as dnf5	1✔
29	SUCCESSFUL = dnf5.repo.DownloadCallbacks.TransferStatus_SUCCESSFUL	1✔
30
31	__all__ = ['LoraxDownloadCallback', 'LoraxRpmCallback']	1✔
32
33	def _paced(fn):	1✔
34	"""Execute `fn` no more often then every 2 seconds."""
35	def paced_fn(self, *args):	1✔
36	now = time.time()	×
37	if now - self.last_time < 2:	×
38	return	×
39	self.last_time = now	×
40	return fn(self, *args)	×
41	return paced_fn	1✔
42
43
44	class LoraxDownloadCallback(dnf5.repo.DownloadCallbacks):	1✔
45	def __init__(self, total_files):	1✔
46	super(LoraxDownloadCallback, self).__init__()	×
47	self.last_time = time.time()	×
48	self.total_files = total_files	×
49	self.pkgno = 0	×
50
51	self.output = output.LoraxOutput()	×
52	self.nevra = "unknown"	×
53
54	def add_new_download(self, user_data, description, total_to_download):	1✔
UNCOV 55	self.nevra = description or "unknown"	×
56
57	# Returning anything here makes it crash
58	return None	×
59
60	@_paced	1✔
61	def _update(self):	1✔
UNCOV 62	msg = "Downloading %(pkgno)s / %(total_files)s RPMs\n"	×
UNCOV 63	vals = {	×
64	'pkgno' : self.pkgno,
65	'total_files' : self.total_files,
66	}
UNCOV 67	self.output.write(msg % vals)	×
68
69	def end(self, user_cb_data, status, msg):	1✔
UNCOV 70	if status == SUCCESSFUL:	×
71	self.pkgno += 1	×
72	self._update()	×
73	else:
74	logger.critical("Failed to download '%s': %d - %s", self.nevra, status, msg)	×
75	return 0	×
76
77	def progress(self, user_cb_data, total_to_download, downloaded):	1✔
UNCOV 78	self._update()	×
UNCOV 79	return 0	×
80
81	def mirror_failure(self, user_cb_data, msg, url, metadata):	1✔
82	message = f"{url} - {msg}"	×
UNCOV 83	logger.critical("Mirror failure on '%s': %s (%s)", self.nevra, message, metadata)	×
UNCOV 84	return 0	×
85
86
87	class LoraxRpmCallback(dnf5.rpm.TransactionCallbacks):	1✔
88	def install_start(self, item, total):	1✔
UNCOV 89	action = dnf5.base.transaction.transaction_item_action_to_string(item.get_action())	×
UNCOV 90	package = item.get_package().get_nevra()	×
UNCOV 91	logger.info("%s %s", action, package)	×
92
93	# pylint: disable=redefined-builtin
94	def script_start(self, item, nevra, type):	1✔
UNCOV 95	if not item or not type:	×
UNCOV 96	return	×
97
UNCOV 98	package = item.get_package().get_nevra()	×
99	script_type = self.script_type_to_string(type)	×
100	logger.info("Running %s for %s", script_type, package)	×
101
102	## NOTE: These likely will not work right, SWIG seems to crash when raising errors
103	## from callbacks.
104	def unpack_error(self, item):	1✔
105	package = item.get_package().get_nevra()	×
106	raise RuntimeError(f"unpack_error on {package}")	×
107
108	def cpio_error(self, item):	1✔
UNCOV 109	package = item.get_package().get_nevra()	×
110	raise RuntimeError(f"cpio_error on {package}")	×
111
112	# pylint: disable=redefined-builtin
113	def script_error(self, item, nevra, type, return_code):	1✔
UNCOV 114	package = item.get_package().get_nevra()	×
UNCOV 115	script_type = self.script_type_to_string(type)	×
UNCOV 116	raise RuntimeError(f"script_error on {package}: {script_type} rc={return_code}")	×

weldr / lorax / 7263519030

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