1#!/usr/bin/env python3 2# -- coding: utf-8 -- 3# 4# Copyright (c) 2024-2025 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# 17import shutil 18from datetime import datetime 19from pathlib import Path 20 21import pytz 22 23from runner import utils 24from runner.code_coverage.coverage import LlvmCov 25from runner.common_exceptions import InvalidConfiguration 26from runner.cpumask import CPUMask 27from runner.enum_types.params import TestEnv 28from runner.logger import Log 29from runner.options.config import Config 30from runner.options.step import StepKind 31from runner.runner_file_based import RunnerFileBased 32from runner.suites.test_suite import TestSuite 33from runner.suites.work_dir import WorkDir 34 35_LOGGER = Log.get_logger(__file__) 36 37 38class RunnerStandardFlow(RunnerFileBased): 39 __DEFAULT_SUITE_NAME = "custom-scheme" 40 41 def __init__(self, config: Config): 42 self.suite_name = config.test_suite.suite_name if config.test_suite.suite_name else self.__DEFAULT_SUITE_NAME 43 self.steps = config.workflow.steps 44 RunnerFileBased.__init__(self, config, self.suite_name) 45 46 self.__aot_check() 47 48 self.test_env: TestEnv = TestEnv( 49 config=config, 50 cmd_prefix=self.cmd_prefix, 51 cmd_env=self.cmd_env, 52 timestamp=int(datetime.timestamp(datetime.now(pytz.UTC))), 53 report_formats={self.config.general.report_format}, 54 work_dir=WorkDir(config, self.default_work_dir_root), 55 coverage=LlvmCov(self.config.general.build, self.work_dir), 56 ) 57 58 self.__remove_intermediate_files() 59 60 test_suite: TestSuite = TestSuite(self.test_env) 61 self.tests = set(test_suite.process(self.config.test_suite.ets.force_generate)) 62 self.list_root = test_suite.list_root 63 self.test_root = self.test_env.work_dir.gen 64 65 cpu_mask = self.config.test_suite.get_parameter("mask") 66 self.cpu_mask: CPUMask | None = CPUMask(cpu_mask) \ 67 if cpu_mask is not None and isinstance(cpu_mask, str) else None 68 69 @property 70 def default_work_dir_root(self) -> Path: 71 return Path("/tmp") / self.suite_name 72 73 def create_execution_plan(self) -> None: 74 _LOGGER.default(f"\n{utils.pretty_divider()}") 75 _LOGGER.default(f"Execution plan for the test suite '{self.name}'") 76 _LOGGER.default(self.config.workflow.pretty_str()) 77 78 def create_coverage_html(self) -> None: 79 pass 80 81 def before_suite(self) -> None: 82 if self.cpu_mask: 83 self.cpu_mask.apply() 84 85 def after_suite(self) -> None: 86 if self.cpu_mask: 87 self.cpu_mask.restore() 88 89 def __aot_check(self) -> None: 90 if self.config.general.aot_check: 91 aot_check_passed: bool = False 92 for step in self.steps: 93 if step.executable_path is None: 94 continue 95 all_args = " ".join(step.args) 96 if step.executable_path.name == 'ark_js_vm' or ( 97 step.executable_path.name == 'hdc' and all_args.find('ark_js_vm') > -1 98 ): 99 aot_check_passed = aot_check_passed or all_args.find('--aot-file') > -1 100 if not aot_check_passed: 101 raise InvalidConfiguration( 102 f"AOT check FAILED. Check command line options of steps running `ark_js_vm` in workflow" 103 f"configuration file {self.config.workflow.name}. The expected option `--aot-file` absent." 104 ) 105 _LOGGER.default("AOT check PASSED") 106 107 def __remove_intermediate_files(self) -> None: 108 compiler_step = [step for step in self.steps if step.step_kind == StepKind.COMPILER and step.enabled] 109 if compiler_step: 110 shutil.rmtree(self.test_env.work_dir.intermediate, ignore_errors=True) 111 _LOGGER.default(f"Intermediate folder {self.test_env.work_dir.intermediate} has been removed. " 112 f"Check whether it exists: {self.test_env.work_dir.intermediate.exists()}") 113 self.test_env.work_dir.intermediate.mkdir(parents=True, exist_ok=True) 114