Jyotisha/tests/test_vedastro_consultation_snapshot.py

"""Consult rectification gate reuses or defers the official VedAstro snapshot."""
from __future__ import annotations

import json

import pytest

from scripts.jyotish_api_server import JyotishAPIHandler
from scripts.vedastro_consultation_snapshot import (
    consult_rectification_vedastro_gateway,
    negative_cache_dir,
    snapshot_identity_from_request,
    store_negative_snapshot,
)
from scripts.vedastro_snapshot_cache import snapshot_cache_dir, store_snapshot


def _handler() -> JyotishAPIHandler:
    return object.__new__(JyotishAPIHandler)


def _planets() -> dict:
    return {
        "Sun": {"lon": 80.0},
        "Moon": {"lon": 123.0},
        "Mars": {"lon": 210.0},
        "Mercury": {"lon": 75.0},
        "Jupiter": {"lon": 15.0},
        "Venus": {"lon": 102.0},
        "Saturn": {"lon": 330.0},
        "Rahu": {"lon": 5.0},
        "Ketu": {"lon": 185.0},
    }


def _body(**extra) -> dict:
    payload = {
        "year": 1990,
        "month": 1,
        "day": 1,
        "hour": 5,
        "minute": 0,
        "second": 0,
        "lat": 19.0,
        "lon": 72.8,
        "tz": 5.5,
        "ayanamsa": "raman",
        "node_mode": "mean",
        "reference_date": "2026-10-05",
        "today": "2026-10-05",
        "current_date": "2026-10-05",
        "planets": _planets(),
        "ascendant": {"lon": 15.0},
        "declared_accuracy": "minute",
        "time_source": "family_clear",
    }
    payload.update(extra)
    return payload


def _verified(request_id: str = "same-day") -> dict:
    return {
        "scope": "vedastro_gateway_run",
        "status": "official_verified",
        "official_closure_state": "official_verified",
        "official_closure_reason": "official_raw_response_present",
        "official_raw_response": {"request_id": request_id, "natal": {"sun": "Leo"}},
    }


@pytest.fixture
def caches(tmp_path, monkeypatch):
    monkeypatch.setenv("JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR", str(tmp_path / "positive"))
    monkeypatch.setenv("JYOTISH_VEDASTRO_SNAPSHOT_NEGATIVE_CACHE_DIR", str(tmp_path / "negative"))
    return tmp_path


def _forbid_runner(monkeypatch) -> list[str]:
    calls: list[str] = []

    def runner(*_args, **_kwargs):
        calls.append("runner")
        raise AssertionError("snapshot runner must not start on the consult gate")

    def report(*_args, **_kwargs):
        calls.append("report")
        raise AssertionError("build_report must not start on the consult gate")

    monkeypatch.setattr(
        "scripts.vedastro_service_adapter._try_official_capability_runner_snapshot_bundle",
        runner,
    )
    monkeypatch.setattr("scripts.vedastro_user_entrypoint.build_report", report)
    return calls


def test_consult_gate_does_not_call_the_snapshot_runner(caches, monkeypatch) -> None:
    calls = _forbid_runner(monkeypatch)
    handler = _handler()

    def live_gateway(_body):
        calls.append("gateway")
        raise AssertionError("consult gate must not call the live gateway")

    handler._compute_vedastro_gateway_run = live_gateway  # type: ignore[method-assign]
    result = handler._compute_rectification_gate({
        **_body(),
        "defer_optional_external_evidence": True,
    })
    gateway = result["vedastro_gateway"]
    assert calls == []
    assert gateway["official_closure_state"] == "official_blocked"
    assert gateway["official_closure_reason"] == "deferred_in_consultation"
    assert gateway["status"] != "official_verified"
    assert "official_raw_response" not in gateway
    assert result["endpoint"] == "rectification_gate"


def test_negative_cache_hits_the_same_utc_day_and_misses_the_next(caches, monkeypatch) -> None:
    calls = _forbid_runner(monkeypatch)
    stored = store_negative_snapshot(_body(), {
        "scope": "vedastro_gateway_run",
        "status": "official_blocked",
        "official_closure_state": "official_blocked",
        "official_closure_reason": "foreground_optional_evidence_timeout",
        "email": "hidden@example.com",
    }, today="2026-10-05")
    assert stored is not None
    assert "email" not in json.dumps(stored["gateway"])

    same_day = consult_rectification_vedastro_gateway({
        **_body(),
        "defer_optional_external_evidence": True,
    })
    assert same_day["official_closure_reason"] == "foreground_optional_evidence_timeout"
    before = sorted(path.read_bytes() for path in negative_cache_dir().glob("*.json"))
    again = consult_rectification_vedastro_gateway({
        **_body(),
        "defer_optional_external_evidence": True,
    })
    after = sorted(path.read_bytes() for path in negative_cache_dir().glob("*.json"))
    assert again["official_closure_reason"] == "foreground_optional_evidence_timeout"
    assert before == after

    next_day = consult_rectification_vedastro_gateway({
        **_body(reference_date="2026-10-06", today="2026-10-06", current_date="2026-10-06"),
        "defer_optional_external_evidence": True,
    })
    assert next_day["official_closure_reason"] == "deferred_in_consultation"
    assert calls == []
    assert negative_cache_dir().resolve() != snapshot_cache_dir().resolve()


def test_same_day_verified_snapshot_is_reused_ahead_of_the_negative_cache(caches, monkeypatch) -> None:
    calls = _forbid_runner(monkeypatch)
    request = _body()
    store_snapshot(request, _verified("cached-official"))
    store_negative_snapshot(request, {
        "status": "official_blocked",
        "official_closure_state": "official_blocked",
        "official_closure_reason": "foreground_optional_evidence_timeout",
    }, today="2026-10-05")
    normalized = {
        **request,
        "hour": 5.0,
        "minute": 0.0,
        "defer_optional_external_evidence": True,
        "vedastro_snapshot_identity": snapshot_identity_from_request(request),
    }
    reused = consult_rectification_vedastro_gateway(normalized)
    assert reused["official_closure_state"] == "official_verified"
    assert reused["official_raw_response"]["request_id"] == "cached-official"
    missed = consult_rectification_vedastro_gateway({
        **request,
        "hour": 5.0,
        "defer_optional_external_evidence": True,
    })
    assert missed["official_closure_reason"] == "deferred_in_consultation"
    assert calls == []


def test_rectification_surface_without_the_flag_still_uses_the_live_gateway(caches) -> None:
    handler = _handler()
    seen: list[dict] = []

    def live_gateway(body):
        seen.append(dict(body))
        return _verified("rectification-surface")

    handler._compute_vedastro_gateway_run = live_gateway  # type: ignore[method-assign]
    result = handler._compute_rectification_gate(_body())
    assert len(seen) == 1
    assert "defer_optional_external_evidence" not in seen[0]
    gateway = result["vedastro_gateway"]
    assert gateway["official_closure_state"] == "official_verified"
    assert gateway["official_closure_reason"] == "official_raw_response_present"
    assert gateway["official_raw_response"]["request_id"] == "rectification-surface"
    assert result["endpoint"] == "rectification_gate"
    assert result["success"] is True