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.coordinator import Coordinator from pytest_abra.env_manager import 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] = [] 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_test_list(self.setups) def run_tests(self): """runs the test scripts if available""" self._execute_test_list(self.tests) def run_cleanups(self): """runs the cleanup scripts if available""" self._execute_test_list(self.cleanups) def _execute_test_list(self, test_list: list[Test]): """runs the main test script and if available and sub test scripts if their running condition is met""" # check if required dependencies have passed if not self._dependencies_passed(): logger.warning(f"skipping run_tests() of {self.env_type}, because some dependencies have not passed") return for test in test_list: self._run_test_with_checks(test) def _run_test_with_checks(self, test: Test): # dependency passed: true / false # already_passed: true / false # prevent_skip: true / false # condition_available: true / pass # condition_met: true / false identifier_string = self.combine_names(self.env_type, test.test_file) results = list(self.DIR.RECIPES.rglob(test.test_file)) assert len(results) == 1, f"{test.test_file} should exist exactly 1 time, but found {len(results)} times" full_test_path = results[0] # check if test aleady passed if self._is_test_passed(identifier_string, remove_existing=True): if test.prevent_skip: logger.info(f"continuing , test {identifier_string} has passed but prevent_skip=True") else: logger.info(f"skipping {identifier_string}, test has passed") return if test.condition: condition_result = self._run_condition(test.condition) if not condition_result: # test condition is defined but not met logger.info(f"skipping {identifier_string}, test condition is not met") return # test condition is undefined or not met logger.info(f"running {identifier_string}") result = self._call_pytest(full_test_path) self._create_result_file(result=result, identifier_string=identifier_string) def _run_condition(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) def _is_test_passed(self, identifier_string: str, remove_existing: bool = False) -> bool: """returns True if the selected test matching identifier_string already passed This is determined by the presence of a specific output file in the RESULTS folder that matches identifier_string remove_existing: If True, result files matching identifier_string with a status other than 'passed' will be deleted""" already_passed = False for result in self.DIR.RESULTS.glob("*"): if identifier_string in result.name: # process any result file (passed / failed / skipped) if it exists if "passed" in result.name: already_passed = True elif remove_existing: result.unlink() return already_passed def _call_pytest(self, full_test_path: Path) -> int: """runs pytest programmatically on 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("-rx") 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.RECORDS / "traces" / full_test_path.stem)) # tracing command_arguments.append("--tracing") command_arguments.append("retain-on-failure") # command_arguments.append("on") # 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") # headed # command_arguments.append("--headed") # html report. Will be combined into one file later. command_arguments.append(f"--html={self.DIR.RECORDS / 'html' / full_test_path.with_suffix('.html').name}") return pytest.main(command_arguments) def _create_result_file( self, result: int, identifier_string: str, ): """create result file to indicated passed/failed or skipped test""" full_name = self.combine_names(self.result_int_to_str(result), identifier_string) file_path = self.DIR.RESULTS / full_name with open(file_path, "w") as _: pass # create empty file 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.RESULTS.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 result_int_to_str(result_int: int) -> str: """converts the pytest exit code (int) into a meaningful string""" match result_int: case -1: return "skipped" case 0: return "passed" case _: return "failed" @staticmethod def combine_names(*names: str) -> str: return "-".join(names)