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# 17 18from dataclasses import dataclass 19from pathlib import Path 20from typing import Any, cast 21 22from runner import utils 23from runner.common_exceptions import FileNotFoundException, IncorrectEnumValue, MalformedStepConfigurationException 24from runner.enum_types.base_enum import BaseEnum 25from runner.logger import Log 26from runner.options.options import IOptions 27from runner.utils import to_bool 28 29_LOGGER = Log.get_logger(__file__) 30 31PropType = str | int | bool | list[str] | dict[str, Any] | BaseEnum # type: ignore[explicit-any] 32 33 34class StepKind(BaseEnum): 35 COMPILER = "compiler" 36 VERIFIER = "verifier" 37 AOT = "aot" 38 RUNTIME = "runtime" 39 OTHER = "other" 40 41 42@dataclass 43class Step(IOptions): 44 name: str 45 timeout: int 46 args: list[str] 47 env: dict[str, str] 48 step_kind: StepKind 49 enabled: bool = True 50 executable_path: Path | None = None 51 can_be_instrumented: bool = False 52 DEFAULT_TIMEOUT = 30 53 54 __MIN_ARG_LENGTH = 15 55 __MAX_ARG_LENGTH = 150 56 __INDENT_STEP = "\t" 57 __INDENT_ARG = __INDENT_STEP * 2 58 __INDENT_SUB_ARG = __INDENT_STEP * 3 59 60 def __init__(self, name: str, step_body: dict[str, Any]): # type: ignore[explicit-any] 61 super().__init__() 62 self.name = name 63 self.executable_path = self.__get_path_property(step_body, 'executable-path') 64 self.timeout = self.__get_int_property(step_body, 'timeout', self.DEFAULT_TIMEOUT) 65 self.can_be_instrumented = self.__get_bool_property(step_body, 'can-be-instrumented', False) 66 self.enabled = self.__get_bool_property(step_body, 'enabled', True) 67 self.step_kind = self.__get_kind_property(step_body, 'step-type', StepKind.OTHER) 68 self.args = self.__get_list_property(step_body, 'args', []) 69 self.env = self.__get_dict_property(step_body, 'env', {}) 70 71 def __str__(self) -> str: 72 indent = 3 73 result = [f"{utils.indent(indent)}{self.name}:"] 74 indent += 1 75 result += [f"{utils.indent(indent)}executable-path: {self.executable_path}"] 76 result += [f"{utils.indent(indent)}timeout: {self.timeout}"] 77 result += [f"{utils.indent(indent)}can-be-instrumented: {self.can_be_instrumented}"] 78 result += [f"{utils.indent(indent)}enabled: {self.enabled}"] 79 result += [f"{utils.indent(indent)}step-type: {self.step_kind}"] 80 81 if self.args: 82 result += self.__str_process_args(indent) 83 84 if self.env: 85 result += self.__str_process_env(indent) 86 return "\n".join(result) 87 88 @staticmethod 89 def __get_int_property(step_body: dict[str, Any], # type: ignore[explicit-any] 90 name: str, default: int | None = None) -> int: 91 value = Step.__get_property(step_body, name, default) 92 try: 93 return int(value) 94 except ValueError: 95 return False 96 97 @staticmethod 98 def __get_bool_property(step_body: dict[str, Any], # type: ignore[explicit-any] 99 name: str, default: bool | None = None) -> bool: 100 value = Step.__get_property(step_body, name, default) 101 try: 102 return to_bool(value) 103 except ValueError as exc: 104 raise MalformedStepConfigurationException(f"Incorrect value '{value}' for property '{name}'") from exc 105 106 @staticmethod 107 def __get_list_property(step_body: dict[str, Any], name: str, # type: ignore[explicit-any] 108 default: list[str] | None = None) -> list[str]: 109 value = Step.__get_property(step_body, name, default) 110 if not isinstance(value, list): 111 raise MalformedStepConfigurationException(f"Incorrect value '{value}' for property '{name}'. " 112 "Expected list.") 113 return cast(list[str], value) 114 115 @staticmethod 116 def __get_dict_property(step_body: dict[str, Any], name: str, # type: ignore[explicit-any] 117 default: dict[str, Any] | None = None) -> dict[str, str]: 118 value = Step.__get_property(step_body, name, default) 119 if not isinstance(value, dict): 120 raise MalformedStepConfigurationException(f"Incorrect value '{value}' for property '{name}'. " 121 "Expected dict.") 122 return cast(dict[str, str], value) 123 124 @staticmethod 125 def __get_str_property(step_body: dict[str, Any], name: str, # type: ignore[explicit-any] 126 default: str | None = None) -> str: 127 value = Step.__get_property(step_body, name, default) 128 if isinstance(value, (dict | list)): 129 raise MalformedStepConfigurationException(f"Incorrect value '{value}' for property '{name}'. " 130 "Expected str.") 131 return str(value) 132 133 @staticmethod 134 def __get_path_property(step_body: dict[str, Any], name: str) -> Path | None: # type: ignore[explicit-any] 135 value = Step.__get_str_property(step_body, name, "") 136 if value == "": 137 return None 138 value_path = Path(value) 139 if not value_path.exists(): 140 raise FileNotFoundException(f"Cannot find {value_path}. Check value of 'executable-path'") 141 return value_path 142 143 @staticmethod 144 def __get_kind_property(step_body: dict[str, Any], name: str, # type: ignore[explicit-any] 145 default: StepKind | None = None) -> StepKind: 146 value = Step.__get_property(step_body, name, default) 147 try: 148 if isinstance(value, StepKind): 149 return value 150 return StepKind.is_value(value, option_name="'step-type' from workflow config") 151 except IncorrectEnumValue as exc: 152 raise MalformedStepConfigurationException(f"Incorrect value '{value}' for property '{name}'. " 153 f"Expected one of {StepKind.values()}.") from exc 154 155 @staticmethod 156 def __get_property(step_body: dict[str, Any], # type: ignore[explicit-any] 157 name: str, 158 default: PropType | None = None) -> \ 159 Any: 160 if name in step_body: 161 value = step_body[name] 162 elif default is not None: 163 value = default 164 else: 165 raise MalformedStepConfigurationException(f"missed required field '{name}'") 166 return value 167 168 def pretty_str(self) -> str: 169 if not str(self.executable_path): 170 return '' 171 result = f"Step '{self.name}':" 172 args: list[str] = [self.__pretty_arg_str(arg) for arg in self.args] 173 args_str = '\n'.join(args) 174 result += f"\n{self.__INDENT_STEP}{self.executable_path}\n{args_str}" 175 return result 176 177 def __pretty_arg_str(self, arg: str) -> str: 178 args: list[str] = [] 179 arg = str(arg) 180 if len(arg) < self.__MAX_ARG_LENGTH: 181 args.append(f"{self.__INDENT_ARG}{arg}") 182 else: 183 sub_args = arg.split(" ") 184 args.extend(self.__pretty_sub_arg(sub_args)) 185 return '\n'.join(args) 186 187 def __pretty_sub_arg(self, sub_args: list[str]) -> list[str]: 188 acc = '' 189 result: list[str] = [] 190 for sub_arg in sub_args: 191 if len(sub_arg) > self.__MIN_ARG_LENGTH: 192 if len(acc) > 0: 193 result.append(f"{self.__INDENT_SUB_ARG}{acc}") 194 acc = "" 195 result.append(f"{self.__INDENT_SUB_ARG}{sub_arg}") 196 else: 197 acc += sub_arg + " " 198 if len(acc) > 0: 199 result.append(f"{self.__INDENT_SUB_ARG}{acc}") 200 return result 201 202 def __str_process_env(self, indent: int) -> list[str]: 203 result = [f"{utils.indent(indent)}env:"] 204 for key in self.env: 205 env_value: str | list[str] = self.env[key] 206 if isinstance(env_value, list): 207 result += [f"{utils.indent(indent + 1)}{key}:"] 208 for item in env_value: 209 result += [f"{utils.indent(indent + 2)}- {item}"] 210 else: 211 result += [f"{utils.indent(indent + 1)}{key}: {env_value}"] 212 return result 213 214 def __str_process_args(self, indent: int) -> list[str]: 215 result = [f"{utils.indent(indent)}args:"] 216 for arg in self.args: 217 if arg.strip(): 218 result += [f"{utils.indent(indent + 1)}{arg}"] 219 return result 220