jesseandClaude Opus 5.5
537beb2934
A task left running wrote its outcome into the next test's cache directory, so the next test saw a remembered failure and never called the gateway. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
104 lines
3.4 KiB
Python
104 lines
3.4 KiB
Python
"""Pytest import guardrails for local project modules."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
# Deliberately fictional birth data; not a real person or an oracle golden.
|
|
FIXTURE_BIRTH_DATE = "1990-06-15"
|
|
FIXTURE_BIRTH_TIME = "10:20"
|
|
FIXTURE_BIRTH_LAT = 31.2
|
|
FIXTURE_BIRTH_LON = 121.5
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPTS = str(ROOT / "scripts")
|
|
WORKBUDDY_SKILL_SCRIPTS = ".workbuddy/skills/jyotish-vedic-astrology/scripts"
|
|
SLOW_API_SECURITY_PREFIXES = (
|
|
"test_vedastro_",
|
|
"test_high_rigor_",
|
|
"test_professional_reading",
|
|
"test_api_prompt_pack",
|
|
"test_consultation_workflow",
|
|
"test_thematic_report",
|
|
"test_capability_audit",
|
|
"test_technique_catalog",
|
|
)
|
|
|
|
|
|
def ensure_project_scripts_first() -> None:
|
|
if SCRIPTS in sys.path:
|
|
sys.path.remove(SCRIPTS)
|
|
sys.path.insert(0, SCRIPTS)
|
|
|
|
|
|
def remove_workbuddy_shadow_modules() -> None:
|
|
for name, module in list(sys.modules.items()):
|
|
module_file = getattr(module, "__file__", "") or ""
|
|
if WORKBUDDY_SKILL_SCRIPTS in module_file:
|
|
del sys.modules[name]
|
|
|
|
|
|
def pytest_runtest_setup() -> None:
|
|
remove_workbuddy_shadow_modules()
|
|
ensure_project_scripts_first()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_vedastro_snapshot_cache(tmp_path_factory, monkeypatch) -> None:
|
|
cache_dir = tmp_path_factory.mktemp("vedastro_snapshot_cache")
|
|
monkeypatch.setenv("JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR", str(cache_dir))
|
|
# Foreground failures are remembered for the day there too (issue #130).
|
|
negative_dir = tmp_path_factory.mktemp("vedastro_snapshot_negative_cache")
|
|
monkeypatch.setenv("JYOTISH_VEDASTRO_SNAPSHOT_NEGATIVE_CACHE_DIR", str(negative_dir))
|
|
from scripts.rectification.vedastro_validate_cache import clear as clear_validate_memory
|
|
|
|
clear_validate_memory()
|
|
|
|
|
|
_FOREGROUND_EXECUTOR_MODULES = (
|
|
"scripts.vedastro_foreground",
|
|
"vedastro_foreground",
|
|
"scripts.jyotish_api_server",
|
|
"jyotish_api_server",
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_foreground_vedastro_executor(monkeypatch):
|
|
"""A fresh foreground pool per test, drained at teardown (BUG-1498).
|
|
|
|
A task left running stores its outcome under whatever cache directory the
|
|
environment names when it finishes, which would be the next test's.
|
|
"""
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
from scripts import vedastro_foreground
|
|
|
|
executor = ThreadPoolExecutor(
|
|
max_workers=vedastro_foreground._FOREGROUND_VEDASTRO_WORKERS,
|
|
thread_name_prefix="jyotish-vedastro-fg-test",
|
|
)
|
|
for name in _FOREGROUND_EXECUTOR_MODULES:
|
|
module = sys.modules.get(name)
|
|
if module is not None and hasattr(module, "_FOREGROUND_VEDASTRO_EXECUTOR"):
|
|
monkeypatch.setattr(module, "_FOREGROUND_VEDASTRO_EXECUTOR", executor)
|
|
yield
|
|
# Queued work is dropped; running work gets a bounded wait so a stuck stub cannot hang the run.
|
|
executor.shutdown(wait=False, cancel_futures=True)
|
|
deadline = time.monotonic() + 10
|
|
for thread in list(getattr(executor, "_threads", ())):
|
|
thread.join(max(deadline - time.monotonic(), 0))
|
|
|
|
|
|
def pytest_collection_modifyitems(items) -> None:
|
|
for item in items:
|
|
if item.fspath.basename == "test_api_server_security.py" and item.name.startswith(SLOW_API_SECURITY_PREFIXES):
|
|
item.add_marker("slow")
|
|
|
|
|
|
ensure_project_scripts_first()
|