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