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