Jyotisha/tests/test_consultation_native_layers.py

"""Native technique layers in the consultation output (TASK-consult-evidence-card-v2-20260927 T1).

Real engine runs on a public AA chart (the evidence-card research charts).
Each layer is compared with its native producer called directly, the annual
pack with `build_annual_tajika_pack`, and the rest of the response with a run
where the new layer is switched off: only the new key may differ.
"""

from __future__ import annotations

import json
import re
import sys
from datetime import datetime
from pathlib import Path
from types import SimpleNamespace

import pytest

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

import consult_evidence_card_run as runner  # noqa: E402
from consult_evidence_card_lib import load_public_charts  # noqa: E402
import scripts.consultation_native_layers as native  # noqa: E402

QUESTIONS = {
    "marriage": "我的关系模式是什么？",
    "annual": "未来一年重点是什么？",
    "timing": "接下来一年哪些时间点重要？",
}


@pytest.fixture(scope="module", autouse=True)
def _no_external_vedastro():
    runner._block_external_vedastro()


def _chart(index: int = 0) -> dict:
    return load_public_charts(ROOT)[index]


def _run(domain: str, index: int = 0) -> dict:
    return runner._run_workflow(runner._workflow_body(_chart(index), {
        "id": domain, "domain": domain, "question": QUESTIONS[domain],
    }))


def _layers(workflow: dict) -> dict:
    layers = workflow["chart"]["consultation_native_layers"]
    assert layers["schema"] == native.SCHEMA
    assert layers["status"] == "executed"
    return layers


def _birth_args(workflow: dict, house: int) -> SimpleNamespace:
    birth = workflow["chart"]["birth"]
    body = _chart()["body"]
    return SimpleNamespace(
        year=body["year"], month=body["month"], day=body["day"], hour=body["hour"], minute=body["minute"],
        second=0, lat=body["lat"], lon=body["lon"], tz=float(body["tz"]), node_mode=birth["node_mode"],
        ayanamsa=birth["ayanamsa_name"], date=body["today"], house=house,
    )


def test_quick_gate_runs_this_file() -> None:
    import run_quality_gate

    assert "tests/test_consultation_native_layers.py" in run_quality_gate.CORE_PYTEST_TARGETS


def test_every_layer_is_the_native_producers_own_output() -> None:
    workflow = _run("marriage")
    layers = _layers(workflow)
    chart = workflow["chart"]
    modules = chart["modules"]
    engine = native._load_engine()

    d9 = modules["varga_full"]["D9_Navamsa"]
    summary = layers["d9_summary"]
    assert summary["lagna"] == d9["ascendant"]["sign"]
    expected_vargottama = engine._calc_vargottama(chart["planets"], {"D9_Navamsa": d9["planets"]})
    context = engine._build_dignity_context(d9["planets"])
    for name, row in summary["planets"].items():
        assert row["d9_sign"] == d9["planets"][name]["sign"]
        assert row["vargottama"] is expected_vargottama[name]["is_vargottama"]
        level = engine._get_dignity_level(name, d9["planets"][name]["sign"], d9["planets"][name]["degree"], context)
        assert row["d9_dignity"] == engine.DIGNITY_LABELS[level]
    assert summary["vargottama"] == [name for name, row in summary["planets"].items() if row["vargottama"]]
    assert len(summary["planets"]) == 9

    from punarphoo import detect_natal_punarphoo

    natal = detect_natal_punarphoo(chart["planets"])
    assert layers["punarphoo"]["status"] == "observation_only"
    assert layers["punarphoo"]["natal_present"] == natal["natal_present"]
    assert layers["punarphoo"]["orb_deg"] == natal["orb_deg"]
    assert "not marriage" in layers["punarphoo"]["claim_boundary"]

    vivah = engine._calc_vivah_saham(
        {name: {"degree": row["lon"]} for name, row in chart["planets"].items()}, chart["ascendant"]["lon"],
    )
    assert layers["vivah_saham"]["longitude"] == vivah["saham_lon"]
    assert layers["vivah_saham"]["house"] == vivah["saham_house"]
    assert layers["vivah_saham"]["sign"] == vivah["saham_sign"]

    assert layers["day_night"]["is_day"] is modules["gulika"]["daynight_evidence"]["is_daytime"]

    seventh = engine.cmd_double_transit_pac(_birth_args(workflow, 7))
    double = layers["double_transit"]
    assert double["house_7_summary"] == seventh["summary"]
    assert [row["target"] for row in double["conclusions"] if row["house"] == 7] == [
        row["target"] for row in seventh["double_transit"]
    ]
    assert set(double["karaka_targets"]) == {"DK", "UL"}
    assert double["karaka_targets"]["DK"]["target"] == f"DK({modules['jaimini']['chara_karakas']['DK']['planet']})"
    assert double["karaka_targets"]["UL"]["target"] == f"UL({modules['arudha_padas']['padas']['UL']['sign']})"
    # Conclusions, not raw degrees.
    assert not re.search(r"\d+\.\d+°", json.dumps(double["conclusions"], ensure_ascii=False))

    slow = layers["slow_transits"]
    assert set(slow["current"]) == {"Jupiter", "Saturn", "Rahu", "Ketu"}
    sav = modules["ashtakavarga"]["sav"]["scores"]
    for body in ("Jupiter", "Saturn"):
        row = slow["current"][body]
        assert row["sav"] == sav[row["sign"]]
        assert row["bav"] == modules["ashtakavarga"]["bav"][body]["bindus"][native.SIGNS.index(row["sign"])]
    from ephemeris_events import build_ephemeris_events

    scan = build_ephemeris_events({
        "start_date": slow["window"]["start"], "end_date": slow["window"]["end"],
        "ayanamsa": chart["birth"]["ayanamsa_name"], "bodies": ["jupiter", "saturn"],
    })
    expected = [(row["kind"], row["date"], row["body"]) for row in scan["events"]]
    got = [(row["kind"], row["date"], row["body"]) for row in slow["key_dates"] if row["body"] in ("jupiter", "saturn")]
    assert got == expected
    assert slow["houses_in_window"]["Jupiter"][0] == slow["current"]["Jupiter"]["natal_house"]

    ak = modules["jaimini"]["chara_karakas"]["AK"]["planet"]
    assert layers["karakamsha"]["atmakaraka"] == ak
    assert layers["karakamsha"]["karakamsha_sign"] == d9["planets"][ak]["sign"]
    assert layers["argala"]["summary"]["supported_count"] == len(layers["argala"]["summary"]["supported_houses"])
    assert set(layers["dispositor_chains"]["chains"]) >= {"Sun", "Moon"}
    assert layers["inter_chart_linkage"]["planets"]["Sun"]["D9"]["sign"] == d9["planets"]["Sun"]["sign"]
    assert layers["moon_transit"]["sign"] in native.SIGNS
    assert layers["annual_tajika"]["status"] == "not_applicable"


def test_annual_route_attaches_the_annual_pack_verbatim() -> None:
    workflow = _run("annual")
    annual = _layers(workflow)["annual_tajika"]
    assert annual["status"] == "parameter_sensitive"
    assert "parameter_sensitive" in annual["boundary"]
    from annual_tajika_pack import build_annual_tajika_pack

    body = _chart()["body"]
    payload = native._annual_payload({**body, "tz": float(body["tz"]), "ayanamsa": "raman", "node_mode": "mean"}, annual["target_year"])
    pack = build_annual_tajika_pack(payload)
    assert annual["annual_lagna"] == pack["annual_chart"]["data"]["asc_sign"]
    assert annual["varshesha"] == pack["year_lord"]["data"]["year_lord"]
    assert annual["muntha"]["sign"] == pack["muntha"]["data"]["muntha_sign"]
    assert annual["muntha"]["lord"] == pack["muntha"]["data"]["muntha_lord"]
    periods = pack["mudda_dasha"]["periods"]
    assert [(row["lord"], row["months"]) for row in annual["mudda_dasha"]] == [(row["lord"], row["months"]) for row in periods]
    segments = annual["mudda_dasha"]
    assert segments[0]["start"] == annual["varsha_year"]["start"]
    assert segments[-1]["end"] == annual["varsha_year"]["end"]
    for before, after in zip(segments, segments[1:]):
        assert before["end"] == after["start"]
    assert annual["varsha_year"]["start"] <= body["today"] < annual["varsha_year"]["end"]
    focus = {"Sun", "Moon", annual["varshesha"]}
    assert annual["tajika_aspects"], "the annual chart has Tajika aspects for the Sun, Moon or year lord"
    for row in annual["tajika_aspects"]:
        assert focus.intersection(row["planets"])
        assert row["name"] in {"Ithasala_candidate", "Easarapha_candidate"}


def test_an_unreliable_annual_pack_is_blocked_and_never_filled_with_natal_data(monkeypatch) -> None:
    import annual_tajika_pack

    real = annual_tajika_pack.build_annual_tajika_pack

    def conflicted(payload, **kwargs):
        pack = real(payload, **kwargs)
        pack["muntha"] = {"status": "conflict", "field": "muntha", "reason": "same_profile_annual_producers_disagree"}
        return pack

    monkeypatch.setattr(annual_tajika_pack, "build_annual_tajika_pack", conflicted)
    annual = _layers(_run("annual"))["annual_tajika"]
    assert annual["status"] == "blocked"
    assert "muntha" in annual["reason"]
    for key in ("annual_lagna", "varshesha", "muntha", "mudda_dasha", "tajika_aspects"):
        assert key not in annual


def test_a_failing_producer_blocks_its_own_layer_only(monkeypatch) -> None:
    engine = native._load_engine()

    def broken(*args, **kwargs):
        raise RuntimeError("fixture")

    monkeypatch.setattr(engine, "cmd_double_transit_pac", broken)
    workflow = _run("timing")
    layers = _layers(workflow)
    assert workflow["success"] is True
    assert layers["double_transit"] == {"status": "blocked", "reason": "RuntimeError"}
    assert layers["d9_summary"]["status"] == "executed"
    assert layers["slow_transits"]["status"] == "executed"


def test_the_new_layer_leaves_every_existing_output_unchanged(monkeypatch) -> None:
    # The first run of a chart fills the chart cache; a cache hit re-reads the
    # stored chart, whose VedAstro request manifest and key order differ from a
    # fresh compute. Compare two cache-hit runs.
    _run("marriage", index=1)
    with_layers = _run("marriage", index=1)
    monkeypatch.setattr(native, "attach_native_consultation_layers", lambda chart, **_: chart)
    without = _run("marriage", index=1)
    assert "consultation_native_layers" not in without["chart"]
    volatile = {"runtime_cache", "vedastro_gateway"}

    def strip(value):
        if isinstance(value, dict):
            return {key: strip(item) for key, item in value.items() if key not in volatile and key != "consultation_native_layers"}
        if isinstance(value, list):
            return [strip(item) for item in value]
        return value

    assert json.dumps(strip(with_layers), sort_keys=True, default=str) == json.dumps(strip(without), sort_keys=True, default=str)


def test_mudda_segment_dates_split_the_solar_year_by_the_native_months() -> None:
    start, end = datetime(2026, 2, 25), datetime(2027, 2, 25)
    rows = native._mudda_segments([{"lord": "Sun", "months": 6.0}, {"lord": "Moon", "months": 6.0}], start, end)
    assert rows == [
        {"lord": "Sun", "months": 6.0, "start": "2026-02-25", "end": "2026-08-26"},
        {"lord": "Moon", "months": 6.0, "start": "2026-08-26", "end": "2027-02-25"},
    ]