import os from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Callable, NamedTuple import pytest from loguru import logger if TYPE_CHECKING: from pytest_abra import Coordinator, DirManager, EnvFile class ConditionArgs(NamedTuple): env_config: dict[str, str] runner_index: int env_files: list["EnvFile"] @dataclass class Test: test_file: str condition: Callable[[ConditionArgs], bool] | None = None prevent_skip: bool = False class Runner: env_type: str = "" setups: list[Test] = [] tests: list[Test] = [] cleanups: list[Test] = [] dependencies: list[str] = [] _tests_path: Path = Path("undefined") def __init__(self, coordinator: "Coordinator", runner_index: int): self.coordinator = coordinator self.runner_index = runner_index self.DIR = coordinator.DIR self.ENV = coordinator.ENV self.RUNNER_DICT = coordinator.RUNNER_DICT logger.info(f"creating instance of {self.__class__.__name__}") def run_setups(self): """runs the setup scripts if available""" self._execute_tests_list(self.setups) def run_tests(self): """runs the test scripts if available""" self._execute_tests_list(self.tests) def run_cleanups(self): """runs the cleanup scripts if available""" self._execute_tests_list(self.cleanups) def _execute_tests_list(self, test_list: list[Test]): """Runs all tests given in the list. If condition is defined, it is also checked.""" # check if required dependencies have passed if not self._dependencies_passed(): logger.warning(f"skipping run_tests() of {self.env_type} (one or more dependencies have not passed)") return for test in test_list: self._run_test_with_checks(test) def _run_test_with_checks(self, test: Test): identifier_string = self.combine_names(self.env_type, test.test_file) test_files = list(self._tests_path.rglob(test.test_file)) assert len(test_files) == 1, f"{test.test_file} should exist exactly once, but found {len(test_files)} times" full_test_path = test_files[0] # check if test aleady passed if self._is_test_passed(self.DIR, identifier_string): if test.prevent_skip: logger.info(f"continuing {identifier_string} (passed before but prevent_skip=True)") else: logger.info(f"skipping {identifier_string} (test has passed)") return if test.condition: condition_result = self._call_condition_function(test.condition) if not condition_result: # test condition is defined but not met logger.info(f"skipping {identifier_string} (test condition is not met)") self._create_status_file(self.DIR, status="skipped", identifier_string=identifier_string) return # test condition is undefined or not met logger.info(f"running {identifier_string}") exit_code = self._call_pytest(full_test_path) self._create_status_file(self.DIR, status=exit_code, identifier_string=identifier_string) def _call_condition_function(self, condition_function: Callable[[ConditionArgs], bool]): """run the test condition function with multiple arguments""" # more arguments can be added later without changing the function signature conditon_args = ConditionArgs( env_files=self.ENV.env_files, runner_index=self.runner_index, env_config=self.ENV.env_files[self.runner_index].env_config, ) return condition_function(conditon_args) @classmethod def _create_status_file( cls, DIR: "DirManager", status: int | str, identifier_string: str, ): """create result file to indicated passed/failed/skipped test""" if isinstance(status, int): status = cls.exit_code_to_str(status) # remove matching files for status_file in cls._get_status_files(DIR, identifier_string): status_file.unlink() full_name = cls.combine_names(status, identifier_string) file_path = DIR.STATUS / full_name with open(file_path, "w") as _: pass # create empty file @staticmethod def _get_status_files(DIR: "DirManager", identifier_string: str) -> list[Path]: return [f for f in DIR.STATUS.glob("*") if identifier_string in f.name] @classmethod def _is_test_passed(cls, DIR: "DirManager", identifier_string: str) -> bool: """returns True if the selected test matching identifier_string already passed""" matching_files = cls._get_status_files(DIR, identifier_string) if len(matching_files) == 1: status_file = matching_files[0] if "passed" in status_file.name: return True elif len(matching_files) > 1: logger.warning("more than one matching status file found") return False def _call_pytest(self, full_test_path: Path) -> int: """runs pytest programmatically with a specific file all tests in the file [full_test_path] will be run along with command line arguments""" command_arguments = [] # command_arguments.append("--traceconfig") command_arguments.append("-v") command_arguments.append(str(full_test_path)) command_arguments.append("--runner_index") command_arguments.append(str(self.runner_index)) # set root dir for tests output (used in DirManager). this is our custom argument command_arguments.append("--output_dir") command_arguments.append(str(self.DIR.OUTPUT_DIR)) command_arguments.append("--session_id") command_arguments.append(self.DIR.session_id) command_arguments.append("--timeout") command_arguments.append(str(self.coordinator.TIMEOUT)) # artifacts dir from pytest # warning: https://github.com/microsoft/playwright-pytest/issues/111 # --output only works with the given context and page fixture # folder needs to be unique! traces will not appear, if every pytest run has same output dir command_arguments.append("--output") command_arguments.append(str(self.DIR.RESULTS / "traces" / full_test_path.stem)) # tracing command_arguments.append("--tracing") # "on", "off", "retain-on-failure" command_arguments.append("retain-on-failure") # Disable capturing. With -s set, prints will go to console as if pytest is not there. if os.environ.get("PWDEBUG") == "1": command_arguments.append("-s") command_arguments.append("-s") # headed # command_arguments.append("--headed") # html report. Will be combined into one file later. command_arguments.append(f"--html={self.DIR.RESULTS / 'html' / full_test_path.with_suffix('.html').name}") return pytest.main(command_arguments) def _dependencies_passed(self): """returns true if all setups of each dependency have passed""" # todo: what about conditional setups? passed_tests = [r.name for r in self.DIR.STATUS.glob("*") if "passed" in r.name] results = [] for dependency in self.dependencies: dependency_runner = self.coordinator.RUNNER_DICT[dependency] for setup_name in dependency_runner.setups: dependencie_identifier = self.combine_names(dependency_runner.env_type, setup_name.test_file) results.append(any(dependencie_identifier in f for f in passed_tests)) return all(results) @staticmethod def exit_code_to_str(result_int: int) -> str: """converts the pytest exit code (int) into a meaningful string""" match result_int: case 0: return "passed" case _: return "failed" @staticmethod def combine_names(*names: str) -> str: return "-".join(names)