216 lines
8 KiB
Python
216 lines
8 KiB
Python
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(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") # "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.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)
|