Jyotisha/tests/test_shadbala_service_api.py

"""BUG-1463: `/api/shadbala` with the birth moment returned 500 (`NameError: swe`).

The endpoint now lives in scripts/shadbala_service/api.py and the handler only
forwards the body. These tests go through the real HTTP handler on a local
port, so the regression is caught at the same surface the caller saw.
Birth data is the fictional golden profile shared with the frontend tests
(1990-06-15 12:00 Beijing, raman).
"""

from __future__ import annotations

import json
import socket
import subprocess
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
import swisseph as swe

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

from scripts.api_heavy_compute_gate import reset_heavy_compute_gate  # noqa: E402
from scripts.shadbala import build_shadbala_context, calc_dig_bala_precise  # noqa: E402

GOLDEN = json.loads((ROOT / "frontend" / "tests" / "fixtures" / "chart-insight-golden.json").read_text(encoding="utf-8"))
PLANETS = GOLDEN["chart"]["planets"]
ASCENDANT = GOLDEN["chart"]["ascendant"]
BIRTH = {
    "year": 1990, "month": 6, "day": 15, "hour": 12, "minute": 0, "second": 0,
    "lat": 39.9042, "lon": 116.4074, "tz": 8, "ayanamsa": "raman",
}
SEVEN = ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn")


@pytest.fixture(scope="module")
def api_base():
    import os

    from scripts.jyotish_api_server import DEFAULT_ALLOWED_HOSTS, DEFAULT_ALLOWED_ORIGINS, JyotishAPIHandler

    previous = os.environ.get("JYOTISH_API_RATE_LIMIT_PER_MINUTE")
    os.environ["JYOTISH_API_RATE_LIMIT_PER_MINUTE"] = "0"
    reset_heavy_compute_gate()
    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
    base = f"http://127.0.0.1:{port}"
    server.allowed_origins = set(DEFAULT_ALLOWED_ORIGINS) | {base}
    server.allowed_hosts = set(DEFAULT_ALLOWED_HOSTS)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        yield base
    finally:
        server.shutdown()
        server.server_close()
        reset_heavy_compute_gate()
        if previous is None:
            os.environ.pop("JYOTISH_API_RATE_LIMIT_PER_MINUTE", None)
        else:
            os.environ["JYOTISH_API_RATE_LIMIT_PER_MINUTE"] = previous


def _post(base: str, body: dict[str, Any]) -> tuple[int, dict[str, Any]]:
    request = urllib.request.Request(
        f"{base}/api/shadbala",
        data=json.dumps(body).encode("utf-8"),
        method="POST",
        headers={"Content-Type": "application/json", "Origin": base, "Host": base.split("://", 1)[1]},
    )
    try:
        with urllib.request.urlopen(request, timeout=120) as response:
            return response.status, json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as error:
        return error.code, json.loads(error.read().decode("utf-8") or "{}")


def test_birth_data_request_returns_precise_shadbala(api_base) -> None:
    status, payload = _post(api_base, {**BIRTH, "planets": PLANETS, "ascendant": ASCENDANT})

    assert status == 200, payload
    assert payload["success"] is True
    result = payload["result"]
    # The birth moment switches Kala and Chesta to the Swiss Ephemeris context.
    assert result["method_variants"] == {
        "kala": "bphs_local_solar_events_ahargana_declination",
        "chesta": "bphs_bounded_surya_mean_motion_seeghrochcha",
    }
    assert sorted(result["ranking"]) == sorted(SEVEN)
    for planet in SEVEN:
        row = result["planets"][planet]
        assert row["total_rupas"] > 0
        assert row["total_virupas"] == pytest.approx(row["total_rupas"] * 60, abs=0.05)

    # Dig Bala is read from the Placidus bhava midpoints of the same instant (UT = 12:00 - 8h).
    context = build_shadbala_context(swe.julday(1990, 6, 15, 4.0), 39.9042, 116.4074, "raman", 8.0)
    for planet in SEVEN:
        expected = calc_dig_bala_precise(planet, float(PLANETS[planet]["lon"]), context["house_midpoints"])
        assert result["planets"][planet]["dig_bala"] == expected

    assert payload["rule_variants"]["selected"] == ["core_sixfold", "advanced_evidence"]
    assert payload["advanced_layer"]["source"] == "scripts/shadbala_advanced.py"
    assert set(payload["advanced_layer"]["kala_vmdh"]) == set(SEVEN)


def test_birth_data_changes_the_strength_versus_planets_only(api_base) -> None:
    status, payload = _post(api_base, {"planets": PLANETS, "ascendant": ASCENDANT, "hour": 12, "minute": 0})
    assert status == 200, payload
    legacy = payload["result"]
    assert legacy["method_variants"]["kala"] == "legacy_approximation"
    # Without the birth moment the answer is the chart page's `modules.shadbala` golden.
    golden = GOLDEN["chart_modules_shadbala"]["planets"]
    for planet in SEVEN:
        assert legacy["planets"][planet]["total_rupas"] == pytest.approx(golden[planet]["total_rupas"], abs=1e-4)

    _, precise = _post(api_base, {**BIRTH, "planets": PLANETS, "ascendant": ASCENDANT})
    assert any(
        precise["result"]["planets"][planet]["dig_bala"] != legacy["planets"][planet]["dig_bala"]
        for planet in SEVEN
    )


def test_bad_inputs_are_400_not_500(api_base) -> None:
    status, _ = _post(api_base, {**BIRTH, "planets": PLANETS, "ascendant": ASCENDANT, "ayanamsa": "not-an-ayanamsa"})
    assert status == 400
    status, _ = _post(api_base, {**BIRTH, "planets": {"Mars": PLANETS["Mars"]}, "ascendant": ASCENDANT})
    assert status == 400


def test_shadbala_service_and_api_server_have_no_undefined_names() -> None:
    """BUG-1463 was a module name used where it was never imported; ruff F821 sees that statically."""
    ruff = pytest.importorskip("ruff")
    targets = ["scripts/jyotish_api_server.py", "scripts/offline_compute_mixins.py", "scripts/shadbala_service"]
    completed = subprocess.run(
        [ruff.find_ruff_bin(), "check", "--isolated", "--no-cache", "--select", "F821", *targets],
        cwd=ROOT, capture_output=True, text=True, check=False,
    )
    assert completed.returncode == 0, completed.stdout + completed.stderr