server/cros/update_engine/chromiumos_test_platform.py

# Copyright 2017 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 logging
import os
import urlparse

from autotest_lib.server import autotest

class ChromiumOSTestPlatform(object):
    """Represents a CrOS device during autoupdate.

    This class is used with autoupdate_EndToEndTest. It has functions for all
    the device specific things that we need during an update: reboot,
    check active slot, login, get logs, start an update etc.
    """

    _UPDATE_ENGINE_PERF_PATH = '/mnt/stateful_partition/unencrypted/preserve'
    _UPDATE_ENGINE_PERF_SCRIPT = 'update_engine_performance_monitor.py'
    _UPDATE_ENGINE_PERF_RESULTS_FILE = 'perf_data_results.json'

    def __init__(self, host, autotest_devserver, results_dir):
        """Initialize the class.

        @param: host: The DUT host.
        @param: autotest_devserver: The devserver to call cros_au on.
        @param: results_dir: Where to save the autoupdate logs files.
        """
        self._host = host
        self._autotest_devserver = autotest_devserver
        self._results_dir = results_dir


    def _install_version(self, payload_uri, clobber_stateful=False):
        """Install the specified payload.

        @param payload_uri: GS URI of the payload to install.
        @param clobber_stateful: force a reinstall of the stateful image.
        """
        build_name, payload_file = self._get_update_parameters_from_uri(
            payload_uri)
        logging.info('Installing %s on the DUT', payload_uri)

        try:
            ds = self._autotest_devserver
            _, pid = ds.auto_update(host_name=self._host.hostname,
                                    build_name=build_name,
                                    force_update=True,
                                    full_update=True,
                                    log_dir=self._results_dir,
                                    payload_filename=payload_file,
                                    clobber_stateful=clobber_stateful)
        except:
            logging.fatal('ERROR: Failed to install image on the DUT.')
            raise
        return pid


    def _run_login_test(self, tag):
        """Runs login_LoginSuccess test on the DUT."""
        client_at = autotest.Autotest(self._host)
        client_at.run_test('login_LoginSuccess', tag=tag)


    @staticmethod
    def _get_update_parameters_from_uri(payload_uri):
        """Extract the two vars needed for cros_au from the Google Storage URI.

        dev_server.auto_update needs two values from this test:
        (1) A build_name string e.g samus-release/R60-9583.0.0
        (2) A filename of the exact payload file to use for the update. This
        payload needs to have already been staged on the devserver.

        This function extracts those two values from a Google Storage URI.

        @param payload_uri: Google Storage URI to extract values from
        """
        archive_url, _, payload_file = payload_uri.rpartition('/')
        build_name = urlparse.urlsplit(archive_url).path.strip('/')

        # This test supports payload uris from two Google Storage buckets.
        # They store their payloads slightly differently. One stores them in
        # a separate payloads directory. E.g
        # gs://chromeos-image-archive/samus-release/R60-9583.0.0/blah.bin
        # gs://chromeos-releases/dev-channel/samus/9334.0.0/payloads/blah.bin
        if build_name.endswith('payloads'):
            build_name = build_name.rpartition('/')[0]
            payload_file = 'payloads/' + payload_file

        logging.debug('Extracted build_name: %s, payload_file: %s from %s.',
                      build_name, payload_file, payload_uri)
        return build_name, payload_file


    def reboot_device(self):
        """Reboot the device."""
        self._host.reboot()


    def install_source_image(self, source_payload_uri):
        """Install source payload on device."""
        if source_payload_uri:
            self._install_version(source_payload_uri, clobber_stateful=True)


    def check_login_after_source_update(self):
        """Make sure we can login before the target update."""
        self._run_login_test('source_update')


    def get_active_slot(self):
        """Returns the current active slot."""
        return self._host.run('rootdev -s').stdout.strip()


    def copy_perf_script_to_device(self, bindir):
        """Copy performance monitoring script to DUT.

        The updater will kick off the script during the update.
        """
        logging.info('Copying %s to device.', self._UPDATE_ENGINE_PERF_SCRIPT)
        path = os.path.join(bindir, self._UPDATE_ENGINE_PERF_SCRIPT)
        self._host.send_file(path, self._UPDATE_ENGINE_PERF_PATH)


    def get_perf_stats_for_update(self, resultdir):
        """ Get the performance metrics created during update."""
        try:
            path = os.path.join('/var/log',
                                self._UPDATE_ENGINE_PERF_RESULTS_FILE)
            self._host.get_file(path, resultdir)
            self._host.run('rm %s' % path)
            script = os.path.join(self._UPDATE_ENGINE_PERF_PATH,
                                  self._UPDATE_ENGINE_PERF_SCRIPT)
            self._host.run('rm %s' % script)
            return os.path.join(resultdir,
                                self._UPDATE_ENGINE_PERF_RESULTS_FILE)
        except:
            logging.warning('Failed to copy performance metrics from DUT.')
            return None


    def install_target_image(self, target_payload_uri):
        """Install target payload on the device."""
        logging.info('Updating device to target image.')
        return self._install_version(target_payload_uri)


    def get_update_log(self, num_lines):
        """Get the latest lines from the update engine log."""
        return self._host.run_output(
                'tail -n %d /var/log/update_engine.log' % num_lines,
                stdout_tee=None)


    def check_login_after_target_update(self):
        """Check we can login after updating."""
        self._run_login_test('target_update')


    def oobe_triggers_update(self):
        """Check if this device has an OOBE that completes itself."""
        return self._host.oobe_triggers_update()


    def get_cros_version(self):
        """Returns the ChromeOS version installed on this device."""
        return self._host.get_release_version()