server/site_tests/power_SuspendShutdown/power_SuspendShutdown.py

# Copyright (c) 2012 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.

import json, logging, threading, time, traceback

from autotest_lib.client.common_lib import error
from autotest_lib.server import autotest, test
from autotest_lib.server.cros.faft.config.config import Config as FAFTConfig

_RETRY_SUSPEND_ATTEMPTS = 1
_RETRY_SUSPEND_MS = 10000
_SUSPEND_WAIT_SECONDS = 30
_BOOT_WAIT_SECONDS = 100


class power_SuspendShutdown(test.test):
    """Test power manager fallback to power-off if suspend fails."""
    version = 1

    def initialize(self, host):
        """
        Initial settings before running test.

        @param host: Host/DUT object to run test on.

        """
        # save original boot id
        self.orig_boot_id = host.get_boot_id()
        self.host = host

        # override /sys/power/state via bind mount
        logging.info('binding /dev/full to /sys/power/state')
        host.run('mount --bind /dev/full /sys/power/state')

        # override suspend retry attempts via bind mount
        logging.info('settings retry_suspend_attempts to %s',
                     _RETRY_SUSPEND_ATTEMPTS)
        host.run('echo %s > /tmp/retry_suspend_attempts;'
                 ' mount --bind /tmp/retry_suspend_attempts'
                 ' /usr/share/power_manager/retry_suspend_attempts'
                 % _RETRY_SUSPEND_ATTEMPTS)

        # override suspend retry interval via bind mount
        logging.info('settings retry_suspend_ms to %s',
                     _RETRY_SUSPEND_MS)
        host.run('echo %s > /tmp/retry_suspend_ms;'
                 ' mount --bind /tmp/retry_suspend_ms'
                 ' /usr/share/power_manager/retry_suspend_ms'
                 % _RETRY_SUSPEND_MS)

        # restart powerd to pick up new retry settings
        logging.info('restarting powerd')
        host.run('restart powerd')
        time.sleep(2)


    def platform_check(self, platform_name):
        """
        Raises error if device does not have a lid.

        @param platform_name: Name of the platform

        """
        client_attr = FAFTConfig(platform_name)

        if not client_attr.has_lid:
            raise error.TestError(
                    'This test does nothing on devices without a lid.')

        if client_attr.chrome_ec and not 'lid' in client_attr.ec_capability:
            raise error.TestNAError("TEST IT MANUALLY! Chrome EC can't control "
                    "lid on the device %s" % client_attr.platform)


    def login_into_dut(self, client_autotest, thread_started_evt,
                       exit_without_logout=True):
        """
        Runs the Desktopui_Simple login client test in a seperate thread. The
        Desktopui_Simple client test will exit without logout.

        @param client_autotest: Client autotest name to login into DUT

        @param thread_started_evt: Thread attribute to start the thread

        @param exit_without_logout: if flag is set thread exists without logout.
                                    if not set, thread will wait fot logout
                                    event.

        """
        logging.info('Login into client started')
        thread_started_evt.set()
        try:
            self.autotest_client.run_test(client_autotest,
                                          exit_without_logout=
                                          exit_without_logout)
        except:
            logging.info('DUT login process failed')


    def create_thread(self, client_autotest, exit_without_logout):
        """
        Created seperate thread for client test

        @param client_autotest: Client autotest name to login into DUT

        @param exit_without_logout: if flag is set thread exists without logout.
                                    if not set, thread will wait fot logout
                                    event.
        @return t: thread object

        """
        thread_started_evt = threading.Event()
        logging.info('Launching Desktopui_simplelogin thread')
        try:
            t = threading.Thread(target=self.login_into_dut,
                                 args=(client_autotest,
                                 thread_started_evt, exit_without_logout))
        except:
            raise error.TestError('Thread creation failed')
        t.start()
        thread_started_evt.wait()
        logging.info('Login thread started')
        return t


    def logged_in(self):
        """
        Checks if the host has a logged in user.

        @param host: Host/DUT object

        @return True if a user is logged in on the device.

        """
        host = self.host
        try:
            out = host.run('cryptohome --action=status').stdout.strip()
        except:
            return False
        try:
            status = json.loads(out)
        except ValueError:
            logging.info('Cryptohome did not return a value, retrying.')
            return False

        return any((mount['mounted'] for mount in status['mounts']))


    def run_once(self, client_autotest):
        """
        Run the acutal test on device.

        @param client_autotest: Client autotest name to login into DUT

        @param host: Host/DUT object

        """
        # check platform is capable of running the test
        host = self.host
        platform = host.run_output('mosys platform name')
        self.platform_check(platform)
        self.autotest_client = autotest.Autotest(host)
        logging.info('platform is %s', platform)
        exit_without_logout = True
        t = self.create_thread(client_autotest, exit_without_logout)
        t.join()

        # Waiting for the login thread to finish
        max_wait_time = 15
        for check_count in range(int(max_wait_time)):
            if check_count == max_wait_time:
                raise error.TestError('Login thread is still'
                                      'alive after %s seconds' % max_wait_time)
            if t.is_alive():
                time.sleep(1)
            else:
                logging.info('Login thread successfully finished')
                break

        # close the lid while logged_in to initiate suspend
        logging.info('closing lid')
        host.servo.lid_close()

        # wait for power manager to give up and shut down
        logging.info('waiting for power off')
        host.wait_down(timeout=_SUSPEND_WAIT_SECONDS,
                       old_boot_id=self.orig_boot_id)

        # ensure host is now off
        if host.is_up():
            raise error.TestFail('DUT still up with lid closed')
        else:
            logging.info('good, host is now off')

        # restart host
        host.servo.lid_open()
        host.wait_up(timeout=_BOOT_WAIT_SECONDS)


    def cleanup(self):
        """Clean up the mounts and restore the settings."""
        # reopen lid - might still be closed due to failure
        host = self.host
        logging.info('reopening lid')
        host.servo.lid_open()

        # try to clean up the mess we've made if shutdown failed
        if host.get_boot_id() == self.orig_boot_id:
            # clean up mounts
            logging.info('cleaning up bind mounts')
            host.run('umount /sys/power/state'
                     ' /usr/share/power_manager/retry_suspend_attempts'
                     ' /usr/share/power_manager/retry_suspend_ms',
                     ignore_status=True)

            # restart powerd to pick up old retry settings
            host.run('restart powerd')

        # Reboot Device to logout and cleanup
        logging.info('Server: reboot client')
        try:
            self.host.reboot()
        except error.AutoservRebootError as e:
            raise error.TestFail('%s.\nTest failed with error %s' % (
                    traceback.format_exc(), str(e)))