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 abc import ABC, abstractmethod 19from concurrent.futures import ThreadPoolExecutor 20from datetime import datetime 21from pathlib import Path 22 23import pytz 24from tqdm import tqdm 25 26from runner.logger import Log 27from runner.options.config import Config 28from runner.test_base import Test 29from runner.utils import correct_path 30 31_LOGGER = Log.get_logger(__file__) 32 33 34def run_test(pair: tuple[Test, int]) -> Test: 35 return pair[0].run(pair[1]) 36 37 38class Runner(ABC): 39 def __init__(self, config: Config, name: str) -> None: 40 # Roots: 41 # directory where test files are located - it's either set explicitly to the absolute value 42 # or the current folder (where this python file is located!) parent 43 self.test_root: Path | None = None 44 # directory where list files (files with list of ignored, excluded, and other tests) are located 45 # it's either set explicitly to the absolute value or 46 # the current folder (where this python file is located!) parent 47 self.list_root: Path | None = config.test_suite.list_root 48 if self.list_root is not None: 49 _LOGGER.summary(f"LIST_ROOT set to {self.list_root}") 50 51 # runner init time 52 self.start_time = datetime.now(pytz.UTC) 53 54 self.config = config 55 self.name: str = name 56 57 # Lists: 58 # excluded test is a test what should not be loaded and should be tried to run 59 # excluded_list: either absolute path or path relative from list_root to the file with the list of such tests 60 self.excluded_lists: list[str] = [] 61 self.excluded_tests: set[str] = set([]) 62 # ignored test is a test what should be loaded and executed, but its failure should be ignored 63 # ignored_list: either absolute path or path relative from list_root to the file with the list of such tests 64 # aka: kfl stands for `known failures list` 65 self.ignored_lists: list[str] = [] 66 self.ignored_tests: set[str] = set([]) 67 # list of file names, each is a name of a test. Every test should be executed 68 # So, it can contain ignored tests, but cannot contain excluded tests 69 self.tests: set[Test] = set([]) 70 # list of results of every executed test 71 self.results: list[Test] = [] 72 # name of file with a list of only tests what should be executed 73 # if it's specified other tests are not executed 74 self.explicit_list = correct_path(self.list_root, config.test_suite.test_lists.explicit_list) \ 75 if config.test_suite.test_lists.explicit_list is not None and self.list_root is not None \ 76 else None 77 # name of the single test file in form of a relative path from test_root what should be executed 78 # if it's specified other tests are not executed even if test_list is set 79 self.explicit_test = config.test_suite.test_lists.explicit_file 80 81 # Counters: 82 # failed + ignored + passed + excluded_after = len of all executed tests 83 # failed + ignored + passed + excluded_after + excluded = len of full set of tests 84 self.failed = 0 85 self.ignored = 0 86 self.passed = 0 87 self.excluded = 0 88 # Test chosen to execute can detect itself as excluded one 89 self.excluded_after = 0 90 self.update_excluded = config.test_suite.test_lists.update_excluded 91 92 @staticmethod 93 def add_tests_param(tests: set[Test], repeat: int) -> list[tuple[Test, int]]: 94 return [(test, repeat) for test in tests] 95 96 @abstractmethod 97 def summarize(self) -> int: 98 pass 99 100 @abstractmethod 101 def create_coverage_html(self) -> None: 102 pass 103 104 def run_threads(self, repeat: int) -> None: 105 _LOGGER.all(f"--Start test running with {self.config.general.processes} threads") 106 with ThreadPoolExecutor(self.config.general.processes) as executor: 107 results = executor.map( 108 run_test, 109 self.add_tests_param(self.tests, repeat)) 110 if self.config.general.show_progress: 111 results = tqdm(results, total=len(self.tests)) 112 self.results = list(results) 113 114 def before_suite(self) -> None: 115 pass 116 117 def after_suite(self) -> None: 118 pass 119