1# Lint as: python2, python3 2# Copyright 2022 The Chromium OS Authors. All rights reserved. 3# Use of this source code is governed by a BSD-style license that can be 4# found in the LICENSE file. 5 6from __future__ import absolute_import 7from __future__ import division 8from __future__ import print_function 9 10import json 11import logging 12import os 13import re 14import time 15 16from autotest_lib.client.common_lib import error 17from autotest_lib.server.cros.minios import minios_util 18from autotest_lib.server.cros.update_engine import update_engine_test 19 20 21class MiniOsTest(update_engine_test.UpdateEngineTest): 22 """ 23 Base class that sets up helper objects/functions for NBR tests. 24 25 """ 26 27 _MINIOS_CLIENT_CMD = 'minios_client' 28 _MINIOS_KERNEL_FLAG = 'cros_minios' 29 30 # Period to wait for firmware screen in seconds. 31 # Value based on Brya, which is the slowest so far. 32 _FIRMWARE_SCREEN_TIMEOUT = 30 33 34 # Number of times to attempt booting into MiniOS. 35 _MINIOS_BOOT_MAX_ATTEMPTS = 3 36 37 # Timeout periods, given in seconds. 38 _MINIOS_SHUTDOWN_TIMEOUT = 30 39 40 # Number of seconds to wait for the host to boot into MiniOS. Should always 41 # be greater than `_FIRMWARE_SCREEN_TIMEOUT`. 42 _MINIOS_WAIT_UP_TIME_SECONDS = 120 43 44 # Version reported to OMAHA/NEBRASKA for recovery. 45 _RECOVERY_VERSION = '0.0.0.0' 46 47 # Files used by the tests. 48 _DEPENDENCY_DIRS = ['bin', 'lib', 'lib64', 'libexec'] 49 _DEPENDENCY_INSTALL_DIR = '/usr/local' 50 _MINIOS_TEMP_STATEFUL_DIR = '/usr/local/tmp/stateful' 51 _STATEFUL_DEV_IMAGE_NAME = 'dev_image_new' 52 53 def initialize(self, host): 54 """ 55 Sets default variables for the test. 56 57 @param host: The DUT we will be running on. 58 59 """ 60 super(MiniOsTest, self).initialize(host) 61 self._nebraska = None 62 self._servo = host.servo 63 self._servo.initialize_dut() 64 65 def cleanup(self): 66 """Clean up minios autotests.""" 67 if self._nebraska: 68 self._nebraska.stop() 69 super(MiniOsTest, self).cleanup() 70 # Make sure to reboot DUT into CroS in case of failures. 71 self._host.reboot() 72 73 def _boot_minios(self): 74 """Boot the DUT into MiniOS.""" 75 # Turn off usbkey to avoid booting into usb-recovery image. 76 self._servo.switch_usbkey('off') 77 psc = self._servo.get_power_state_controller() 78 psc.power_off() 79 psc.power_on(psc.REC_ON) 80 self._host.test_wait_for_shutdown(self._MINIOS_SHUTDOWN_TIMEOUT) 81 logging.info('Waiting for firmware screen') 82 time.sleep(self._FIRMWARE_SCREEN_TIMEOUT) 83 84 # Attempt multiple times to boot into MiniOS. If all attempts fail then 85 # this is some kind of firmware issue. Since we failed to boot an OS use 86 # servo to reset the unit and then report test failure. 87 attempts = 0 88 minios_is_up = False 89 while not minios_is_up and attempts < self._MINIOS_BOOT_MAX_ATTEMPTS: 90 # Use Ctrl+R shortcut to boot 'MiniOS 91 logging.info('Try to boot MiniOS') 92 self._servo.ctrl_r() 93 minios_is_up = self._host.wait_up( 94 timeout=self._MINIOS_WAIT_UP_TIME_SECONDS, 95 host_is_down=True) 96 attempts += 1 97 98 if minios_is_up: 99 # If mainfw_type is recovery then we are in MiniOS. 100 mainfw_type = self._host.run_output('crossystem mainfw_type') 101 if mainfw_type != 'recovery': 102 raise error.TestError( 103 'Boot to MiniOS - invalid firmware: %s.' % mainfw_type) 104 # There are multiple types of recovery images, make sure we booted 105 # into minios. 106 pattern = r'\b%s\b' % self._MINIOS_KERNEL_FLAG 107 if not re.search(pattern, self._host.get_cmdline()): 108 raise error.TestError( 109 'Boot to MiniOS - recovery image is not minios.') 110 else: 111 # Try to not leave unit on recovery firmware screen. 112 self._host.power_cycle() 113 raise error.TestError('Boot to MiniOS failed.') 114 115 def _create_minios_hostlog(self): 116 """Create the minios hostlog file. 117 118 To ensure the recovery was successful we need to compare the update 119 events against expected update events. This function creates the hostlog 120 for minios before the recovery reboots the DUT. 121 122 """ 123 # Check that update logs exist. 124 if len(self._get_update_engine_logs()) < 1: 125 err_msg = 'update_engine logs are missing. Cannot verify recovery.' 126 raise error.TestFail(err_msg) 127 128 # Download the logs instead of reading it over the network since it will 129 # disappear after MiniOS reboots the DUT. 130 logfile = os.path.join(self.resultsdir, 'minios_update_engine.log') 131 self._host.get_file(self._UPDATE_ENGINE_LOG, logfile) 132 logfile_content = None 133 with open(logfile) as f: 134 logfile_content = f.read() 135 minios_hostlog = os.path.join(self.resultsdir, 'hostlog_minios') 136 with open(minios_hostlog, 'w') as fp: 137 # There are four expected hostlog events during recovery. 138 extract_logs = self._extract_request_logs(logfile_content) 139 json.dump(extract_logs[-4:], fp) 140 return minios_hostlog 141 142 def _install_test_dependencies(self, public_bucket=False): 143 """ 144 Install test dependencies from a downloaded stateful archive. 145 146 @param public_bucket: True to download stateful from a public bucket. 147 148 """ 149 if not self._job_repo_url: 150 raise error.TestError('No job repo url set.') 151 152 statefuldev_url = self._stage_stateful(public_bucket) 153 logging.info('Installing dependencies from %s', statefuldev_url) 154 155 # Create destination directories. 156 minios_dev_image_dir = os.path.join(self._MINIOS_TEMP_STATEFUL_DIR, 157 self._STATEFUL_DEV_IMAGE_NAME) 158 install_dirs = [ 159 os.path.join(self._DEPENDENCY_INSTALL_DIR, dir) 160 for dir in self._DEPENDENCY_DIRS 161 ] 162 self._run(['mkdir', '-p', minios_dev_image_dir] + install_dirs) 163 # Symlink the install dirs into the staging destination. 164 for dir in install_dirs: 165 self._run(['ln', '-s', dir, minios_dev_image_dir]) 166 167 # Generate the list of stateful archive members that we want to extract. 168 members = [ 169 os.path.join(self._STATEFUL_DEV_IMAGE_NAME, dir) 170 for dir in self._DEPENDENCY_DIRS 171 ] 172 try: 173 self._download_and_extract_stateful(statefuldev_url, 174 self._MINIOS_TEMP_STATEFUL_DIR, 175 members=members, 176 keep_symlinks=True) 177 except error.AutoservRunError as e: 178 err_str = 'Failed to install the test dependencies' 179 raise error.TestFail('%s: %s' % (err_str, str(e))) 180 181 self._setup_python_symlinks() 182 183 # Clean-up unused files to save memory. 184 self._run(['rm', '-rf', self._MINIOS_TEMP_STATEFUL_DIR]) 185 186 def _setup_python_symlinks(self): 187 """ 188 Create symlinks in the root to point to all python paths in /usr/local 189 for stateful installed python to work. This is needed because Gentoo 190 creates wrappers with hardcoded paths to the root (e.g. python-exec). 191 192 """ 193 for path in self._DEPENDENCY_DIRS: 194 self._run([ 195 'find', 196 os.path.join(self._DEPENDENCY_INSTALL_DIR, path), 197 '-maxdepth', '1', '\(', '-name', 'python*', '-o', '-name', 198 'portage', '\)', '-exec', 'ln', '-s', '{}', 199 os.path.join('/usr', path), '\;' 200 ]) 201 202 def _start_nebraska(self, payload_url=None): 203 """ 204 Initialize and start nebraska on the DUT. 205 206 @param payload_url: The payload to served by nebraska. 207 208 """ 209 if not self._nebraska: 210 self._nebraska = minios_util.NebraskaService( 211 self, self._host, payload_url) 212 self._nebraska.start() 213 214 def _verify_reboot(self, old_boot_id): 215 """ 216 Verify that the unit rebooted using the boot_id. 217 218 @param old_boot_id A boot id value obtained before the 219 reboot. 220 221 """ 222 self._host.test_wait_for_shutdown(self._MINIOS_SHUTDOWN_TIMEOUT) 223 self._host.test_wait_for_boot(old_boot_id) 224