169 lines
4.6 KiB
Python
169 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[3]
|
|
RUNNER = REPO_ROOT / "scripts" / "ops" / "run-paid-provider-canary.py"
|
|
|
|
|
|
def _load_runner_module(name: str):
|
|
spec = importlib.util.spec_from_file_location(name, RUNNER)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_runner_help_bootstraps_api_import_from_repo_root() -> None:
|
|
env = dict(os.environ)
|
|
env["PYTHONDONTWRITEBYTECODE"] = "1"
|
|
|
|
result = subprocess.run(
|
|
[sys.executable, str(RUNNER), "--help"],
|
|
cwd=REPO_ROOT,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
assert "--run-ref" in result.stdout
|
|
assert "ModuleNotFoundError" not in result.stderr
|
|
|
|
|
|
def test_runner_bootstraps_flat_production_api_image_layout(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
module = _load_runner_module("paid_canary_runner_flat_image")
|
|
(tmp_path / "src").mkdir()
|
|
|
|
selected = module._bootstrap_api_import(tmp_path)
|
|
|
|
assert selected == tmp_path
|
|
assert str(tmp_path) in sys.path
|
|
|
|
|
|
def test_runner_initializes_and_closes_redis_around_validation(monkeypatch) -> None:
|
|
module = _load_runner_module("paid_canary_runner_lifecycle")
|
|
events: list[object] = []
|
|
|
|
async def _init() -> None:
|
|
events.append("init")
|
|
|
|
async def _execute(**kwargs):
|
|
events.append(("execute", kwargs))
|
|
return {"status": "ok"}
|
|
|
|
async def _close() -> None:
|
|
events.append("close")
|
|
|
|
monkeypatch.setattr(module, "_init_runtime", _init)
|
|
monkeypatch.setattr(module, "_execute_validation", _execute)
|
|
monkeypatch.setattr(module, "_close_runtime", _close)
|
|
|
|
result = asyncio.run(module._run_validation(run_ref="source-sha"))
|
|
|
|
assert result == {"status": "ok"}
|
|
assert events == ["init", ("execute", {"run_ref": "source-sha"}), "close"]
|
|
|
|
|
|
def test_runner_closes_redis_when_validation_fails(monkeypatch) -> None:
|
|
module = _load_runner_module("paid_canary_runner_failure_lifecycle")
|
|
events: list[str] = []
|
|
|
|
async def _init() -> None:
|
|
events.append("init")
|
|
|
|
async def _execute(**_kwargs):
|
|
events.append("execute")
|
|
raise RuntimeError("bounded_failure")
|
|
|
|
async def _close() -> None:
|
|
events.append("close")
|
|
|
|
monkeypatch.setattr(module, "_init_runtime", _init)
|
|
monkeypatch.setattr(module, "_execute_validation", _execute)
|
|
monkeypatch.setattr(module, "_close_runtime", _close)
|
|
|
|
with pytest.raises(RuntimeError, match="bounded_failure"):
|
|
asyncio.run(module._run_validation(run_ref="source-sha"))
|
|
|
|
assert events == ["init", "execute", "close"]
|
|
|
|
|
|
def test_runner_redacts_unexpected_exception_text(
|
|
monkeypatch,
|
|
capsys,
|
|
) -> None:
|
|
module = _load_runner_module("paid_canary_runner")
|
|
|
|
leak = "http://192.168.0.111:11434/?key=secret-like-value"
|
|
|
|
async def _failed(**_kwargs):
|
|
raise LookupError(leak)
|
|
|
|
monkeypatch.setattr(module, "_run_validation", _failed)
|
|
monkeypatch.setattr(
|
|
sys,
|
|
"argv",
|
|
[
|
|
str(RUNNER),
|
|
"--run-ref",
|
|
"source-sha",
|
|
"--authorization-ref",
|
|
"user-approved-20260716",
|
|
],
|
|
)
|
|
|
|
assert module.main() == 2
|
|
output = capsys.readouterr()
|
|
payload = json.loads(output.out)
|
|
assert payload["error_code"] == "canary_failed_LookupError"
|
|
assert leak not in output.out
|
|
assert leak not in output.err
|
|
|
|
|
|
def test_runner_returns_nonzero_for_partial_five_lane_validation(
|
|
monkeypatch,
|
|
capsys,
|
|
) -> None:
|
|
module = _load_runner_module("paid_canary_runner_partial")
|
|
|
|
async def _partial(**_kwargs):
|
|
return {
|
|
"schema_version": "paid_provider_sre_canary_receipt_v3",
|
|
"independent_verifier": {
|
|
"passed": False,
|
|
"paid_activation_verifier_passed": True,
|
|
"five_lane_verifier_passed": False,
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(module, "_run_validation", _partial)
|
|
monkeypatch.setattr(
|
|
sys,
|
|
"argv",
|
|
[
|
|
str(RUNNER),
|
|
"--run-ref",
|
|
"source-sha",
|
|
"--authorization-ref",
|
|
"user-approved-20260716",
|
|
],
|
|
)
|
|
|
|
assert module.main() == 1
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["independent_verifier"]["paid_activation_verifier_passed"] is True
|
|
assert payload["independent_verifier"]["five_lane_verifier_passed"] is False
|