HAL9000
bd955696f9
Address all 4 remaining blocking issues from PR #7811 review: - Delete duplicate _validation_pipeline_mock.py in features/steps/ (the inline MockValidationExecutor is kept where it belongs) - Replace # type: ignore[return-value] with cast(dict[str, Any], ...) to satisfy the zero-type-suppressions policy for test additions - Add @tdd_issue_7623 regression tag on concurrency scenario - Remove unreachable dead-code RuntimeError guard in _install_thread_local_streams() (constructors never return None) All CI lint checks now pass. Source files remain within line limits.
115 lines
3.9 KiB
Python
115 lines
3.9 KiB
Python
"""Mock validation executor for testing pipelines.
|
|
|
|
This class belongs in ``features/mocks/``, not ``features/steps/``, per
|
|
the mock placement rule documented in the CONTRIBUTING guide.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import time
|
|
from typing import Any, cast
|
|
|
|
|
|
class MockValidationExecutor:
|
|
"""Configurable mock executor for testing the validation pipeline."""
|
|
|
|
def __init__(self) -> None:
|
|
self._results: dict[str, dict[str, Any]] = {}
|
|
self._timeout_names: set[str] = set()
|
|
self._exception_names: dict[str, Exception] = {}
|
|
self._non_dict_names: set[str] = set()
|
|
self._missing_passed_names: set[str] = set()
|
|
self._default_passed: bool = False
|
|
self._stdout_names: set[str] = set()
|
|
self._stderr_names: set[str] = set()
|
|
|
|
def set_passed(self, name: str) -> None:
|
|
self._results[name] = {"passed": True, "message": f"{name} passed"}
|
|
|
|
def set_passed_with_data(self, name: str) -> None:
|
|
self._results[name] = {
|
|
"passed": True,
|
|
"message": f"{name} passed with data",
|
|
"data": {"detail": "some-value", "count": 42},
|
|
}
|
|
|
|
def set_failed(self, name: str, message: str) -> None:
|
|
self._results[name] = {"passed": False, "message": message}
|
|
|
|
def set_timeout(self, name: str) -> None:
|
|
self._timeout_names.add(name)
|
|
|
|
def set_exception(self, name: str, exc: Exception) -> None:
|
|
self._exception_names[name] = exc
|
|
|
|
def set_non_dict(self, name: str) -> None:
|
|
self._non_dict_names.add(name)
|
|
|
|
def set_missing_passed(self, name: str) -> None:
|
|
self._missing_passed_names.add(name)
|
|
|
|
def set_default_passed(self) -> None:
|
|
self._default_passed = True
|
|
|
|
def set_non_bool_passed(self, name: str) -> None:
|
|
self._results[name] = {"passed": "yes", "message": f"{name} passed"}
|
|
|
|
def set_non_string_message(self, name: str) -> None:
|
|
self._results[name] = {"passed": True, "message": 12345}
|
|
|
|
def set_non_dict_data(self, name: str) -> None:
|
|
self._results[name] = {
|
|
"passed": True,
|
|
"message": f"{name} ok",
|
|
"data": [1, 2],
|
|
}
|
|
|
|
def set_prints_stdout(self, name: str) -> None:
|
|
self._results[name] = {"passed": True, "message": f"{name} ok"}
|
|
self._stdout_names.add(name)
|
|
|
|
def set_prints_stderr(self, name: str) -> None:
|
|
self._results[name] = {"passed": True, "message": f"{name} ok"}
|
|
self._stderr_names.add(name)
|
|
|
|
def __call__(
|
|
self, validation_name: str, arguments: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
if validation_name in self._timeout_names:
|
|
# Sleep longer than the test timeout (0.2 s) but not excessively.
|
|
_real_sleep = getattr(time, "_original_sleep", time.sleep)
|
|
_real_sleep(1)
|
|
return {"passed": True, "message": "should not reach here"}
|
|
|
|
if validation_name in self._exception_names:
|
|
raise self._exception_names[validation_name]
|
|
|
|
if validation_name in self._non_dict_names:
|
|
# Cast needed because the mock intentionally returns non-dict
|
|
# output to exercise _normalise_executor_output normalisation.
|
|
return cast(dict[str, Any], "not-a-dict")
|
|
|
|
if validation_name in self._missing_passed_names:
|
|
return {"message": "no passed key here"}
|
|
|
|
if validation_name in self._stdout_names:
|
|
print(f"stdout from {validation_name}")
|
|
|
|
if validation_name in self._stderr_names:
|
|
print(f"stderr from {validation_name}", file=sys.stderr)
|
|
|
|
if validation_name in self._results:
|
|
return self._results[validation_name]
|
|
|
|
if self._default_passed:
|
|
return {
|
|
"passed": True,
|
|
"message": f"{validation_name} passed (default)",
|
|
}
|
|
|
|
return {
|
|
"passed": False,
|
|
"message": f"{validation_name}: no result configured",
|
|
}
|