1# Lint as: python2, python3 2# Copyright (c) 2021 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 6import base64 7import logging 8import os 9 10from autotest_lib.client.common_lib.error import TestFail 11from autotest_lib.server import test 12from autotest_lib.server import utils 13 14 15class chromium(test.test): 16 """Run Chromium tests built on a Skylab DUT.""" 17 18 version = 1 19 20 PROVISION_POINT = '/var/lib/imageloader/lacros' 21 MOUNT_POINT = '/usr/local/tmp/chromium' 22 CHRONOS_RES_DIR = '/home/chronos/user/results' 23 24 def initialize(self, host=None, args=None): 25 self.host = host 26 assert host.path_exists(self.PROVISION_POINT), ( 27 'chromium test artifact is not provisioned by CTP. ' 28 'Please check the CTP request.') 29 self._mount_runtime() 30 args_dict = utils.args_to_dict(args) 31 self.exe_rel_path = args_dict.get('exe_rel_path', '') 32 path_to_executable = os.path.join(self.MOUNT_POINT, self.exe_rel_path) 33 assert self.host.path_exists(path_to_executable), ( 34 'chromium test executable is not mounted at the ' 35 'expected path, %s' % path_to_executable) 36 37 test_args = args_dict.get('test_args') 38 if not test_args: 39 test_args_b64 = args_dict.get('test_args_b64') 40 if test_args_b64: 41 test_args = base64.b64decode(test_args_b64) 42 if isinstance(test_args, bytes): 43 test_args = test_args.decode() 44 self.test_args = test_args 45 46 self.shard_number = args_dict.get('shard_number', 1) 47 self.shard_index = args_dict.get('shard_index', 0) 48 49 def _mount_runtime(self): 50 try: 51 self.host.run( 52 'mkdir -p {mount} && ' 53 'imageloader --mount --mount_component=lacros' 54 ' --mount_point={mount}'.format(mount=self.MOUNT_POINT)) 55 except Exception as e: 56 raise TestFail('Exception while mount test artifact: %s', e) 57 58 def cleanup(self): 59 try: 60 self.host.run('imageloader --unmount --mount_point={mount};' 61 'rm -rf {mount} {chronos_res}'.format( 62 chronos_res=self.CHRONOS_RES_DIR, 63 mount=self.MOUNT_POINT)) 64 except Exception as e: 65 logging.exception('Exception while clear test files: %s', e) 66 67 def run_once(self): 68 cmd = ('{mount}/{exe} ' 69 '--test-launcher-summary-output {chronos_res}/output.json ' 70 '--test-launcher-shard-index {idx} ' 71 '--test-launcher-total-shards {num} '.format( 72 mount=self.MOUNT_POINT, 73 exe=self.exe_rel_path, 74 chronos_res=self.CHRONOS_RES_DIR, 75 idx=self.shard_index, 76 num=self.shard_number)) 77 if self.test_args: 78 cmd += '-- %s' % self.test_args 79 try: 80 self.host.run('su chronos -c -- "%s"' % cmd) 81 except Exception as e: 82 raise TestFail('Exception while execute test: %s', e) 83 finally: 84 self.host.get_file('%s/*' % self.CHRONOS_RES_DIR, 85 self.resultsdir, 86 delete_dest=True) 87