Jyotisha/tests/test_vedastro_health_probe_cache.py

#!/usr/bin/env python3
"""BUG-734: 官方 REST 健康探测的 TTL 缓存。

探测不是 ping——它 POST 一份虚构 smoke 排盘到官方 HoroscopePredictions，
并消耗对方每分钟 5 个限流令牌之一。改前每一轮前台请求都做一次。

两条边界同时成立：
- 前台路径（`run_gateway_packet`）读缓存；
- 诊断端点仍然实时，运维不得看到一个 60 秒前的假象。
"""

from __future__ import annotations

import inspect

import pytest

from scripts import vedastro_gateway


@pytest.fixture(autouse=True)
def _clean_cache(monkeypatch):
    """每个用例都从空缓存、确定的配置起步。"""
    monkeypatch.delenv("JYOTISH_SKIP_LOCAL_ENV", raising=False)
    monkeypatch.delenv("VEDASTRO_API_KEY", raising=False)
    monkeypatch.delenv("VEDASTRO_SELF_HOST_ENDPOINT", raising=False)
    monkeypatch.delenv("VEDASTRO_API_ENDPOINT", raising=False)
    monkeypatch.delenv("VEDASTRO_GATEWAY_MODE", raising=False)
    monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "1")
    vedastro_gateway.reset_rest_health_cache()
    yield
    vedastro_gateway.reset_rest_health_cache()


class _Clock:
    """可推进的单调时钟。"""

    def __init__(self) -> None:
        self.now = 1000.0

    def __call__(self) -> float:
        return self.now

    def advance(self, seconds: float) -> None:
        self.now += seconds


@pytest.fixture
def clock(monkeypatch) -> _Clock:
    c = _Clock()
    monkeypatch.setattr(vedastro_gateway, "_monotonic", c)
    return c


class _ProbeCounter:
    """替掉真正的外网探测，统计它被调了几次。"""

    def __init__(self, status: str = "official_verified") -> None:
        self.calls = 0
        self.status = status

    def __call__(self, *args, **kwargs) -> dict:
        self.calls += 1
        reason = "status_pass" if self.status == "official_verified" else "URLError"
        return {
            "status": self.status,
            "reason": reason,
            "endpoint": vedastro_gateway.OFFICIAL_CALCULATE_ENDPOINT,
            "transport": "rest",
        }


@pytest.fixture
def probe(monkeypatch) -> _ProbeCounter:
    counter = _ProbeCounter()
    from scripts import vedastro_rest_bridge

    monkeypatch.setattr(vedastro_rest_bridge, "probe_calculate_health", counter)
    return counter


# ---------------------------------------------------------------------------
# TTL 本身
# ---------------------------------------------------------------------------
def test_three_cached_probes_hit_the_network_once(probe, clock) -> None:
    results = [vedastro_gateway.probe_official_rest_health(force_refresh=False) for _ in range(3)]

    assert probe.calls == 1, f"连调 3 次只应探测 1 次，实际 {probe.calls} 次"
    assert all(r["status"] == "official_verified" for r in results)
    assert results[0] == results[1] == results[2]


def test_success_result_is_re_probed_after_the_success_ttl(probe, clock) -> None:
    vedastro_gateway.probe_official_rest_health(force_refresh=False)
    assert probe.calls == 1

    clock.advance(vedastro_gateway._REST_HEALTH_SUCCESS_TTL_SECONDS - 1)
    vedastro_gateway.probe_official_rest_health(force_refresh=False)
    assert probe.calls == 1, "TTL 之内不得重新探测"

    clock.advance(2)
    vedastro_gateway.probe_official_rest_health(force_refresh=False)
    assert probe.calls == 2, "推过 TTL 后必须重新探测"


def test_failure_is_not_cached_for_the_success_ttl(probe, clock) -> None:
    """红线 §5.4：对方恢复了我们不能还要再瞎等一分钟。"""
    probe.status = "official_blocked"
    assert vedastro_gateway.probe_official_rest_health(force_refresh=False)["status"] == "official_blocked"
    assert probe.calls == 1

    # 失败 TTL 之内：不重探。
    clock.advance(vedastro_gateway._REST_HEALTH_FAILURE_TTL_SECONDS - 1)
    vedastro_gateway.probe_official_rest_health(force_refresh=False)
    assert probe.calls == 1

    # 过了失败 TTL、但还远在成功 TTL 之内：必须重探。
    clock.advance(2)
    assert clock.now < 1000.0 + vedastro_gateway._REST_HEALTH_SUCCESS_TTL_SECONDS
    probe.status = "official_verified"
    recovered = vedastro_gateway.probe_official_rest_health(force_refresh=False)
    assert probe.calls == 2
    assert recovered["status"] == "official_verified", "服务恢复后必须能在失败 TTL 内被看见"


def test_failure_ttl_is_clearly_shorter_than_success_ttl() -> None:
    assert vedastro_gateway._REST_HEALTH_FAILURE_TTL_SECONDS < vedastro_gateway._REST_HEALTH_SUCCESS_TTL_SECONDS
    assert vedastro_gateway._REST_HEALTH_SUCCESS_TTL_SECONDS <= 60.0


# ---------------------------------------------------------------------------
# 配置一变立刻失效
# ---------------------------------------------------------------------------
def test_configuration_change_invalidates_the_cache(probe, clock, monkeypatch) -> None:
    vedastro_gateway.probe_official_rest_health(force_refresh=False)
    assert probe.calls == 1

    monkeypatch.setenv("VEDASTRO_API_KEY", "fictional-test-key")
    vedastro_gateway.probe_official_rest_health(force_refresh=False)
    assert probe.calls == 2, "配了 API key 之后必须重新探测"

    vedastro_gateway.probe_official_rest_health(force_refresh=False)
    assert probe.calls == 2, "同一份新配置应当重新命中缓存"

    monkeypatch.setenv("VEDASTRO_API_ENDPOINT", "https://example.invalid/api")
    vedastro_gateway.probe_official_rest_health(force_refresh=False)
    assert probe.calls == 3, "换 endpoint 必须重新探测"


def test_cache_key_never_carries_the_secret_value(monkeypatch) -> None:
    """AGENTS §8：缓存键只记「配没配」，不记 key 本身。"""
    monkeypatch.setenv("VEDASTRO_API_KEY", "fictional-test-key")
    key_with = vedastro_gateway._rest_health_cache_key()
    monkeypatch.delenv("VEDASTRO_API_KEY")
    key_without = vedastro_gateway._rest_health_cache_key()

    assert "fictional-test-key" not in repr(key_with), "缓存键里出现了 key 的值"
    assert key_with != key_without, "配没配 key 必须能区分开，否则缓存会跨配置串"


def test_network_disabled_is_not_probed_at_all(monkeypatch, clock) -> None:
    from scripts import vedastro_rest_bridge

    def explode(*args, **kwargs):  # pragma: no cover - must never run
        raise AssertionError("network disabled 时不得触碰外网")

    monkeypatch.setattr(vedastro_rest_bridge, "probe_calculate_health", explode)
    monkeypatch.setenv("VEDASTRO_ENABLE_NETWORK", "0")
    result = vedastro_gateway.probe_official_rest_health(force_refresh=False)
    assert result["status"] == "official_blocked"
    assert result["reason"] == "network_disabled"


# ---------------------------------------------------------------------------
# 诊断端点必须实时
# ---------------------------------------------------------------------------
def test_force_refresh_always_probes(probe, clock) -> None:
    for _ in range(3):
        vedastro_gateway.probe_official_rest_health(force_refresh=True)
    assert probe.calls == 3, "force_refresh 必须每次都打真探测"


def test_force_refresh_defaults_to_true_so_the_diagnostic_endpoint_stays_live(probe, clock) -> None:
    """`_compute_vedastro_gateway_status` 调的是裸 `gateway_status()`（该文件本单不得改）。

    因此默认值必须是实时探测，否则运维会看到一个 60 秒前的假象。
    """
    for fn in (vedastro_gateway.probe_official_rest_health, vedastro_gateway.gateway_status):
        assert inspect.signature(fn).parameters["force_refresh"].default is True

    vedastro_gateway.gateway_status()
    vedastro_gateway.gateway_status()
    assert probe.calls == 2, "诊断端点默认路径不得走缓存"


def test_forced_probe_sees_the_current_state_even_while_a_stale_entry_exists(probe, clock) -> None:
    """前台缓存里躺着一条旧结论时，诊断端点仍必须看到此刻的真实状态。"""
    probe.status = "official_blocked"
    vedastro_gateway.probe_official_rest_health(force_refresh=False)
    assert probe.calls == 1

    probe.status = "official_verified"
    forced = vedastro_gateway.probe_official_rest_health(force_refresh=True)
    assert forced["status"] == "official_verified"
    assert probe.calls == 2


# ---------------------------------------------------------------------------
# 前台路径
# ---------------------------------------------------------------------------
def test_gateway_status_passes_force_refresh_through(probe, clock) -> None:
    vedastro_gateway.gateway_status(force_refresh=False)
    vedastro_gateway.gateway_status(force_refresh=False)
    vedastro_gateway.gateway_status(force_refresh=False)
    assert probe.calls == 1


def test_run_gateway_packet_uses_the_cached_probe(monkeypatch, probe, clock) -> None:
    """每一轮前台请求不得各拿一个对方的限流令牌去做健康探测。"""
    seen: list[bool] = []
    real_status = vedastro_gateway.gateway_status

    def spy(*, force_refresh: bool = True):
        seen.append(force_refresh)
        return real_status(force_refresh=force_refresh)

    monkeypatch.setattr(vedastro_gateway, "gateway_status", spy)
    monkeypatch.setattr(
        vedastro_gateway,
        "_entrypoint_args",
        lambda *a, **k: __import__("types").SimpleNamespace(),
    )
    import sys
    import types as _types

    stub = _types.ModuleType("scripts.vedastro_user_entrypoint")
    stub.build_report = lambda args: {"input": {}, "runtime_mode": {}}
    monkeypatch.setitem(sys.modules, "scripts.vedastro_user_entrypoint", stub)

    for _ in range(4):
        vedastro_gateway.run_gateway_packet({"year": 1955}, question="x")

    assert seen == [False, False, False, False], "前台路径必须显式走缓存"
    assert probe.calls == 1, f"4 轮前台请求只应探测 1 次，实际 {probe.calls} 次"


def test_cached_result_cannot_be_mutated_by_a_caller(probe, clock) -> None:
    first = vedastro_gateway.probe_official_rest_health(force_refresh=False)
    first["status"] = "tampered"
    second = vedastro_gateway.probe_official_rest_health(force_refresh=False)
    assert second["status"] == "official_verified"
    assert probe.calls == 1


def test_probe_does_not_hold_the_lock_while_on_the_network(monkeypatch, clock) -> None:
    """不得在等外网时把进程级锁按住（BUG-718 的教训）。"""
    observed: list[bool] = []

    def probing(*args, **kwargs):
        acquired = vedastro_gateway._REST_HEALTH_LOCK.acquire(blocking=False)
        observed.append(acquired)
        if acquired:
            vedastro_gateway._REST_HEALTH_LOCK.release()
        return {"status": "official_verified", "reason": "status_pass", "transport": "rest"}

    from scripts import vedastro_rest_bridge

    monkeypatch.setattr(vedastro_rest_bridge, "probe_calculate_health", probing)
    vedastro_gateway.probe_official_rest_health(force_refresh=False)

    assert observed == [True], "探测期间锁必须是可获取的"