# Copyright (C) 2023 Red Hat, Inc. # SPDX-License-Identifier: LGPL-2.1-or-later import json import os import subprocess import sys import tempfile # import Cockpit's machinery for test VMs and its browser test API TEST_DIR = os.path.dirname(__file__) ROOT_DIR = os.path.dirname(TEST_DIR) BOTS_DIR = f'{ROOT_DIR}/bots' sys.path.append(os.path.join(TEST_DIR, "common")) sys.path.insert(0, os.path.join(TEST_DIR, "helpers")) sys.path.append(os.path.join(os.path.dirname(TEST_DIR), "bots/machine")) from installer import Installer from language import Language from machine_install import INSTALLER_VM_MEMORY_MB, VirtInstallMachine # noqa: F401 (INSTALLER_VM_MEMORY_MB re-exported) from payload_dnf import PayloadDNFDBus from progress import Progress from storage import Storage from testlib import Browser, MachineCase, wait # pylint: disable=import-error from timezone import DateAndTime from users import Users from utils import add_public_key pixel_tests_ignore = [ "#anaconda-screen-review-target-system-timezone", "#anaconda-screen-review-target-operating-system", ".pf-v6-c-page__main-group", # ignore the header mentioning the OS details ] class TLSBrowser(Browser): def open(self, href, cookie=None, tls=True): super().open(href, cookie=cookie, tls=tls) class VirtInstallMachineCase(MachineCase): # The boot modes in which the test should run boot_modes = ["efi"] is_efi = os.environ.get("TEST_FIRMWARE", "efi") == "efi" report_to_wiki = os.path.exists(os.path.join(TEST_DIR, "report.json")) MachineCase.machine_class = VirtInstallMachine report_file = os.path.join(TEST_DIR, "report.json") run_on_vm_setups: list[str] = [""] vm_setup = "" def new_browser(self, *args, **kwargs): # Swap the class to TLSBrowser so open() defaults to tls=True. # Ideally MachineCase would favor composition over inheritance # and accept a browser_class attribute (like machine_class), # letting subclasses inject their own Browser without hacking # the factory method. # TODO: propose a PR in cockpit so we don't need this hack browser = super().new_browser(*args, **kwargs) browser.__class__ = TLSBrowser return browser def partition_disk(self): """ Override this method to partition the disk """ pass @property def temp_dir(self): """Get temp directory for libvirt resources We need to set the directory based on the fact if the test is started in the toolbx """ # toolbox compatibility: /tmp is shared with the host, but may be too small for big overlays (tmpfs!) # $HOME is shared, but we don't want to put our junk there (NFS, backups) # /var/tmp is not shared with the host but the right place; just in case session libvirtd is already # running, use the shared path so that the daemon can actually see our overlay. # But this only makes sense if the host also has /run/host set up (toolbox ships a tmpfiles.d) if os.path.exists("/run/host/var/tmp") and os.path.exists("/run/host/run/host"): return "/run/host/var/tmp" return "/var/tmp" @classmethod def setUpClass(cls): cls.ext_logging = bool(int(os.environ.get('EXTENDED_LOGGING', '0'))) def setUp(self): method = getattr(self, self._testMethodName) boot_modes = getattr(method, "boot_modes", ["efi"]) self.run_on_vm_setups = getattr(method, "run_on_vm_setups", [""]) self.disk_images = getattr(method, "disk_images", [("", 15)]) if self.is_efi and "efi" not in boot_modes: self.skipTest("Skipping for EFI boot mode") elif not self.is_efi and "bios" not in boot_modes: self.skipTest("Skipping for BIOS boot mode") self.vm_setup = os.environ.get("TEST_VM_SETUP", "") if self.vm_setup not in self.run_on_vm_setups: self.skipTest(f"Skipping for VM setup {self.vm_setup}" f", requires VM setups: {self.run_on_vm_setups}") # FIXME: running this in destructive tests fails because the SSH session closes before this is run if self.is_nondestructive(): self.addCleanup(self.resetUsers) self.addCleanup(self.resetStorage) self.addCleanup(self.resetLanguage) self.addCleanup(self.resetMisc) self.addCleanup(self.resetTimezone) self.addCleanup(self.resetPayloadDNF) super().setUp() m = self.machine b = self.browser s = Storage(b, m) self.addAllDisks() s.udevadm_settle() # Wait for minimum /dev/vda to be detected before proceeding wait(lambda: "vda" in m.execute("ls /dev"), tries=5, delay=5) partition_disk_method_name = f"_{self._testMethodName}_partition_disk" if getattr(self, partition_disk_method_name, None): self.partition_disk = getattr(self, partition_disk_method_name) self.partition_disk() s.udevadm_settle() s.dbus_scan_devices() # Set the first disk as the installation target s.dbus_set_selected_disk("vda") self.resetLanguage() self.allow_journal_messages('.*cockpit.bridge-WARNING: Could not start ssh-agent.*') # Nested Cockpit (e.g. storage/network iframes) and language changes which reload the page can log this; harmless. self.allow_journal_messages("Error .* data: Connection reset by peer") self.installation_finished = False if not self.is_nondestructive(): # Assume destructive tests may reboot the machine and ignore errors related to that self.allow_browser_errors(".*client closed.*") self.allow_browser_errors(".*Server has closed the connection.*") def add_disk(self, size, backing_file=None, target="vda"): image = self._create_disk_image(size, backing_file=backing_file) subprocess.check_call([ "virt-xml", "-c", "qemu:///session", self.machine.label, "--update", "--add-device", "--disk", f"{image},format=qcow2,target={target}" ]) return image def rem_disk(self, disk): subprocess.check_call([ "virt-xml", "-c", "qemu:///session", self.machine.label, "--update", "--remove-device", "--disk", disk ]) os.remove(disk) def _create_disk_image(self, size, image_path=None, backing_file=None): if not image_path: _, image_path = tempfile.mkstemp(suffix='.qcow2', prefix=f"disk-anaconda-{self.machine.label}", dir=self.temp_dir) subprocess.check_call([ "qemu-img", "create", "-f", "qcow2", *(["-o", f"backing_file={backing_file},backing_fmt=qcow2"] if backing_file else []), image_path, f"{size}G" ]) return image_path def resetLanguage(self): m = self.machine b = self.browser lang = Language(b, m) if not lang.dbus_get_language_kickstarted(): lang.dbus_set_language("en_US.UTF-8") lang.dbus_set_locale("en_US.UTF-8") lang.dbus_set_compositor_layouts(["us"]) lang.dbus_reset_xlayouts() lang.dbus_reset_virtual_console_keymap() def resetUsers(self): m = self.machine b = self.browser users = Users(b, m) users.dbus_clear_users() users.dbus_set_root_locked(True) def resetTimezone(self): m = self.machine b = self.browser dt = DateAndTime(b, m) dt.dbus_set_ntp_enabled(True) def addAllDisks(self): # Add installation target disks for index, (disk, size) in enumerate(self.disk_images): target = f"vd{chr(97 + index)}" backing_file = os.path.join(BOTS_DIR, f"./images/{disk}") if disk else None # Download the image if it doesn't exist if backing_file and not os.path.exists(backing_file): subprocess.check_call([os.path.join(BOTS_DIR, "image-download"), disk]) self.add_disk(size, backing_file, target) # Select the disk as boot device subprocess.check_call([ "virt-xml", "-c", "qemu:///session", self.machine.label, "--edit", "--boot", "hd" ]) def removeAllDisks(self): # Remove all disks domblklist = subprocess.getoutput( f"virsh domblklist {self.machine.label}", ) for line in domblklist.splitlines(): name = line.split()[0] if "vd" in name: file = line.split()[1] self.rem_disk(file) def resetStorage(self): # Ensures that anaconda has the latest storage configuration data m = self.machine b = self.browser s = Storage(b, m) self.removeAllDisks() s.dbus_reset_scenario() # Create an AUTOMATIC partitioning because MANUAL partitioning tests might take the last created if len(s.dbus_get_created_partitioning()) != 0: s.dbus_create_partitioning("AUTOMATIC") s.dbus_reset_selected_disks() # CLEAR_PARTITIONS_DEFAULT = -1 s.dbus_set_initialization_mode(-1) s.dbus_scan_devices() def resetMisc(self): # Restart cockpit-ws/cockpit-bridge to avoid crashes in the next test m = self.machine m.execute("systemctl restart webui-cockpit-ws.service") def resetPayloadDNF(self): m = self.machine if getattr(m, "payload_type", None) != "dnf": return dnf_dbus = PayloadDNFDBus(m) dnf_dbus.dbus_reset_to_default_environment("server-product-environment") def downloadLogs(self): if not self.ext_logging: return self.logs_dir = os.path.join('./test_logs', f'{self.__class__.__name__}.{self._testMethodName}') if not os.path.isdir(self.logs_dir): os.makedirs(self.logs_dir) self.machine.download('/tmp/anaconda.log', 'anaconda.log', self.logs_dir) self.machine.download('/tmp/packaging.log', 'packaging.log', self.logs_dir) self.machine.download('/tmp/storage.log', 'storage.log', self.logs_dir) self.machine.download('/tmp/dbus.log', 'dbus.log', self.logs_dir) self.machine.download('/tmp/syslog', 'syslog', self.logs_dir) try: self.machine.download('/tmp/anaconda-tb-*', '.', self.logs_dir) except subprocess.CalledProcessError: pass def handleReboot(self): """ Method for rebooting into the installed system. Should be called when installation is finished. Performs necessary steps so the test can continue with checking installed system. """ add_public_key(self.machine) self.downloadLogs() self.installation_finished = True p = Progress(self.browser) p.reboot() # The installed machine does not need to skip the nologin check del os.environ["TEST_ALLOW_NOLOGIN"] self.addCleanup(lambda: os.environ.__setitem__("TEST_ALLOW_NOLOGIN", "1")) self.machine.wait_reboot() def selectBootMenuEntry(self, entry): grub_cfg = """ GRUB_DEFAULT=saved GRUB_TIMEOUT=0 GRUB_HIDDEN_TIMEOUT=0 GRUB_HIDDEN_TIMEOUT_QUIET=true """ self.write_file("/etc/default/grub", grub_cfg) self.machine.execute(f"grub2-set-default {entry}") def install(self, needs_confirmation, button_text="Install"): b = self.browser m = self.machine i = Installer(b, m) p = Progress(b) i.begin_installation(button_text=button_text, needs_confirmation=needs_confirmation) with b.wait_timeout(300): p.wait_done() # FIXME: https://bugzilla.redhat.com/show_bug.cgi?id=2325707 # This should be removed from the test if self.is_efi: # Add efibootmgr entry for the second OS distro_name = self.disk_images[0][0].split("-")[0] m.execute(f"efibootmgr -c -d /dev/vda -p 15 -L {distro_name} -l '/EFI/{distro_name}/shimx64.efi'") # Select the Fedora grub entry for first boot m.execute("efibootmgr -n 0002") self.handleReboot() def appendResultsToReport(self): with open(self.report_file, "r+") as f: test_name = f"{self.__class__.__name__}.{self._testMethodName}" firmware = "UEFI" if self.is_efi else "BIOS" arch = "x86_64" error = super().getError() status = "fail" if error else "pass" # Add the new entry in the "tests" array in the JSON report file data = json.load(f) new_entry = { "arch": arch, "test_name": test_name, "firmware": firmware, "status": status, "error": error, } data["tests"].append(new_entry) f.seek(0) json.dump(data, f, indent=4) f.truncate() def tearDown(self): if not self.installation_finished: self.downloadLogs() if self.report_to_wiki: self.appendResultsToReport() super().tearDown() def run_boot(*modes): """ Decorator to run tests only on specific boot modes ('bios', 'efi'). The VirtMachine has self.is_efi = True/False set. We need to skip the test if self.is_efi is True but 'efi' is not in the modes list. The absence of the decorator is equivalent to run_boot("efi"). :param modes: Boot modes in which the test should run (e.g., "bios", "efi"). """ def decorator(func): func.boot_modes = list(modes) return func return decorator def disk_images(disks): """ Decorator to add installation target disks to the test. :param disks: List of tuples with disk image OS to be used as backing file and size in GB. If the disk image OS is not provided an empty disk will be created. """ def decorator(func): func.disk_images = list(disks) return func return decorator def run_on_vm_setups(*vm_setups): """ Decorator to select tests for particular Virtual Machine setups. The setup is configured via environment variables and defined by TEST_VM_SETUP (see test/run). :param vm_setups: VM setups that this test should run on, include "" if it should run also on default setup """ def decorator(func): func.run_on_vm_setups = list(vm_setups) return func return decorator