Jyotisha/tests/test_marriage_match_api.py

"""Marriage-match API glue: order, cache, and route registration."""

from __future__ import annotations

import json
import os
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
SCRIPTS = ROOT / "scripts"
if str(SCRIPTS) not in sys.path:
    sys.path.insert(0, str(SCRIPTS))

from marriage_match import build_marriage_match_dossier
from marriage_match_service.api import (
    CACHE_NAME,
    MarriageMatchInputError,
    compute_marriage_match,
    compute_synastry_full_dossier,
)
import marriage_match_service.api as api
from marriage_match_service.full_dossier import RULES_VERSION
import api_heavy_compute_gate


def chart(asc, moon, venus, mars):
    return {
        "ascendant": {"longitude": asc},
        "planets": {
            "Moon": {"longitude": moon},
            "Venus": {"longitude": venus},
            "Mars": {"longitude": mars},
            "Saturn": {"longitude": 210},
            "Sun": {"longitude": 120},
            "Rahu": {"longitude": 300},
            "Ketu": {"longitude": 120},
        },
    }


def _pressure_signature(report):
    people = []
    for key in ("person_a", "person_b"):
        rows = report["papasamya"][key]["active_placements"]
        people.append(tuple(sorted(
            (row["planet"], row["reference"], row["house"]) for row in rows
        )))
    return tuple(sorted(people))


def test_swapped_people_keep_pairing_kuja_and_pressure():
    left = chart(10, 40, 70, 190)
    right = chart(100, 130, 160, 250)
    forward = build_marriage_match_dossier(left, right)
    reverse = build_marriage_match_dossier(right, left)
    assert forward["ashtakoota_36"]["international_core"]["scores"] == reverse["ashtakoota_36"]["international_core"]["scores"]
    assert forward["ashtakoota_36"]["international_core"]["total_score"] == reverse["ashtakoota_36"]["international_core"]["total_score"]
    assert forward["kuja_mangal_dosha"]["outcome"] == reverse["kuja_mangal_dosha"]["outcome"]
    assert forward["kuja_mangal_dosha"]["person_a"]["active_reference_count"] == reverse["kuja_mangal_dosha"]["person_b"]["active_reference_count"]
    assert _pressure_signature(forward) == _pressure_signature(reverse)


def test_snapshot_endpoint_rejects_a_non_object():
    with pytest.raises(MarriageMatchInputError):
        compute_marriage_match({"person_a": []})


def test_disk_cache_reuses_one_build(monkeypatch, tmp_path):
    monkeypatch.setenv("GITHUB_SHA", "marriage-match-cache-test")
    def cache_dir(name: str = "consult_full_data_cache"):
        path = tmp_path / name
        path.mkdir(parents=True, exist_ok=True)
        return path

    monkeypatch.setattr(api._cache, "cache_dir", cache_dir)
    calls = {"n": 0}

    def fake(_handler, _person_a, _person_b):
        calls["n"] += 1
        return {"rules_version": RULES_VERSION, "dossier": {"schema": "fixture"}, "raman_marriage_match": {}}

    monkeypatch.setattr(
        "marriage_match_service.api.build_synastry_full_dossier",
        fake,
    )
    body = {
        "person_a": {"year": 1990, "month": 1, "day": 2, "hour": 3, "minute": 4, "lat": 1, "lon": 2, "tz": 8},
        "person_b": {"year": 1991, "month": 5, "day": 6, "hour": 7, "minute": 8, "lat": 3, "lon": 4, "tz": 8},
    }
    first = compute_synastry_full_dossier(object(), body)
    second = compute_synastry_full_dossier(object(), body)
    assert first == second
    assert calls["n"] == 1
    assert list((tmp_path / CACHE_NAME).glob("*.json"))


def test_missing_commit_does_not_reuse_the_disk_cache(monkeypatch):
    monkeypatch.delenv("GITHUB_SHA", raising=False)
    calls = {"n": 0}

    def fake(_handler, _person_a, _person_b):
        calls["n"] += 1
        return {"rules_version": RULES_VERSION, "dossier": {}, "raman_marriage_match": {}}

    monkeypatch.setattr("marriage_match_service.api.build_synastry_full_dossier", fake)
    body = {
        "person_a": {"year": 1990, "month": 1, "day": 2, "hour": 3, "minute": 4, "lat": 1, "lon": 2, "tz": 8},
        "person_b": {"year": 1991, "month": 5, "day": 6, "hour": 7, "minute": 8, "lat": 3, "lon": 4, "tz": 8},
    }
    compute_synastry_full_dossier(object(), body)
    compute_synastry_full_dossier(object(), body)
    assert calls["n"] == 2


def test_routes_are_registered_and_the_dossier_is_heavy():
    source = (ROOT / "scripts" / "jyotish_api_server.py").read_text(encoding="utf-8")
    assert "elif path == '/api/marriage-match':" in source
    assert "elif path == '/api/synastry_full_dossier':" in source
    assert "/api/synastry_full_dossier" in api_heavy_compute_gate.HEAVY_COMPUTE_PATHS
    assert "authorizeUsage" not in (ROOT / "frontend" / "src" / "lib" / "marriage-match-service.ts").read_text(encoding="utf-8")
    assert not (ROOT / "scripts" / "synastry.py").exists()


def test_snapshot_golden_matches_the_engine():
    fixture_path = ROOT / "frontend" / "tests" / "fixtures" / "marriage-match-snapshot.json"
    payload = {"person_a": chart(12, 48, 96, 200), "person_b": chart(140, 210, 30, 300)}
    result = compute_marriage_match(payload)
    if os.environ.get("WRITE_MARRIAGE_MATCH_GOLDEN") == "1":
        fixture_path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    stored = json.loads(fixture_path.read_text(encoding="utf-8"))
    assert result == stored