File size: 2,269 Bytes
905b846 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | """Shared subprocess driver for tests/test_eval_*.py.
Phase 1.6 introduced three test files that spawn a fresh Python
subprocess per region — necessary because `src.pipeline` (and its
transitive imports) capture `from config import REGION, MANDIS, ...`
at module load time, so reloading config doesn't retroactively
update importers.
All three files shared the same env + sentinel + JSON-parse machinery;
this module factors it out so each test contributes only its driver
body (the code that actually runs inside the subprocess).
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
_REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
_SENTINEL_BEGIN = "__PIPELINE_RESULT_BEGIN__"
_SENTINEL_END = "__PIPELINE_RESULT_END__"
_REPO_ROOT_PLACEHOLDER = "__REPO_ROOT__"
def run_driver(region: str, driver_body: str, timeout: int = 600) -> dict:
"""Run `driver_body` as the sole `-c` argument in a fresh Python process.
The driver is expected to print one JSON object between the
`__PIPELINE_RESULT_BEGIN__` / `__PIPELINE_RESULT_END__` sentinels.
The placeholder `__REPO_ROOT__` in the driver body is replaced with
`repr(REPO_ROOT)` so the subprocess can `sys.path.insert(0, ...)`
before importing `src.*`.
"""
env = os.environ.copy()
env["MARKET_INTEL_REGION"] = region
env["MARKET_INTEL_DEMO_MODE"] = "1"
env.pop("ANTHROPIC_API_KEY", None)
driver = driver_body.replace(_REPO_ROOT_PLACEHOLDER, repr(_REPO_ROOT))
proc = subprocess.run(
[sys.executable, "-c", driver],
env=env,
cwd=_REPO_ROOT,
capture_output=True,
text=True,
timeout=timeout,
)
if proc.returncode != 0:
raise AssertionError(
f"Subprocess failed for region={region}:\n"
f"stdout:\n{proc.stdout}\nstderr:\n{proc.stderr}"
)
stdout = proc.stdout
start = stdout.find(_SENTINEL_BEGIN)
end = stdout.find(_SENTINEL_END)
if start < 0 or end < 0:
raise AssertionError(
f"Missing result sentinels in subprocess output:\n{stdout}"
)
payload = stdout[start + len(_SENTINEL_BEGIN) : end].strip()
return json.loads(payload)
__all__ = ["run_driver"]
|