Jyotisha/tests/test_consultation_contract_golden.py

"""Consultation and chart API contract golden: key paths and types only.

Frozen fixtures record the baseline response shape. After later engine merges,
a fresh capture must be a key-path superset with matching types. Values are
never stored. Smoke birth data is fictional (1990-01-01 Beijing).
"""

from __future__ import annotations

import json
import os
import socket
import sys
import threading
import urllib.error
import urllib.request
from http.server import ThreadingHTTPServer
from pathlib import Path
from typing import Any

import pytest

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))
GOLDEN = ROOT / "tests" / "golden" / "consultation_contract_keypaths_v1.json"
CAPTURE_ENV = "JYOTISH_CAPTURE_CONSULTATION_GOLDEN"

# Fictional smoke chart used by existing golden_cases.json. Not a real person.
SMOKE_BIRTH = {
    "year": 1990,
    "month": 1,
    "day": 1,
    "hour": 12,
    "minute": 0,
    "lat": 39.9,
    "lon": 116.4,
    "tz": 8,
    "city": "Beijing",
    "note": "fictional_smoke",
}

THEMES = ("career", "marriage", "wealth", "timing", "health")
FORBIDDEN_PRIVACY_MARKERS = ("1993-04-17", "17/04/1993", "14:49")

_ROUTE_LAYERS = {
    "career": ["D1", "D10", "10th house/lord", "A10", "AmK", "Vimshottari", "Narayana", "Transit"],
    "marriage": ["D1", "D9", "7th house/lord", "Venus/Jupiter", "DK", "UL", "A7", "Vimshottari", "Narayana", "Transit"],
    "wealth": ["D1", "D2", "D11", "2nd/11th/9th/5th houses", "Wealth Yogas", "Ashtakavarga", "Dasha"],
    "timing": ["Vimshottari", "Narayana", "Transit", "Varga", "negative holdout gate"],
    "health": ["D1", "D6", "D8", "6th/8th houses", "Dasha", "non-medical boundary"],
}
_ROUTE_CLAIMS = {
    "career": "career_direction_and_broad_timing_only",
    "marriage": "relationship_pattern_and_broad_window_only",
    "wealth": "wealth_structure_not_financial_advice",
    "timing": "candidate_day_month_window_only_until_holdout_passes",
    "health": "wellbeing_pressure_patterns_not_medical_diagnosis",
}


def json_type(value: Any) -> str:
    if value is None:
        return "null"
    if isinstance(value, bool):
        return "boolean"
    if isinstance(value, (int, float)):
        return "number"
    if isinstance(value, str):
        return "string"
    if isinstance(value, list):
        return "array"
    if isinstance(value, dict):
        return "object"
    return type(value).__name__


def flatten_keypaths(value: Any, prefix: str = "", *, depth: int = 0, max_depth: int = 2) -> dict[str, str]:
    """Record dotted key paths and JSON types, two object levels deep.

    Arrays contribute their type and the first element's type plus one child
    level. This keeps the golden small while locking frontend-consumed keys.
    """
    out: dict[str, str] = {}
    if prefix:
        out[prefix] = json_type(value)
    if depth >= max_depth:
        return out
    if isinstance(value, dict):
        for key, child in sorted(value.items(), key=lambda item: str(item[0])):
            path = f"{prefix}.{key}" if prefix else str(key)
            out.update(flatten_keypaths(child, path, depth=depth + 1, max_depth=max_depth))
    elif isinstance(value, list) and value:
        out.update(flatten_keypaths(value[0], f"{prefix}[0]", depth=depth + 1, max_depth=max_depth))
    return out


def _workflow_body(theme: str, *, skip_full_reading: bool) -> dict[str, Any]:
    timing_horizon = "next_12_months" if theme in {"timing", "annual"} else None
    return {
        **{key: SMOKE_BIRTH[key] for key in ("year", "month", "day", "hour", "minute", "lat", "lon", "tz", "city")},
        "question": f"请用虚构资料计算 {theme} 主题",
        "question_text": f"请用虚构资料计算 {theme} 主题",
        "theme": [theme],
        "entry_mode": "direct_chart",
        "plan_version": "consultation-plan-v2",
        "strict_workflow_route": theme,
        "required_layers": list(_ROUTE_LAYERS[theme]),
        "claim_boundary": _ROUTE_CLAIMS[theme],
        "plan_depth": "standard",
        "requested_domains": [theme],
        "timing_horizon": timing_horizon,
        "precision_boundary": "server_evidence_required",
        "required_evidence_categories": ["natal_foundation", "domain", "timing", "validation"],
        "defer_optional_external_evidence": True,
        "skip_full_reading_for_thematic": skip_full_reading,
    }


class _ApiThread(threading.Thread):
    def __init__(self, server: ThreadingHTTPServer) -> None:
        super().__init__(daemon=True)
        self.server = server

    def run(self) -> None:
        self.server.serve_forever()


def _start_local_api() -> tuple[ThreadingHTTPServer, str]:
    from scripts.jyotish_api_server import DEFAULT_ALLOWED_HOSTS, DEFAULT_ALLOWED_ORIGINS, JyotishAPIHandler

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(("127.0.0.1", 0))
    port = sock.getsockname()[1]
    sock.close()
    server = ThreadingHTTPServer(("127.0.0.1", port), JyotishAPIHandler)
    server.daemon_threads = True
    origin = f"http://127.0.0.1:{port}"
    server.allowed_origins = set(DEFAULT_ALLOWED_ORIGINS) | {origin}
    server.allowed_hosts = set(DEFAULT_ALLOWED_HOSTS)
    thread = _ApiThread(server)
    thread.start()
    return server, origin


def _post(base: str, path: str, body: dict[str, Any]) -> dict[str, Any]:
    payload = json.dumps(body).encode("utf-8")
    request = urllib.request.Request(
        f"{base}{path}",
        data=payload,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "Origin": base,
            "Host": base.split("://", 1)[1],
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=180) as response:
            return json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as error:
        detail = error.read().decode("utf-8", errors="replace")
        raise AssertionError(f"{path} HTTP {error.code}: {detail[:500]}") from error


def capture_contract_shapes(base: str) -> dict[str, dict[str, str]]:
    cases: dict[str, dict[str, str]] = {}
    for theme in THEMES:
        for skip in (True, False):
            name = f"consultation_workflow.{theme}.skip_full_reading_{str(skip).lower()}"
            body = _workflow_body(theme, skip_full_reading=skip)
            cases[name] = flatten_keypaths(_post(base, "/api/consultation_workflow", body))

    chart = _post(base, "/api/chart", {**SMOKE_BIRTH, "ayanamsa": "raman", "node_mode": "mean"})
    cases["chart"] = flatten_keypaths(chart)

    planets = chart.get("planets") if isinstance(chart.get("planets"), dict) else {}
    ascendant = chart.get("ascendant") if isinstance(chart.get("ascendant"), dict) else {}
    varga = _post(
        base,
        "/api/varga_full",
        {
            **{key: SMOKE_BIRTH[key] for key in ("year", "month", "day", "hour", "minute", "lat", "lon", "tz")},
            "planets": planets,
            "ascendant": ascendant,
            "divisions": ["D9", "D10"],
        },
    )
    cases["varga_full"] = flatten_keypaths(varga)

    daily = _post(
        base,
        "/api/daily_guidance",
        {
            **{key: SMOKE_BIRTH[key] for key in ("year", "month", "day", "hour", "minute", "lat", "lon", "tz")},
            "date": "2026-09-03",
            "today": "2026-09-03",
            "reference_date": "2026-09-03",
            "transit_date": "2026-09-03",
            "ayanamsa": "raman",
            "node_mode": "mean",
        },
    )
    cases["daily_guidance"] = flatten_keypaths(daily)

    moon = planets.get("Moon") if isinstance(planets.get("Moon"), dict) else {}
    moon_lon = moon.get("lon", 0)
    synastry = _post(base, "/api/synastry", {"male_moon": moon_lon, "female_moon": (float(moon_lon) + 30) % 360})
    cases["synastry"] = flatten_keypaths(synastry)
    return cases


def write_golden(cases: dict[str, dict[str, str]]) -> None:
    GOLDEN.parent.mkdir(parents=True, exist_ok=True)
    document = {
        "schema_version": 1,
        "kind": "keypath_types_only",
        "max_path_segments": 2,
        "baseline_commit": "58cae371bbcb43ac9d4dfe8a66c78be46c88a1ee",
        "smoke_birth": SMOKE_BIRTH,
        "cases": cases,
    }
    GOLDEN.write_text(json.dumps(document, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")


def load_golden() -> dict[str, Any]:
    return json.loads(GOLDEN.read_text(encoding="utf-8"))


def assert_superset(frozen: dict[str, str], live: dict[str, str], *, case: str) -> None:
    missing = sorted(path for path in frozen if path not in live)
    type_mismatch = sorted(
        f"{path}: frozen={frozen[path]} live={live[path]}"
        for path in frozen
        if path in live and frozen[path] != live[path]
    )
    assert not missing, f"{case} lost key paths: {missing[:20]}"
    assert not type_mismatch, f"{case} type changes: {type_mismatch[:20]}"


def test_golden_fixture_is_fictional_and_value_free() -> None:
    assert GOLDEN.exists(), "run capture first (JYOTISH_CAPTURE_CONSULTATION_GOLDEN=1)"
    raw = GOLDEN.read_text(encoding="utf-8")
    for marker in FORBIDDEN_PRIVACY_MARKERS:
        assert marker not in raw
    document = load_golden()
    assert document["kind"] == "keypath_types_only"
    assert document["smoke_birth"]["note"] == "fictional_smoke"
    assert document["smoke_birth"]["year"] == 1990
    for case, paths in document["cases"].items():
        assert paths, case
        assert all(kind in {"null", "boolean", "number", "string", "array", "object"} for kind in paths.values()), case


@pytest.mark.slow
def test_live_http_capture_is_keypath_superset() -> None:
    if os.environ.get("JYOTISH_LIVE_CONTRACT_GOLDEN", "").strip().lower() not in {"1", "true", "yes"}:
        pytest.skip("set JYOTISH_LIVE_CONTRACT_GOLDEN=1 to recapture over HTTP (~1 min after API import)")
    os.environ.setdefault("JYOTISH_SKIP_LOCAL_ENV", "1")
    os.environ["VEDASTRO_ENABLE_NETWORK"] = "0"
    os.environ["PYTHONDONTWRITEBYTECODE"] = "1"
    frozen = load_golden()["cases"]
    server, base = _start_local_api()
    try:
        live = capture_contract_shapes(base)
    finally:
        server.shutdown()
        server.server_close()
    if os.environ.get(CAPTURE_ENV, "").strip() in {"1", "true", "yes"}:
        write_golden(live)
        frozen = live
    assert set(frozen) <= set(live)
    for case, paths in frozen.items():
        assert_superset(paths, live[case], case=case)


def test_golden_covers_required_frontend_surfaces() -> None:
    cases = load_golden()["cases"]
    required = {
        "consultation_workflow.career.skip_full_reading_true",
        "consultation_workflow.career.skip_full_reading_false",
        "consultation_workflow.marriage.skip_full_reading_true",
        "consultation_workflow.marriage.skip_full_reading_false",
        "consultation_workflow.wealth.skip_full_reading_true",
        "consultation_workflow.wealth.skip_full_reading_false",
        "consultation_workflow.timing.skip_full_reading_true",
        "consultation_workflow.timing.skip_full_reading_false",
        "consultation_workflow.health.skip_full_reading_true",
        "consultation_workflow.health.skip_full_reading_false",
        "chart",
        "varga_full",
        "daily_guidance",
        "synastry",
    }
    assert required <= set(cases)
    for name in required:
        assert "success" in cases[name] or name == "synastry"


if __name__ == "__main__":
    os.environ.setdefault("JYOTISH_SKIP_LOCAL_ENV", "1")
    os.environ["VEDASTRO_ENABLE_NETWORK"] = "0"
    server, base = _start_local_api()
    try:
        cases = capture_contract_shapes(base)
    finally:
        server.shutdown()
        server.server_close()
    write_golden(cases)
    print(f"wrote {GOLDEN} cases={len(cases)}")