server/site_tests/chromium/chromium.py

# Copyright (c) 2021 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 base64
import logging
import os

from autotest_lib.client.common_lib.error import TestFail
from autotest_lib.server import test
from autotest_lib.server import utils


class chromium(test.test):
    """Run Chromium tests built on a Skylab DUT."""

    version = 1

    PROVISION_POINT = '/var/lib/imageloader/lacros'
    MOUNT_POINT = '/usr/local/tmp/chromium'
    CHRONOS_RES_DIR = '/home/chronos/user/results'

    def initialize(self, host=None, args=None):
        self.host = host
        assert host.path_exists(self.PROVISION_POINT), (
                'chromium test artifact is not provisioned by CTP. '
                'Please check the CTP request.')
        self._mount_runtime()
        args_dict = utils.args_to_dict(args)
        self.exe_rel_path = args_dict.get('exe_rel_path', '')
        path_to_executable = os.path.join(self.MOUNT_POINT, self.exe_rel_path)
        assert self.host.path_exists(path_to_executable), (
                'chromium test executable is not mounted at the '
                'expected path, %s' % path_to_executable)

        test_args = args_dict.get('test_args')
        if not test_args:
            test_args_b64 = args_dict.get('test_args_b64')
            if test_args_b64:
                test_args = base64.b64decode(test_args_b64)
        self.test_args = test_args

        self.shard_number = args_dict.get('shard_number', 1)
        self.shard_index = args_dict.get('shard_index', 0)

    def _mount_runtime(self):
        try:
            self.host.run(
                    'mkdir -p {mount} && '
                    'imageloader --mount --mount_component=lacros'
                    ' --mount_point={mount}'.format(mount=self.MOUNT_POINT))
        except Exception as e:
            raise TestFail('Exception while mount test artifact: %s', e)

    def cleanup(self):
        try:
            self.host.run('imageloader --unmount --mount_point={mount};'
                          'rm -rf {mount} {chronos_res}'.format(
                                  chronos_res=self.CHRONOS_RES_DIR,
                                  mount=self.MOUNT_POINT))
        except Exception as e:
            logging.exception('Exception while clear test files: %s', e)

    def run_once(self):
        cmd = ('{mount}/{exe} '
               '--test-launcher-summary-output {chronos_res}/output.json '
               '--test-launcher-shard-index {idx} '
               '--test-launcher-total-shards {num} '.format(
                       mount=self.MOUNT_POINT,
                       exe=self.exe_rel_path,
                       chronos_res=self.CHRONOS_RES_DIR,
                       idx=self.shard_index,
                       num=self.shard_number))
        if self.test_args:
            cmd += '-- %s' % self.test_args
        try:
            self.host.run('su chronos -c -- "%s"' % cmd)
        except Exception as e:
            raise TestFail('Exception while execute test: %s', e)
        finally:
            self.host.get_file('%s/*' % self.CHRONOS_RES_DIR,
                               self.resultsdir,
                               delete_dest=True)