xref: /aosp_15_r20/external/autotest/server/cros/minios/minios_test.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

# Lint as: python2, python3
# Copyright 2022 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import json
import logging
import os
import re
import time

from autotest_lib.client.common_lib import error
from autotest_lib.server.cros.minios import minios_util
from autotest_lib.server.cros.update_engine import update_engine_test


class MiniOsTest(update_engine_test.UpdateEngineTest):
    """
    Base class that sets up helper objects/functions for NBR tests.

    """

    _MINIOS_CLIENT_CMD = 'minios_client'
    _MINIOS_KERNEL_FLAG = 'cros_minios'

    # Period to wait for firmware screen in seconds.
    # Value based on Brya, which is the slowest so far.
    _FIRMWARE_SCREEN_TIMEOUT = 30

    # Number of times to attempt booting into MiniOS.
    _MINIOS_BOOT_MAX_ATTEMPTS = 3

    # Timeout periods, given in seconds.
    _MINIOS_SHUTDOWN_TIMEOUT = 30

    # Number of seconds to wait for the host to boot into MiniOS. Should always
    # be greater than `_FIRMWARE_SCREEN_TIMEOUT`.
    _MINIOS_WAIT_UP_TIME_SECONDS = 120

    # Version reported to OMAHA/NEBRASKA for recovery.
    _RECOVERY_VERSION = '0.0.0.0'

    # Files used by the tests.
    _DEPENDENCY_DIRS = ['bin', 'lib', 'lib64', 'libexec']
    _DEPENDENCY_INSTALL_DIR = '/usr/local'
    _MINIOS_TEMP_STATEFUL_DIR = '/usr/local/tmp/stateful'
    _STATEFUL_DEV_IMAGE_NAME = 'dev_image_new'

    def initialize(self, host):
        """
        Sets default variables for the test.

        @param host: The DUT we will be running on.

        """
        super(MiniOsTest, self).initialize(host)
        self._nebraska = None
        self._servo = host.servo
        self._servo.initialize_dut()

    def cleanup(self):
        """Clean up minios autotests."""
        if self._nebraska:
            self._nebraska.stop()
        super(MiniOsTest, self).cleanup()
        # Make sure to reboot DUT into CroS in case of failures.
        self._host.reboot()

    def _boot_minios(self):
        """Boot the DUT into MiniOS."""
        # Turn off usbkey to avoid booting into usb-recovery image.
        self._servo.switch_usbkey('off')
        psc = self._servo.get_power_state_controller()
        psc.power_off()
        psc.power_on(psc.REC_ON)
        self._host.test_wait_for_shutdown(self._MINIOS_SHUTDOWN_TIMEOUT)
        logging.info('Waiting for firmware screen')
        time.sleep(self._FIRMWARE_SCREEN_TIMEOUT)

        # Attempt multiple times to boot into MiniOS. If all attempts fail then
        # this is some kind of firmware issue. Since we failed to boot an OS use
        # servo to reset the unit and then report test failure.
        attempts = 0
        minios_is_up = False
        while not minios_is_up and attempts < self._MINIOS_BOOT_MAX_ATTEMPTS:
            # Use Ctrl+R shortcut to boot 'MiniOS
            logging.info('Try to boot MiniOS')
            self._servo.ctrl_r()
            minios_is_up = self._host.wait_up(
                    timeout=self._MINIOS_WAIT_UP_TIME_SECONDS,
                    host_is_down=True)
            attempts += 1

        if minios_is_up:
            # If mainfw_type is recovery then we are in MiniOS.
            mainfw_type = self._host.run_output('crossystem mainfw_type')
            if mainfw_type != 'recovery':
                raise error.TestError(
                        'Boot to MiniOS - invalid firmware: %s.' % mainfw_type)
            # There are multiple types of recovery images, make sure we booted
            # into minios.
            pattern = r'\b%s\b' % self._MINIOS_KERNEL_FLAG
            if not re.search(pattern, self._host.get_cmdline()):
                raise error.TestError(
                        'Boot to MiniOS - recovery image is not minios.')
        else:
            # Try to not leave unit on recovery firmware screen.
            self._host.power_cycle()
            raise error.TestError('Boot to MiniOS failed.')

    def _create_minios_hostlog(self):
        """Create the minios hostlog file.

        To ensure the recovery was successful we need to compare the update
        events against expected update events. This function creates the hostlog
        for minios before the recovery reboots the DUT.

        """
        # Check that update logs exist.
        if len(self._get_update_engine_logs()) < 1:
            err_msg = 'update_engine logs are missing. Cannot verify recovery.'
            raise error.TestFail(err_msg)

        # Download the logs instead of reading it over the network since it will
        # disappear after MiniOS reboots the DUT.
        logfile = os.path.join(self.resultsdir, 'minios_update_engine.log')
        self._host.get_file(self._UPDATE_ENGINE_LOG, logfile)
        logfile_content = None
        with open(logfile) as f:
            logfile_content = f.read()
        minios_hostlog = os.path.join(self.resultsdir, 'hostlog_minios')
        with open(minios_hostlog, 'w') as fp:
            # There are four expected hostlog events during recovery.
            extract_logs = self._extract_request_logs(logfile_content)
            json.dump(extract_logs[-4:], fp)
        return minios_hostlog

    def _install_test_dependencies(self, public_bucket=False):
        """
        Install test dependencies from a downloaded stateful archive.

        @param public_bucket: True to download stateful from a public bucket.

        """
        if not self._job_repo_url:
            raise error.TestError('No job repo url set.')

        statefuldev_url = self._stage_stateful(public_bucket)
        logging.info('Installing dependencies from %s', statefuldev_url)

        # Create destination directories.
        minios_dev_image_dir = os.path.join(self._MINIOS_TEMP_STATEFUL_DIR,
                                            self._STATEFUL_DEV_IMAGE_NAME)
        install_dirs = [
                os.path.join(self._DEPENDENCY_INSTALL_DIR, dir)
                for dir in self._DEPENDENCY_DIRS
        ]
        self._run(['mkdir', '-p', minios_dev_image_dir] + install_dirs)
        # Symlink the install dirs into the staging destination.
        for dir in install_dirs:
            self._run(['ln', '-s', dir, minios_dev_image_dir])

        # Generate the list of stateful archive members that we want to extract.
        members = [
                os.path.join(self._STATEFUL_DEV_IMAGE_NAME, dir)
                for dir in self._DEPENDENCY_DIRS
        ]
        try:
            self._download_and_extract_stateful(statefuldev_url,
                                                self._MINIOS_TEMP_STATEFUL_DIR,
                                                members=members,
                                                keep_symlinks=True)
        except error.AutoservRunError as e:
            err_str = 'Failed to install the test dependencies'
            raise error.TestFail('%s: %s' % (err_str, str(e)))

        self._setup_python_symlinks()

        # Clean-up unused files to save memory.
        self._run(['rm', '-rf', self._MINIOS_TEMP_STATEFUL_DIR])

    def _setup_python_symlinks(self):
        """
        Create symlinks in the root to point to all python paths in /usr/local
        for stateful installed python to work. This is needed because Gentoo
        creates wrappers with hardcoded paths to the root (e.g. python-exec).

        """
        for path in self._DEPENDENCY_DIRS:
            self._run([
                    'find',
                    os.path.join(self._DEPENDENCY_INSTALL_DIR, path),
                    '-maxdepth', '1', '\(', '-name', 'python*', '-o', '-name',
                    'portage', '\)', '-exec', 'ln', '-s', '{}',
                    os.path.join('/usr', path), '\;'
            ])

    def _start_nebraska(self, payload_url=None):
        """
        Initialize and start nebraska on the DUT.

        @param payload_url: The payload to served by nebraska.

        """
        if not self._nebraska:
            self._nebraska = minios_util.NebraskaService(
                    self, self._host, payload_url)
        self._nebraska.start()

    def _verify_reboot(self, old_boot_id):
        """
        Verify that the unit rebooted using the boot_id.

        @param old_boot_id A boot id value obtained before the
                               reboot.

        """
        self._host.test_wait_for_shutdown(self._MINIOS_SHUTDOWN_TIMEOUT)
        self._host.test_wait_for_boot(old_boot_id)