142 lines
4.4 KiB
Python
142 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import pkgutil
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from types import ModuleType
|
|
from typing import Any, Callable
|
|
|
|
from .agent_process import AgentProcess
|
|
from .didactyl_client import DidactylClient, DidactylTimeoutError
|
|
from .log_watcher import LogWatcher
|
|
|
|
|
|
class SkipTest(Exception):
|
|
pass
|
|
|
|
|
|
TestFn = Callable[["TestContext"], tuple[bool, str, dict[str, Any]]]
|
|
|
|
|
|
@dataclass
|
|
class TestCase:
|
|
suite: str
|
|
name: str
|
|
description: str
|
|
fn: TestFn
|
|
requires_restart: bool = False
|
|
|
|
|
|
@dataclass
|
|
class TestResult:
|
|
suite: str
|
|
name: str
|
|
status: str
|
|
message: str
|
|
duration_seconds: float
|
|
agent_errors: list[str] = field(default_factory=list)
|
|
details: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class TestContext:
|
|
client: DidactylClient
|
|
agent: AgentProcess
|
|
log: LogWatcher
|
|
args: Any
|
|
|
|
|
|
class TestRunner:
|
|
def __init__(self, agent: AgentProcess, client: DidactylClient, log: LogWatcher, args: Any) -> None:
|
|
self.agent = agent
|
|
self.client = client
|
|
self.log = log
|
|
self.args = args
|
|
|
|
def discover_suites(self, package: str = "tests.suites") -> list[TestCase]:
|
|
mod = importlib.import_module(package)
|
|
tests: list[TestCase] = []
|
|
for info in pkgutil.iter_modules(mod.__path__):
|
|
if not info.name.startswith("test_"):
|
|
continue
|
|
if self.args.suite and info.name not in self.args.suite:
|
|
continue
|
|
module = importlib.import_module(f"{package}.{info.name}")
|
|
tests.extend(self._tests_from_module(module))
|
|
return tests
|
|
|
|
def _tests_from_module(self, module: ModuleType) -> list[TestCase]:
|
|
if not hasattr(module, "get_tests"):
|
|
return []
|
|
suite_tests = module.get_tests()
|
|
out: list[TestCase] = []
|
|
for t in suite_tests:
|
|
if self.args.test and t.name not in self.args.test:
|
|
continue
|
|
out.append(t)
|
|
return out
|
|
|
|
def run_all(self, tests: list[TestCase]) -> list[TestResult]:
|
|
results: list[TestResult] = []
|
|
ctx = TestContext(client=self.client, agent=self.agent, log=self.log, args=self.args)
|
|
|
|
for tc in tests:
|
|
marker = f"{tc.suite}.{tc.name}.{int(time.time() * 1000)}"
|
|
if tc.requires_restart:
|
|
self.agent.restart(timeout=30)
|
|
if not self.agent.is_alive():
|
|
ok = self.agent.restart(timeout=30)
|
|
if not ok:
|
|
results.append(
|
|
TestResult(
|
|
suite=tc.suite,
|
|
name=tc.name,
|
|
status="error",
|
|
message="Agent not alive and restart failed",
|
|
duration_seconds=0.0,
|
|
)
|
|
)
|
|
continue
|
|
|
|
self.log.set_marker(marker)
|
|
start = time.monotonic()
|
|
try:
|
|
passed, message, details = tc.fn(ctx)
|
|
status = "pass" if passed else "fail"
|
|
except SkipTest as e:
|
|
status = "skip"
|
|
message = str(e)
|
|
details = {}
|
|
except DidactylTimeoutError as e:
|
|
status = "timeout"
|
|
message = str(e)
|
|
details = {}
|
|
if not getattr(self.args, "no_restart", False):
|
|
self.agent.restart(timeout=30)
|
|
except Exception as e:
|
|
status = "error"
|
|
message = repr(e)
|
|
details = {}
|
|
if not self.agent.is_alive() and not getattr(self.args, "no_restart", False):
|
|
self.agent.restart(timeout=30)
|
|
|
|
duration = time.monotonic() - start
|
|
agent_errors = self.log.error_lines(marker)
|
|
results.append(
|
|
TestResult(
|
|
suite=tc.suite,
|
|
name=tc.name,
|
|
status=status,
|
|
message=message,
|
|
duration_seconds=duration,
|
|
agent_errors=agent_errors,
|
|
details=details,
|
|
)
|
|
)
|
|
|
|
if getattr(self.args, "verbose", False):
|
|
print(f"[{status.upper()}] {tc.suite}::{tc.name} - {message}")
|
|
|
|
return results
|