Inhibit the upgrade when multiple rescue boot entries used on s390x.

When performing an IPU on s390x the upgrade will not finish
successfully, due to the failure of conversion of the Zipl configuration
to the BLS, when multiple Zipl rescue entries are present in the
configuration.

To address this problem the BootEntriesScanner was introduced along with
the `BootEntry` model, capable of scanning the bootloader configuration
and providing the information to other actors.

The added `ZiplCheckMultipleRescueEntries` actor consumes this
information and inhibits the upgrade if there are multiple rescue
entries found, and therefore, the upgrade would not be successful.
The actor runs only on the s390x architecture as multiple rescue entries
pose no problem when running the GRUB2 bootloader.

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

@@ -0,0 +1,18 @@
+from leapp.actors import Actor
+from leapp.libraries.actor.sourcebootloaderscanner import scan_source_boot_loader_configuration
+from leapp.models import SourceBootLoaderConfiguration
+from leapp.tags import FactsPhaseTag, IPUWorkflowTag
+
+
+class SourceBootLoaderScanner(Actor):
+    """
+    Scans the boot loader configuration on the source system.
+    """
+
+    name = 'source_boot_loader_scanner'
+    consumes = ()
+    produces = (SourceBootLoaderConfiguration,)
+    tags = (FactsPhaseTag, IPUWorkflowTag)
+
+    def process(self):
+        scan_source_boot_loader_configuration()

repos/system_upgrade/el7toel8/actors/sourcebootloaderscanner/libraries/sourcebootloaderscanner.py

@@ -0,0 +1,57 @@
+from leapp.models import BootEntry, SourceBootLoaderConfiguration
+from leapp.libraries.stdlib import api, CalledProcessError, run
+from leapp.exceptions import StopActorExecutionError
+
+
+CMD_GRUBBY_INFO_ALL = ['grubby', '--info', 'ALL']
+
+
+def scan_boot_entries():
+    """
+    Scans the available boot entries.
+
+    :rtype: list
+    :returns: A list of available boot entries found in the boot loader configuration.
+    """
+    try:
+        grubby_output = run(CMD_GRUBBY_INFO_ALL, split=True)
+    except CalledProcessError as err:
+        # We have failed to call `grubby` - something is probably wrong here.
+        raise StopActorExecutionError(
+            message='Failed to call `grubby` to list available boot entries.',
+            details={
+                'details': str(err),
+                'stderr': err.stderr
+            }
+        )
+
+    boot_entries = []
+    # Identify the available boot entries by searching for their titles in the grubby output
+    for output_line in grubby_output['stdout']:
+        # For now it is sufficient to look only for the titles as that is the only
+        # information we use. If need be, we would have to parse the structure
+        # of the grubby output into sections according to the `index` lines
+        if output_line.startswith('title='):
+            boot_entry = output_line[6:]  # Remove the `title=` prefix
+
+            # On s390x grubby produces quotes only when needed (whitespace in
+            # the configuration values), on x86 the values are quoted either way
+            boot_entry = boot_entry.strip('\'"')
+
+            boot_entries.append(BootEntry(title=boot_entry))
+
+    return boot_entries
+
+
+def scan_source_boot_loader_configuration():
+    """
+    Scans the boot loader configuration.
+
+    Produces :class:`SourceBootLoaderConfiguration for other actors to act upon.
+    """
+
+    boot_loader_configuration = SourceBootLoaderConfiguration(
+        entries=scan_boot_entries()
+    )
+
+    api.produce(boot_loader_configuration)

repos/system_upgrade/el7toel8/actors/sourcebootloaderscanner/tests/test_bootentryscanner.py

@@ -0,0 +1,52 @@
+import pytest
+
+from leapp.libraries import stdlib
+from leapp.libraries.actor import sourcebootloaderscanner
+from leapp.libraries.common.testutils import produce_mocked
+
+
+GRUBBY_INFO_ALL_STDOUT = '''index=0
+kernel="/boot/vmlinuz-4.18.0-305.7.1.el8_4.x86_64"
+args="ro uned_params"
+root="/someroot"
+initrd="/boot/initramfs-4.18.0-305.7.1.el8_4.x86_64.img"
+title="Linux"
+id="some_id"
+index=1
+kernel="/boot/vmlinuz-4.18.0-305.3.1.el8_4.x86_64"
+args="ro"
+root="/someroot"
+initrd="/boot/initramfs-4.18.0-305.3.1.el8_4.x86_64.img"
+title="Linux old-kernel"
+id="some_id2"'''
+
+
+def test_scan_boot_entries(monkeypatch):
+    """Tests whether the library correctly identifies boot entries in the grubby output."""
+    def run_mocked(cmd, **kwargs):
+        if cmd == ['grubby', '--info', 'ALL']:
+            return {
+                'stdout': GRUBBY_INFO_ALL_STDOUT.split('\n')
+            }
+        raise ValueError('Tried to run unexpected command.')
+
+    actor_produces = produce_mocked()
+
+    # The library imports `run` all the way (from ... import run), therefore,
+    # we must monkeypatch the reference directly in the actor's library namespace
+    monkeypatch.setattr(sourcebootloaderscanner, 'run', run_mocked)
+    monkeypatch.setattr(stdlib.api, 'produce', actor_produces)
+
+    sourcebootloaderscanner.scan_source_boot_loader_configuration()
+
+    fail_description = 'Only one SourceBootLoaderConfiguration message should be produced.'
+    assert len(actor_produces.model_instances) == 1, fail_description
+
+    bootloader_config = actor_produces.model_instances[0]
+
+    fail_description = 'Found different number of boot entries than present in provided mocks.'
+    assert len(bootloader_config.entries) == 2, fail_description
+
+    expected_boot_entry_titles = ['Linux', 'Linux old-kernel']
+    for actual_boot_entry in bootloader_config.entries:
+        assert actual_boot_entry.title in expected_boot_entry_titles

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

@@ -0,0 +1,21 @@
+from leapp.actors import Actor
+from leapp.libraries.actor.ziplcheckmultiplerescueentries import inhibit_if_multiple_zipl_rescue_entries_present
+from leapp.models import SourceBootLoaderConfiguration
+from leapp.tags import ChecksPhaseTag, IPUWorkflowTag
+from leapp.reporting import Report
+
+
+class ZiplCheckMultipleRescueEntries(Actor):
+    """
+    Inhibits the upgrade process if there are more than one rescue entries in
+    the zipl configuration.
+    """
+
+    name = 'zipl_check_multiple_rescue_entries'
+    consumes = (SourceBootLoaderConfiguration,)
+    produces = (Report,)
+    tags = (ChecksPhaseTag, IPUWorkflowTag)
+
+    def process(self):
+        boot_loader_configuration = next(self.consume(SourceBootLoaderConfiguration))
+        inhibit_if_multiple_zipl_rescue_entries_present(boot_loader_configuration)

repos/system_upgrade/el7toel8/actors/ziplcheckmultiplerescueentries/libraries/ziplcheckmultiplerescueentries.py

@@ -0,0 +1,45 @@
+from leapp import reporting
+from leapp.libraries.common.config import architecture
+
+FMT_LIST_SEPARATOR = '\n    - '
+
+
+def inhibit_if_multiple_zipl_rescue_entries_present(bootloader_config):
+    """
+    Inhibits the upgrade if we are running on s390x and the bootloader configuration
+    contains multiple rescue boot entries.
+
+    A boot entry is recognized as a rescue entry when its title contains the `rescue` substring.
+
+    :param SourceBootloaderConfiguration bootloader_config: The configuration of the source boot loader.
+    """
+
+    if not architecture.matches_architecture(architecture.ARCH_S390X):
+        # Zipl is used only on s390x
+        return
+
+    # Keep the whole information about boot entries not just their count as
+    # we want to provide user with the details
+    rescue_entries = []
+    for boot_entry in bootloader_config.entries:
+        if 'rescue' in boot_entry.title.lower():
+            rescue_entries.append(boot_entry)
+
+    if len(rescue_entries) > 1:
+        # Prepare the list of available rescue entries for user
+        rescue_entries_text = ''
+        for rescue_entry in rescue_entries:
+            rescue_entries_text += '{0}{1}'.format(FMT_LIST_SEPARATOR, rescue_entry.title)
+        zipl_config_path = '/etc/zipl.conf'
+
+        reporting.create_report([
+            reporting.Title('Multiple rescue boot entries present in the bootloader configuration.'),
+            reporting.Summary(
+                'The zipl configuration file {0} contains multiple rescue boot entries:{1}'
+                .format(zipl_config_path, rescue_entries_text)
+            ),
+            reporting.Severity(reporting.Severity.HIGH),
+            reporting.Tags([reporting.Tags.BOOT]),
+            reporting.Remediation(hint='Remove rescue boot entries from the configuration and leave just one.'),
+            reporting.Flags([reporting.Flags.INHIBITOR])
+        ])

repos/system_upgrade/el7toel8/actors/ziplcheckmultiplerescueentries/tests/test_ziplcheckmultiplerescueentries.py

@@ -0,0 +1,75 @@
+import pytest
+
+from leapp import reporting
+from leapp.libraries.actor import ziplcheckmultiplerescueentries
+from leapp.libraries.actor.ziplcheckmultiplerescueentries import inhibit_if_multiple_zipl_rescue_entries_present
+from leapp.libraries.common.testutils import create_report_mocked, CurrentActorMocked
+from leapp.libraries.common.config import architecture
+from leapp.libraries.stdlib import api
+from leapp.models import BootEntry, SourceBootLoaderConfiguration
+from leapp.snactor.fixture import current_actor_context
+
+
+def test_inhibition_multiple_rescue_entries_present(monkeypatch):
+    """Tests whether the upgrade process is inhibited when multiple rescue boot entries are present."""
+    mocked_report = create_report_mocked()
+    monkeypatch.setattr(architecture, 'matches_architecture', lambda dummy: True)
+    monkeypatch.setattr(reporting, 'create_report', mocked_report)
+
+    boot_entries = [
+        BootEntry(title='entry_1'),
+        BootEntry(title='entry_1_Rescue'),
+        BootEntry(title='entry_2'),
+        BootEntry(title='entry_2_rescue-ver2.3'),  # Typically is the `rescue` substring surrounded
+    ]
+
+    inhibit_if_multiple_zipl_rescue_entries_present(SourceBootLoaderConfiguration(entries=boot_entries))
+
+    assert mocked_report.called, 'Report should be created when multiple rescue entries are present.'
+
+    fail_description = 'The correct rescue entries are not present in the report summary.'
+    report_summary = mocked_report.report_fields['summary']
+    for expected_rescue_entry in ['entry_1_Rescue', 'entry_2_rescue-ver2.3']:
+        assert expected_rescue_entry in report_summary, fail_description
+
+    fail_description = 'Upgrade should be inhibited on multiple rescue entries.'
+    assert 'inhibitor' in mocked_report.report_fields['flags'], fail_description
+
+
+def test_inhibition_multiple_rescue_entries_not_present(monkeypatch):
+    """Tests whether the upgrade process is not inhibited when multiple rescue boot entries are not present."""
+    monkeypatch.setattr(architecture, 'matches_architecture', lambda dummy: True)
+    monkeypatch.setattr(reporting, 'create_report', create_report_mocked())
+
+    boot_entries = [
+        BootEntry(title='entry_1'),
+        BootEntry(title='entry_2'),
+        BootEntry(title='entry_2_rescue-ver2.3'),
+    ]
+
+    inhibit_if_multiple_zipl_rescue_entries_present(SourceBootLoaderConfiguration(entries=boot_entries))
+
+    assert not reporting.create_report.called, 'Report was created, even if multiple rescue entries were not present.'
+
+
+@pytest.mark.parametrize(
+    ('arch',),
+    [(arch,) for arch in architecture.ARCH_SUPPORTED]
+)
+def test_checks_performed_only_on_s390x_arch(arch, monkeypatch):
+    """Tests whether the actor doesn't perform different architectures than s390x."""
+
+    should_perform = False
+    if arch == architecture.ARCH_S390X:  # Rescue entries should be checked only on s390x.
+        should_perform = True
+
+    monkeypatch.setattr(api, 'current_actor', CurrentActorMocked(arch=arch))
+    monkeypatch.setattr(reporting, 'create_report', create_report_mocked())
+
+    boot_entries = [BootEntry(title='rescue0'), BootEntry(title='rescue1')]
+    inhibit_if_multiple_zipl_rescue_entries_present(SourceBootLoaderConfiguration(entries=boot_entries))
+
+    fail_description = 'Rescue entries should not be checked on non s390x architecture.'
+    if should_perform:
+        fail_description = 'No report was created when running on s390x and multiple rescue entries were used.'
+    assert bool(reporting.create_report.called) == should_perform, fail_description

repos/system_upgrade/el7toel8/models/bootloaderconfiguration.py

@@ -0,0 +1,22 @@
+from leapp.models import Model, fields
+from leapp.topics import SystemInfoTopic
+
+
+class BootEntry(Model):
+    """
+    One entry in the boot loader configuration.
+
+    Not meant to be produced directly, only as a part of :class:`SourceBootLoaderConfiguration`.
+    """
+    topic = SystemInfoTopic
+
+    title = fields.String()
+    """Title of the boot entry."""
+
+
+class SourceBootLoaderConfiguration(Model):
+    """Describes the bootloader configuration found on the source system."""
+    topic = SystemInfoTopic
+
+    entries = fields.List(fields.Model(BootEntry))
+    """Boot entries available in the bootloader configuration."""