• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 multiprocessing
20import os
21from os import path
22from typing import Union, List, Tuple, Any, Callable, Optional, Set, Type
23
24from runner.enum_types.qemu import QemuKind
25from runner.logger import Log
26from runner.options.cli_args_wrapper import CliArgsWrapper
27from runner.options.config_keeper import ConfigKeeper
28from runner.utils import enum_from_str, is_type_of, EnumT
29
30CliOptionType = Union[str, List[str]]
31CastToTypeFunction = Callable[[Any], Any]
32
33_LOGGER = logging.getLogger("runner.options.decorator_value")
34
35
36def value(
37        *,
38        yaml_path: str,
39        cli_name: CliOptionType,
40        required: bool = False,
41        cast_to_type: Optional[CastToTypeFunction] = None
42) -> Any:
43    def decorator(func: Any) -> Any:
44        def decorated(*args: Any, **kwargs: Any) -> Any:
45            cli = _process_cli_option(cli_name, cast_to_type)
46            if cli is not None:
47                return cli
48            yaml = ConfigKeeper.get().get_value_by_path(yaml_path)
49            yaml = cast_to_type(yaml) if cast_to_type is not None and yaml is not None else yaml
50            if yaml is not None:
51                return yaml
52            default_value = func(*args, **kwargs)
53            if default_value is None and required:
54                Log.exception_and_raise(
55                    _LOGGER,
56                    f"Missed required property. "
57                    f"Expected either '{yaml_path}' in the config file "
58                    f"or '{cli_name}' cli option(s)"
59                )
60            return default_value
61
62        return decorated
63
64    return decorator
65
66
67def _process_cli_option(
68        cli_name: CliOptionType,
69        cast_to_type: Optional[CastToTypeFunction] = None
70) -> Any:
71    cli = None
72    if isinstance(cli_name, str):
73        cli = CliArgsWrapper.get_by_name(cli_name)
74        cli = cast_to_type(cli) if cast_to_type is not None and cli is not None else cli
75    elif isinstance(cli_name, list):
76        clis = [(n, CliArgsWrapper.get_by_name(n)) for n in cli_name]
77        if cast_to_type is None:
78            Log.exception_and_raise(
79                _LOGGER,
80                f"Cannot convert {cli_name} from command line. Provide 'cast_to_type' parameter"
81            )
82        cli = cast_to_type(clis)
83    else:
84        Log.exception_and_raise(_LOGGER, f"Cannot process CLI names {cli_name}")
85    return cli
86
87
88def _to_qemu(names: Union[str, List[Tuple[str, bool]], None]) -> Optional[QemuKind]:
89    if names is None:
90        return None
91    if isinstance(names, str):
92        return enum_from_str(names, QemuKind)
93    result = [n for n in names if n[1] is not None]
94    if len(result) == 0:
95        return None
96    if result[0][0] == "arm64_qemu":
97        return QemuKind.ARM64
98    if result[0][0] == "arm32_qemu":
99        return QemuKind.ARM32
100    return None
101
102
103TestSuiteFromCliValue = Optional[Union[bool, List[str]]]
104TestSuiteFromCliKey = str
105TestSuitesFromCliTuple = Tuple[TestSuiteFromCliKey, TestSuiteFromCliValue]
106
107
108def _to_test_suites(names: Optional[List[Union[str, TestSuitesFromCliTuple]]]) -> Optional[Set[str]]:
109    if names is None:
110        return None
111    suites: Set[str] = set([])
112    for name in names:
113        if isinstance(name, str):
114            # from yaml: simple list of strings
115            suites.add(name)
116        else:
117            # from cli: name has type TestSuitesFromCliTuple
118            name_key: TestSuiteFromCliKey = name[0]
119            name_value: TestSuiteFromCliValue = name[1]
120            if name_value is not None and isinstance(name_value, list):
121                suites.update(name_value)
122            elif name_value:
123                suites.add(name_key)
124    return suites if len(suites) > 0 else None
125
126
127def _to_bool(cli_value: Union[str, bool, None]) -> Optional[bool]:
128    if cli_value is None:
129        return None
130    if isinstance(cli_value, bool):
131        return cli_value
132    return cli_value.lower() == "true"
133
134
135def _to_int(cli_value: Union[str, int, None]) -> Optional[int]:
136    if cli_value is None:
137        return None
138    if isinstance(cli_value, int):
139        return cli_value
140    return int(cli_value)
141
142
143def _to_processes(cli_value: Union[str, int, None]) -> Optional[int]:
144    if cli_value is None:
145        return None
146    if isinstance(cli_value, str) and cli_value.lower() == "all":
147        return multiprocessing.cpu_count()
148    return _to_int(cli_value)
149
150
151def _to_jit_preheats(cli_value: Union[str, int, None], *, prop: str, default_if_empty: int) -> Optional[int]:
152    if cli_value is None:
153        return None
154    if isinstance(cli_value, int):
155        return cli_value
156    if cli_value == "":
157        return default_if_empty
158    parts = [p.strip() for p in cli_value.split(",") if p.strip().startswith(prop)]
159    if len(parts) == 0:
160        return None
161    prop_parts = [p.strip() for p in parts[0].split("=")]
162    if len(prop_parts) != 2:
163        Log.exception_and_raise(_LOGGER, f"Incorrect value for option --jit-preheat-repeats '{cli_value}'")
164    return int(prop_parts[1])
165
166
167def _to_time_edges(cli_value: Union[str, List[int], None]) -> Optional[List[int]]:
168    if cli_value is None:
169        return None
170    if isinstance(cli_value, str):
171        return [int(n.strip()) for n in cli_value.split(",")]
172    return cli_value
173
174
175def _to_enum(cli_value: Union[str, EnumT, None], enum_cls: Type[EnumT]) -> Optional[EnumT]:
176    if isinstance(cli_value, str):
177        return enum_from_str(cli_value, enum_cls)
178    return cli_value
179
180
181def _to_str(obj: Any, indent: int = 0) -> str:
182    attrs = dir(obj)
183    attrs = [n for n in attrs if not n.startswith("_") and not is_type_of(getattr(obj, n), "method")]
184    result = [f"{obj.__class__.__name__}"]
185    indent_str = "\n" + "\t" * (indent + 1)
186    result += [f"{indent_str}{attr}: {getattr(obj, attr)}" for attr in attrs]
187    return "".join(result)
188
189
190def _to_path(cli_value: Optional[str]) -> Optional[str]:
191    if cli_value is None:
192        return cli_value
193    return path.abspath(path.expanduser(cli_value))
194
195
196def _to_dir(cli_value: Optional[str]) -> Optional[str]:
197    abspath = _to_path(cli_value)
198    if abspath is None:
199        return None
200    os.makedirs(abspath, exist_ok=True)
201    return abspath
202