awoooi/apps/api/tests/test_paid_provider_canary_runner.py

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_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