InstalledTargetKernelVersion: respect kernel-rt

Previously just the kernel rpm has been detected. However, in case
of real time systems, we need to detect the kernel-rt rpm. In case
of real time kernels, the kernel rpm doesn't have to be installed
at all.

Currently it's not clear whether we should handle stuff realted to
both kernels (kernel and kernel-rt) when both present. Right now,
detect whether the original system was booted with preemptive kernel
or not and based on that prefer the expected kernel.

In case the target kernel-rt rpm is not present and it is expected,
fallback to the kernel rpm to make possible finish of the upgrade,
but log warning msg.

TODO: raise report in such a case

repos/system_upgrade/el7toel8/actors/scaninstalledtargetkernelversion/actor.py

@@ -8,7 +8,10 @@ class ScanInstalledTargetKernelVersion(Actor):
     """
     Scan for the version of the newly installed kernel
 
-    This actor will query rpm for all kernel packages and reports the first matching el8 kernel RPM version.
+    This actor will query rpm for all kernel packages and reports the first
+    matching el8 kernel RPM. In case the RHEL Real Time has been detected on
+    the original system, the kernel-rt rpm is searched. If the rpm is missing,
+    fallback for standard kernel RPM.
     """
 
     name = 'scan_installed_target_kernel_version'

repos/system_upgrade/el7toel8/actors/scaninstalledtargetkernelversion/libraries/scankernel.py

@@ -1,12 +1,51 @@
-from leapp.libraries import stdlib
-from leapp.libraries.stdlib import api
+from leapp.libraries.common.config.version import is_rhel_realtime
+from leapp.libraries.stdlib import api, CalledProcessError, run
 from leapp.models import InstalledTargetKernelVersion
 
 
-def process():
-    kernels = stdlib.run(["rpm", "-q", "kernel"], split=True)["stdout"]
+def _get_kernel_version(kernel_name):
+    try:
+        kernels = run(['rpm', '-q', kernel_name], split=True)['stdout']
+    except CalledProcessError:
+        return ''
+
     for kernel in kernels:
-        version = kernel.split("-", 1)[1]
-        if "el8" in version:
+        # name-version-release - we want the last two fields only
+        version = '-'.join(kernel.split('-')[-2:])
+        if 'el8' in version:
+            return version
+    return ''
+
+
+def process():
+    # pylint: disable=no-else-return  - false positive
+    # TODO: should we take care about stuff of kernel-rt and kernel in the same
+    # time when both are present? or just one? currently, handle only one
+    # of these during the upgrade. kernel-rt has higher prio when original sys
+    # was realtime
+
+    if is_rhel_realtime():
+        version = _get_kernel_version('kernel-rt')
+        if version:
             api.produce(InstalledTargetKernelVersion(version=version))
-            break
+            return
+        else:
+            api.current_logger().warning(
+                'The kernel-rt rpm from RHEL 8 has not been detected. Switching to non-preemptive kernel.')
+            # TODO: create report with instructions to install kernel-rt manually
+            # - attach link to article if any
+            # - this possibly happens just in case the repository with kernel-rt
+            # # is not enabled during the upgrade.
+
+    # standard (non-preemptive) kernel
+    version = _get_kernel_version('kernel')
+    if version:
+        api.produce(InstalledTargetKernelVersion(version=version))
+    else:
+        # This is very unexpected situation. At least one kernel has to be
+        # installed always. Some actors consuming the InstalledTargetKernelVersion
+        # will crash without the created message. I am keeping kind of original
+        # behaviour in this case, but at least the let me log the error msg
+        #
+        api.current_logger().error('Cannot detect any kernel RPM')
+        # StopActorExecutionError('Cannot detect any RHEL 8 kernel RPM.')

repos/system_upgrade/el7toel8/actors/scaninstalledtargetkernelversion/tests/test_scaninstalledkernel_scaninstalledtargetkernelversion.py

@@ -1,31 +1,76 @@
+import pytest
+
 from leapp.libraries import stdlib
 from leapp.libraries.actor import scankernel
+from leapp.libraries.common.testutils import CurrentActorMocked, logger_mocked
 from leapp.libraries.stdlib import api
 
 TARGET_KERNEL_VERSION = '1.2.3-4.el8.x86_64'
+TARGET_RT_KERNEL_VERSION = '1.2.3-4.rt56.7.el8.x86_64'
 TARGET_KERNEL = 'kernel-{}'.format(TARGET_KERNEL_VERSION)
+TARGET_RT_KERNEL = 'kernel-{}'.format(TARGET_RT_KERNEL_VERSION)
 OLD_KERNEL = 'kernel-0.1.2-3.el7.x86_64'
+OLD_RT_KERNEL = 'kernel-rt-0.1.2-3.rt4.5.el7.x86_64'
+
+
+class MockedRun(object):
 
+    def __init__(self, stdouts):
+        # stdouts should be dict of list of strings: { str: [str1,str2,...]}
+        self._stdouts = stdouts
 
-def mocked_run_with_target_kernel(*args, **kwargs):
-    return {'stdout': [TARGET_KERNEL, OLD_KERNEL]}
+    def __call__(self, *args, **kwargs):
+        for key in ('kernel', 'kernel-rt'):
+            if key in args[0]:
+                return {'stdout': self._stdouts.get(key, [])}
+        return {'stdout': []}
 
 
-def mocked_run_without_target_kernel(*args, **kwargs):
-    return {'stdout': [OLD_KERNEL]}
+@pytest.mark.parametrize('is_rt,exp_version,stdouts', [
+    (False, TARGET_KERNEL_VERSION, {'kernel': [OLD_KERNEL, TARGET_KERNEL]}),
+    (False, TARGET_KERNEL_VERSION, {'kernel': [TARGET_KERNEL, OLD_KERNEL]}),
+    (False, TARGET_KERNEL_VERSION, {
+        'kernel': [TARGET_KERNEL, OLD_KERNEL],
+        'kernel-rt': [TARGET_RT_KERNEL, OLD_RT_KERNEL],
+    }),
+    (True, TARGET_RT_KERNEL_VERSION, {'kernel-rt': [OLD_RT_KERNEL, TARGET_RT_KERNEL]}),
+    (True, TARGET_RT_KERNEL_VERSION, {'kernel-rt': [TARGET_RT_KERNEL, OLD_RT_KERNEL]}),
+    (True, TARGET_RT_KERNEL_VERSION, {
+        'kernel': [TARGET_KERNEL, OLD_KERNEL],
+        'kernel-rt': [TARGET_RT_KERNEL, OLD_RT_KERNEL],
+    }),
+])
+def test_scaninstalledkernel(monkeypatch, is_rt, exp_version, stdouts):
+    result = []
+    old_kver = '0.1.2-3.rt4.5.el7.x86_64' if is_rt else 'kernel-0.1.2-3.el7.x86_64'
+    monkeypatch.setattr(api, 'current_actor', CurrentActorMocked(kernel=old_kver))
+    monkeypatch.setattr(api, 'produce', result.append)
+    monkeypatch.setattr(scankernel, 'run', MockedRun(stdouts))
+    scankernel.process()
+    assert len(result) == 1 and result[0].version == exp_version
 
 
-def test_scaninstalledkernel(monkeypatch):
+def test_scaninstalledkernel_missing_rt(monkeypatch):
     result = []
-    monkeypatch.setattr(stdlib, 'run', mocked_run_with_target_kernel)
+    old_kver = '0.1.2-3.rt4.5.el7.x86_64'
+    stdouts = {'kernel': [TARGET_KERNEL], 'kernel-rt': [OLD_RT_KERNEL]}
+    monkeypatch.setattr(api, 'current_actor', CurrentActorMocked(kernel=old_kver))
+    monkeypatch.setattr(api, 'current_logger', logger_mocked())
     monkeypatch.setattr(api, 'produce', result.append)
+    monkeypatch.setattr(scankernel, 'run', MockedRun(stdouts))
     scankernel.process()
-    assert result and result[0].version == TARGET_KERNEL_VERSION
+    assert api.current_logger.warnmsg
+    assert len(result) == 1 and result[0].version == TARGET_KERNEL_VERSION
 
 
 def test_scaninstalledkernel_missing(monkeypatch):
     result = []
-    monkeypatch.setattr(stdlib, 'run', mocked_run_without_target_kernel)
+    old_kver = '0.1.2-3.rt4.5.el7.x86_64'
+    monkeypatch.setattr(api, 'current_actor', CurrentActorMocked(kernel=old_kver))
+    monkeypatch.setattr(api, 'current_logger', logger_mocked())
     monkeypatch.setattr(api, 'produce', result.append)
+    monkeypatch.setattr(scankernel, 'run', MockedRun({}))
     scankernel.process()
+    assert api.current_logger.warnmsg
+    assert api.current_logger.errmsg
     assert not result