1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3 4# Copyright (c) 2021-2023 Huawei Device Co., Ltd. 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16# 17# This file does only contain a selection of the most common options. For a 18# full list see the documentation: 19# http://www.sphinx-doc.org/en/master/config 20 21import logging 22import subprocess 23from copy import deepcopy 24from os import path, remove 25from typing import List, Callable, Tuple, Optional 26 27from runner.enum_types.configuration_kind import ConfigurationKind 28from runner.enum_types.fail_kind import FailKind 29from runner.enum_types.params import TestEnv, Params, TestReport 30from runner.logger import Log 31from runner.options.options_jit import JitOptions 32from runner.test_base import Test 33 34_LOGGER = logging.getLogger("runner.test_file_based") 35 36ResultValidator = Callable[[str, str, int], bool] 37 38 39class TestFileBased(Test): 40 def __init__(self, test_env: TestEnv, test_path: str, flags: List[str], test_id: str) -> None: 41 Test.__init__(self, test_env, test_path, flags, test_id) 42 # If test fails it contains reason (of FailKind enum) of first failed step 43 # It's supposed if the first step is failed then no step is executed further 44 self.fail_kind: Optional[FailKind] = None 45 self.main_entry_point = "_GLOBAL::func_main_0" 46 47 def ark_extra_options(self) -> List[str]: 48 return [] 49 50 @property 51 def ark_timeout(self) -> int: 52 return int(self.test_env.config.ark.timeout) 53 54 @property 55 def runtime_args(self) -> List[str]: 56 return self.test_env.runtime_args 57 58 @property 59 def verifier_args(self) -> List[str]: 60 return self.test_env.verifier_args 61 62 # pylint: disable=too-many-locals 63 def run_one_step(self, name: str, params: Params, result_validator: ResultValidator) \ 64 -> Tuple[bool, TestReport, Optional[FailKind]]: 65 if self.test_env.config.general.coverage.use_llvm_cov: 66 params = deepcopy(params) 67 profraw_file, profdata_file = self.test_env.coverage.get_uniq_profraw_profdata_file_paths() 68 params.env['LLVM_PROFILE_FILE'] = profraw_file 69 70 cmd = self.test_env.cmd_prefix + [params.executor] 71 cmd.extend(params.flags) 72 73 self.log_cmd(cmd) 74 Log.all(_LOGGER, f"Run {name}: {' '.join(cmd)}") 75 76 passed = False 77 output = "" 78 79 with subprocess.Popen( 80 cmd, 81 stdout=subprocess.PIPE, 82 stderr=subprocess.PIPE, 83 env=params.env, 84 encoding='utf-8', 85 errors='ignore', 86 ) as process: 87 fail_kind: Optional[FailKind] = None 88 try: 89 output, error = process.communicate(timeout=params.timeout) 90 return_code = process.returncode 91 passed = result_validator(output, error, return_code) 92 if not passed: 93 fail_kind = params.fail_kind_fail 94 except subprocess.TimeoutExpired: 95 self.log_cmd(f"Failed by timeout after {params.timeout} sec") 96 fail_kind = params.fail_kind_timeout 97 error = fail_kind.name 98 return_code = process.returncode 99 process.kill() 100 except Exception as ex: # pylint: disable=broad-except 101 self.log_cmd(f"Failed with {ex}") 102 fail_kind = params.fail_kind_other 103 error = fail_kind.name 104 return_code = -1 105 106 if self.test_env.config.general.coverage.use_llvm_cov: 107 self.test_env.coverage.merge_and_delete_prowraw_files(profraw_file, profdata_file) 108 109 report = TestReport( 110 output=output, 111 error=error, 112 return_code=return_code 113 ) 114 115 return passed, report, fail_kind 116 117 def run_es2panda(self, flags: List[str], test_abc: str, result_validator: ResultValidator) \ 118 -> Tuple[bool, TestReport, Optional[FailKind]]: 119 es2panda_flags = flags[:] 120 es2panda_flags.append("--thread=0") 121 if len(test_abc) > 0: 122 es2panda_flags.append(f"--output={test_abc}") 123 124 es2panda_flags.append(self.path) 125 126 params = Params( 127 executor=self.test_env.es2panda, 128 flags=es2panda_flags, 129 env=self.test_env.cmd_env, 130 timeout=self.test_env.config.es2panda.timeout, 131 fail_kind_fail=FailKind.ES2PANDA_FAIL, 132 fail_kind_timeout=FailKind.ES2PANDA_TIMEOUT, 133 fail_kind_other=FailKind.ES2PANDA_OTHER, 134 ) 135 136 return self.run_one_step("es2panda", params, result_validator) 137 138 def run_runtime(self, test_an: str, test_abc: str, result_validator: ResultValidator) \ 139 -> Tuple[bool, TestReport, Optional[FailKind]]: 140 ark_flags = [] 141 ark_flags.extend(self.ark_extra_options()) 142 ark_flags.extend(self.runtime_args) 143 if self.test_env.conf_kind in [ConfigurationKind.AOT, ConfigurationKind.AOT_FULL]: 144 ark_flags.extend(["--aot-files", test_an]) 145 146 if self.test_env.conf_kind == ConfigurationKind.JIT: 147 ark_flags.extend([ 148 '--compiler-enable-jit=true', 149 '--compiler-check-final=true']) 150 jit_options: JitOptions = self.test_env.config.ark.jit 151 if jit_options.compiler_threshold is not None: 152 ark_flags.append( 153 f'--compiler-hotness-threshold={jit_options.compiler_threshold}' 154 ) 155 else: 156 ark_flags.extend(['--compiler-enable-jit=false']) 157 158 if self.test_env.config.ark.interpreter_type is not None: 159 ark_flags.extend([f'--interpreter-type={self.test_env.config.ark.interpreter_type}']) 160 161 ark_flags.extend([test_abc, self.main_entry_point]) 162 163 params = Params( 164 timeout=self.ark_timeout, 165 executor=self.test_env.runtime, 166 flags=ark_flags, 167 env=self.test_env.cmd_env, 168 fail_kind_fail=FailKind.RUNTIME_FAIL, 169 fail_kind_timeout=FailKind.RUNTIME_TIMEOUT, 170 fail_kind_other=FailKind.RUNTIME_OTHER, 171 ) 172 173 return self.run_one_step("ark", params, result_validator) 174 175 def run_aot(self, test_an: str, test_abc: str, result_validator: ResultValidator) \ 176 -> Tuple[bool, TestReport, Optional[FailKind]]: 177 aot_flags = [] 178 aot_flags.extend(self.test_env.aot_args) 179 aot_flags = [flag.strip("'\"") for flag in aot_flags] 180 aot_flags.extend(['--paoc-panda-files', test_abc]) 181 aot_flags.extend(['--paoc-output', test_an]) 182 183 if path.isfile(test_an): 184 remove(test_an) 185 186 assert self.test_env.ark_aot is not None 187 params = Params( 188 timeout=self.test_env.config.ark_aot.timeout, 189 executor=self.test_env.ark_aot, 190 flags=aot_flags, 191 env=self.test_env.cmd_env, 192 fail_kind_fail=FailKind.AOT_FAIL, 193 fail_kind_timeout=FailKind.AOT_TIMEOUT, 194 fail_kind_other=FailKind.AOT_OTHER, 195 ) 196 197 return self.run_one_step("ark_aot", params, result_validator) 198 199 def run_ark_quick(self, flags: List[str], test_abc: str, result_validator: ResultValidator) \ 200 -> Tuple[bool, TestReport, Optional[FailKind], str]: 201 quick_flags = flags[:] 202 quick_flags.extend(self.test_env.quick_args) 203 204 src_abc = test_abc 205 root, ext = path.splitext(src_abc) 206 dst_abc = f'{root}.quick{ext}' 207 quick_flags.extend([src_abc, dst_abc]) 208 209 params = Params( 210 timeout=self.ark_timeout, 211 executor=self.test_env.ark_quick, 212 flags=quick_flags, 213 env=self.test_env.cmd_env, 214 fail_kind_fail=FailKind.QUICK_FAIL, 215 fail_kind_timeout=FailKind.QUICK_TIMEOUT, 216 fail_kind_other=FailKind.QUICK_OTHER, 217 ) 218 219 return *(self.run_one_step("ark_quick", params, result_validator)), dst_abc 220