1# Copyright (c) 2017 The Chromium OS Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import datetime
6import logging
7import os
8import re
9import time
10
11from autotest_lib.client.bin import utils
12from autotest_lib.client.common_lib import error
13from autotest_lib.server.cros import stress
14from autotest_lib.server.cros.faft.firmware_test import FirmwareTest
15
16class firmware_EmmcWriteLoad(FirmwareTest):
17    """
18    Runs chromeos-install repeatedly while monitoring dmesg output for EMMC
19    timeout errors.
20
21    This test requires a USB disk plugged-in, which contains a Chrome OS test
22    image (built by "build_image test"). On runtime, this test first switches
23    DUT to developer mode. When dev_boot_usb=0, pressing Ctrl-U on developer
24    screen should not boot the USB disk. When dev_boot_usb=1, pressing Ctrl-U
25    should boot the USB disk.
26
27    The length of time in minutes should be specified by the parameter
28    -a 'minutes_to_run=240'
29    """
30    version = 1
31
32    INSTALL_COMMAND = '/usr/sbin/chromeos-install --yes'
33    ERROR_MESSAGE_REGEX = re.compile(
34            r'mmc[0-9]+: Timeout waiting for hardware interrupt', re.MULTILINE)
35
36    def initialize(self, host, cmdline_args, ec_wp=None):
37        dict_args = utils.args_to_dict(cmdline_args)
38        self.minutes_to_run = int(dict_args.get('minutes_to_run', 5))
39        super(firmware_EmmcWriteLoad, self).initialize(
40            host, cmdline_args, ec_wp=ec_wp)
41
42        self.assert_test_image_in_usb_disk()
43        self.switcher.setup_mode('dev')
44        self.setup_usbkey(usbkey=True, host=False)
45
46        self.original_dev_boot_usb = self.faft_client.system.get_dev_boot_usb()
47        logging.info('Original dev_boot_usb value: %s',
48                     str(self.original_dev_boot_usb))
49
50
51    def read_dmesg(self, filename):
52        """Put the contents of 'dmesg -cT' into the given file.
53
54        @param filename: The file to write 'dmesg -cT' into.
55        """
56        with open(filename, 'w') as f:
57            self._client.run('dmesg -cT', stdout_tee=f)
58
59        return utils.read_file(filename)
60
61    def check_for_emmc_error(self, dmesg):
62        """Check the current dmesg output for the specified error message regex.
63
64        @param dmesg: Contents of the dmesg buffer.
65
66        @return True if error found.
67        """
68        for line in dmesg.splitlines():
69            if self.ERROR_MESSAGE_REGEX.search(line):
70                return True
71
72        return False
73
74    def install_chrome_os(self):
75        """Runs the install command. """
76        self.faft_client.system.run_shell_command(self.INSTALL_COMMAND)
77
78    def poll_for_emmc_error(self, dmesg_file, poll_seconds=20):
79        """Continuously polls the contents of dmesg for the emmc failure message
80
81        @param dmesg_file: Contents of the dmesg buffer.
82        @param poll_seconds: Time to wait before checking dmesg again.
83
84        @return True if error found.
85        """
86        end_time = datetime.datetime.now() + \
87                   datetime.timedelta(minutes=self.minutes_to_run)
88
89        while datetime.datetime.now() <= end_time:
90            dmesg = self.read_dmesg(dmesg_file)
91            contains_error = self.check_for_emmc_error(dmesg)
92
93            if contains_error:
94                raise error.TestFail('eMMC error found. Dmesg output: %s' %
95                                     dmesg)
96            time.sleep(poll_seconds)
97
98    def cleanup(self):
99        self.ensure_internal_device_boot()
100        super(firmware_EmmcWriteLoad, self).cleanup()
101
102    def ensure_internal_device_boot(self):
103        """Ensure internal device boot; if not, reboot into it.
104
105        If not, it may be a test failure during step 2 or 3, try to reboot
106        and press Ctrl-D to internal device boot.
107        """
108        if self.faft_client.system.is_removable_device_boot():
109            logging.info('Reboot into internal disk...')
110            self.faft_client.system.set_dev_boot_usb(self.original_dev_boot_usb)
111            self.switcher.mode_aware_reboot()
112
113    def run_once(self):
114        self.faft_client.system.set_dev_boot_usb(1)
115        self.switcher.simple_reboot()
116        self.switcher.bypass_dev_boot_usb()
117        self.switcher.wait_for_client()
118
119        logging.info('Expected USB boot, set dev_boot_usb to the original.')
120        self.check_state((self.checkers.dev_boot_usb_checker,
121                          True,
122                          'Not USB boot, Ctrl-U not work'))
123        stressor = stress.ControlledStressor(self.install_chrome_os)
124
125        dmesg_filename = os.path.join(self.resultsdir, 'dmesg')
126
127        logging.info('===== Starting OS install loop. =====')
128        logging.info('===== Running install for %s minutes. =====',
129                     self.minutes_to_run)
130        stressor.start()
131
132        self.poll_for_emmc_error(dmesg_file=dmesg_filename)
133
134        logging.info('Stopping install loop.')
135        # Usually takes a little over 3 minutes to install so make sure we
136        # wait long enough for a install iteration to complete.
137        stressor.stop(timeout=300)
138
139        logging.info("Installing OS one more time.")
140        # Installing OS one more time to ensure DUT is left in a good state
141        self.install_chrome_os()
142