• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
18import os
19import sys
20import traceback
21from datetime import datetime, timedelta
22from pathlib import Path
23from typing import Any, cast
24
25import pytz
26
27from runner.common_exceptions import InvalidConfiguration, RunnerException
28from runner.enum_types.verbose_format import VerboseKind
29from runner.environment import RunnerEnv
30from runner.init_runner import InitRunner
31from runner.logger import Log
32from runner.options.cli_options import get_args
33from runner.options.config import Config
34from runner.runner_base import Runner
35from runner.suites.runner_standard_flow import RunnerStandardFlow
36from runner.utils import pretty_divider
37
38
39def main() -> None:
40    init_runner = InitRunner()
41    if init_runner.should_runner_initialize(sys.argv):
42        init_runner.initialize(RunnerEnv.get_mandatory_props())
43        sys.exit(0)
44    local_env = Path(__file__).with_name(".env")
45    urunner_path = Path(__file__).parent
46    RunnerEnv(local_env=local_env, urunner_path=urunner_path).load_environment()
47
48    args = get_args()
49    logger = load_config(args)
50    config = Config(args)
51
52    config.workflow.check_binary_artifacts()
53    config.workflow.check_types()
54    logger.summary(f"Loaded configuration: {config}")
55
56    if config.general.processes == 1:
57        Log.default(logger, "Attention: tests are going to take only 1 process. The execution can be slow. "
58                            "You can set the option `--processes` to wished processes quantity "
59                            "or use special value `all` to use all available cores.")
60    failed_tests = 0
61    try:
62        failed_tests = main_cycle(config, logger)
63    except RunnerException:
64        logger.logger.critical(traceback.format_exc())
65    finally:
66        sys.exit(0 if failed_tests == 0 else 1)
67
68
69def main_cycle(config: Config, logger: Log) -> int:
70    start = datetime.now(pytz.UTC)
71    runner = RunnerStandardFlow(config)
72
73    failed_tests = 0
74
75    if config.test_suite.repeats_by_time == 0:
76        for repeat in range(1, config.test_suite.repeats + 1):
77            repeat_str = f"Run #{repeat} of {config.test_suite.repeats}"
78            failed_tests += launch_runners(runner, logger, config, repeat, repeat_str)
79    else:
80        before = datetime.now(pytz.UTC)
81        current = before
82        end = current + timedelta(seconds=float(config.test_suite.repeats_by_time))
83        repeat = 1
84        delta: float = 0.0
85        while current < end:
86            remains = round(config.test_suite.repeats_by_time - delta, 1)
87            repeat_str = (f"Run #{repeat} for {config.test_suite.repeats_by_time} sec. "
88                          f"Remains {remains} sec")
89            failed_tests += launch_runners(runner, logger, config, repeat, repeat_str)
90            repeat += 1
91            current = datetime.now(pytz.UTC)
92            delta = round((current - before).total_seconds(), 1)
93
94    finish = datetime.now(pytz.UTC)
95    Log.default(logger, f"Runner has been working for {round((finish - start).total_seconds())} sec")
96
97    return failed_tests
98
99
100def launch_runners(runner: Runner, logger: Log, config: Config, repeat: int, repeat_str: str) -> int:
101    failed_tests = 0
102    Log.all(logger, f"{repeat_str}: Runner {runner.name} started")
103    runner.before_suite()
104    runner.run_threads(repeat)
105    runner.after_suite()
106    Log.all(logger, f"{repeat_str}: Runner {runner.name} finished")
107    Log.all(logger, pretty_divider())
108    failed_tests += runner.summarize()
109    Log.default(logger, f"{repeat_str}: Runner {runner.name}: {failed_tests} failed tests")
110    if config.general.coverage.use_llvm_cov:
111        runner.create_coverage_html()
112    return failed_tests
113
114
115def load_config(args: dict[str, Any]) -> Log: # type: ignore[explicit-any]
116    runner_verbose = "runner.verbose"
117    test_suite_const = "test-suite"
118
119    verbose = args[runner_verbose] if runner_verbose in args else VerboseKind.SILENT
120    if test_suite_const not in args:
121        raise InvalidConfiguration(f"Incorrect configuration: cannot file element '{test_suite_const}'")
122    test_suite = args[test_suite_const]
123    work_dir = Path(cast(str, os.getenv("WORK_DIR")), test_suite)
124
125    return Log.setup(verbose, work_dir)
126
127
128if __name__ == "__main__":
129    main()
130