Jyotisha/tests/test_upstream_engine_sync_bug1699.py

"""BUG-1699: upstream yinduzhanxing calculation fixes ported into the engine.

Covers the sidecars that keep the frozen rectification files untouched
(``shadbala_chesta_profiles``, ``ashtakavarga_profiles``), the book rule layer,
the annual Raman Saham layer and the new ``consultation_native_layers`` keys.
Birth data is the fictional golden profile shared with the frontend tests
(1990-06-15 12:00 Beijing, raman) or the public AA research chart the
native-layer tests already use.
"""

from __future__ import annotations

import copy
import json
import sys
from pathlib import Path

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))

from scripts.jyotish_api_server import JyotishAPIHandler  # 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")
FROZEN = ("scripts/shadbala.py", "scripts/ashtakavarga.py")


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


def _shadbala(**extra) -> dict:
    return _handler()._compute_shadbala({**BIRTH, "planets": PLANETS, "ascendant": ASCENDANT, **extra})["result"]


def test_quick_gate_runs_this_file() -> None:
    import run_quality_gate

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


# ── Shadbala Chesta profiles ────────────────────────────────────────────────


def test_default_shadbala_gains_profile_keys_without_changing_values() -> None:
    result = _shadbala()
    variants = result["method_variants"]
    assert variants["chesta"] == "bphs_bounded_surya_mean_motion_seeghrochcha"
    candidates = variants["chesta_profile_candidates"]
    ids = [row["profile_id"] for row in candidates["candidate_profiles"]]
    assert ids == [
        "bphs_bounded_surya_mean_motion_seeghrochcha", "pyjhora_jhora_replay_target",
        "raman_luminary_zero", "raman_inferior_planet_assignment", "vp_jain_reference_family",
    ]
    assert len(candidates["planet_rows"]) == 7
    assert variants["chesta_profile_selection"]["selection_status"] == "active_local_profile"
    for key in ("chesta_target_diagnostic", "chesta_target_preview", "chesta_shadow_profile_result", "chesta_profile_result"):
        assert variants[key] is None
    assert result["source_reference_packets"] == ["science_of_light_v1_strength_interpretation_boundary"]

    import shadbala_chesta_profiles

    original = shadbala_chesta_profiles.attach_chesta_profiles
    shadbala_chesta_profiles.attach_chesta_profiles = lambda result, **_: result
    try:
        bare = _shadbala()
    finally:
        shadbala_chesta_profiles.attach_chesta_profiles = original
    assert bare["planets"] == result["planets"]
    assert {k: v for k, v in result["method_variants"].items() if not k.startswith("chesta_")} == {
        k: v for k, v in bare["method_variants"].items() if not k.startswith("chesta_")
    }


def test_raman_luminary_zero_is_reported_next_to_the_default() -> None:
    default = _shadbala()
    result = _shadbala(chesta_profile="raman_luminary_zero")
    assert result["planets"] == default["planets"]  # default values untouched
    profile = result["method_variants"]["chesta_profile_result"]
    assert profile["status"] == "computed"
    rows = {row["planet"]: row for row in profile["rows"]}
    for planet in ("Sun", "Moon"):
        assert rows[planet]["profile_chesta_virupa"] == 0.0
        expected = round(default["planets"][planet]["total_virupas"] - default["planets"][planet]["chesta_bala"], 2)
        assert rows[planet]["profile_total_virupas"] == pytest.approx(max(1.0, expected), abs=0.011)
    for planet in ("Mars", "Mercury", "Jupiter", "Venus", "Saturn"):
        assert rows[planet]["profile_chesta_virupa"] == pytest.approx(default["planets"][planet]["chesta_bala"], abs=0.006)
    assert result["method_variants"]["chesta_shadow_profile_result"]["status"] == "shadow_compare_only"


def test_inferior_planet_swap_only_moves_mercury_and_venus() -> None:
    default = _shadbala()
    rows = {row["planet"]: row for row in _shadbala(chesta_profile="raman_inferior_planet_assignment")[
        "method_variants"]["chesta_profile_result"]["rows"]}
    for planet in ("Sun", "Moon", "Mars", "Jupiter", "Saturn"):
        assert rows[planet]["profile_chesta_virupa"] == pytest.approx(default["planets"][planet]["chesta_bala"], abs=0.006)
    assert rows["Mercury"]["chesta_basis"] == rows["Venus"]["chesta_basis"] == "raman_inferior_planet_assignment_swap"


def test_unknown_profile_is_not_executed() -> None:
    variants = _shadbala(chesta_profile="vp_jain_reference_family")["method_variants"]
    assert variants["chesta_profile_selection"]["selection_status"] == "requested_profile_not_executable_yet"
    assert variants["chesta_profile_result"]["status"] == "not_executable"


# ── Ashtakavarga score fields and PL9 profile ───────────────────────────────


def test_ashtakavarga_score_fields_and_opt_in_pl9_profile() -> None:
    import ashtakavarga as av
    from ashtakavarga_profiles import PL9_BAV_CELL_OVERRIDES, calc_ashtakavarga_profile

    handler = _handler()
    planets, _, asc_idx = handler._normalized_planets_from_body({"planets": PLANETS, "ascendant": ASCENDANT})
    standard = av.calc_ashtakavarga(planets, asc_idx)
    snapshot = copy.deepcopy(standard)
    result = calc_ashtakavarga_profile(planets, asc_idx, "standard", standard=standard)
    assert {k: v for k, v in result.items() if k not in {"profile", "available_profiles", "score_fields"}} == snapshot
    fields = result["score_fields"]
    for sign in av.SIGNS:
        row = fields["sign_scores"][sign]
        assert row["sav_sign_score"] == snapshot["sav"]["scores"][sign]
        assert row["full_with_lagna_score"] == snapshot["sav"]["full_scores_with_lagna"][sign]
    assert fields["house_scores"]["house_1"]["sign"] == av.SIGNS[asc_idx]

    pl9 = calc_ashtakavarga_profile(planets, asc_idx, "pl9", standard=snapshot)
    assert pl9["profile"] == "pl9"
    for planet, cells in PL9_BAV_CELL_OVERRIDES.items():
        for sign, value in cells.items():
            assert pl9["bav"][planet]["bindus"][av.SIGNS.index(sign)] == value
    for index, sign in enumerate(av.SIGNS):
        assert pl9["sav"]["scores"][sign] == sum(pl9["bav"][p]["bindus"][index] for p in av.SEVEN_PLANETS)
    assert snapshot["bav"]["Moon"]["bindus"] == standard["bav"]["Moon"]["bindus"]  # standard not mutated

    api = handler._compute_ashtakavarga({"planets": PLANETS, "ascendant": ASCENDANT, "ashtakavarga_profile": "pl9"})["result"]
    assert api["profile"] == "standard" and api["pl9_profile"]["profile"] == "pl9"
    assert "pl9_profile" not in handler._compute_ashtakavarga({"planets": PLANETS, "ascendant": ASCENDANT})["result"]


def test_frozen_rectification_files_are_not_edited() -> None:
    import subprocess

    base = subprocess.run(
        ["git", "merge-base", "HEAD", "origin/staging"], cwd=ROOT, capture_output=True, text=True, check=False,
    )
    if base.returncode != 0 or not base.stdout.strip():
        pytest.skip("origin/staging not available")
    # Only this branch's own edits count; later staging changes are not ours.
    diff = subprocess.run(
        ["git", "diff", "--name-only", base.stdout.strip(), "--", *FROZEN],
        cwd=ROOT, capture_output=True, text=True, check=False,
    )
    assert diff.stdout.strip() == ""


# ── Book rule layer ─────────────────────────────────────────────────────────


def test_book_rule_layer_runtime_and_route_rules() -> None:
    from book_rule_layer import build_book_rule_layer

    layer = build_book_rule_layer({"marriage"})
    assert layer["status"] == "executed"
    assert sorted(row["rule_id"] for row in layer["runtime_packets"]) == [
        "science_of_light_v1_arudha_reflection", "science_of_light_v1_vimshottari_moon_balance",
    ]
    assert layer["gate_failures"] == []
    ids = {row["rule_id"] for row in layer["route_reference_rules"]}
    for path in ("web_rule_intake_2026_10_06.json", "web_rule_intake_2026_10_06_romance.json"):
        intake = json.loads((ROOT / "references" / path).read_text(encoding="utf-8"))
        assert {row["rule_id"] for row in intake["packets"]} <= ids
    assert all(not row["runtime_usable"] for row in layer["route_reference_rules"])
    assert "never implies Technique Audit execution" in layer["claim_boundary"]


# ── Annual Raman Sahams ─────────────────────────────────────────────────────


def test_annual_pack_carries_raman_sahams_from_the_same_return() -> None:
    from annual_tajika_pack import build_annual_tajika_pack
    from raman_sahams import build_raman_sahams

    pack = build_annual_tajika_pack({
        "birth": {"date": "1990-06-15", "time": "12:00:00", "latitude": 39.9042, "longitude": 116.4074, "utc_offset": "+08:00"},
        "settings": {"ayanamsa": "raman", "node_mode": "mean"},
        "target_year": 2026,
    })
    sahams = pack["raman_sahams"]
    assert sahams["schema"] == "jyotish.raman_sahams.v1"
    assert len(sahams["rows"]) == 37
    assert {row["name"] for row in sahams["rows"]} >= {"Punya", "Vivaha", "Labha"}
    is_day = pack["sahams"]["data"]["daynight_evidence"]["is_daytime"]
    assert sahams["is_daytime"] is is_day


# ── Consultation native layers ──────────────────────────────────────────────


@pytest.fixture(scope="module")
def career_workflow():
    import consult_evidence_card_run as runner
    from consult_evidence_card_lib import load_public_charts

    runner._block_external_vedastro()
    chart = load_public_charts(ROOT)[0]
    return runner._run_workflow(runner._workflow_body(chart, {
        "id": "career", "domain": "career", "question": "我的事业方向是什么？",
    }))


def test_native_layers_carry_the_upstream_layers(career_workflow) -> None:
    from affliction_evidence import collect_affliction_evidence

    chart = career_workflow["chart"]
    layers = chart["consultation_native_layers"]
    for key in ("affliction_evidence", "career_analysis", "sade_sati_timeline", "shadbala_chesta_profiles", "book_rules"):
        assert layers[key]["status"] in {"executed", "partial_verified"}, (key, layers[key])

    affliction = layers["affliction_evidence"]
    assert set(affliction["houses"]) == {"1", "4", "5", "6", "7", "9", "10"}
    career = layers["career_analysis"]
    assert career["career_claim_gate"]["industry_claim_allowed"] is False
    assert career["functional_benefic_malefic"]["functional_malefics"]
    assert career["affliction_evidence"]["target_house"] == 10
    ten = affliction["houses"]["10"]
    direct = collect_affliction_evidence(
        target_house=10, target_lord=ten["target_lord"],
        planets={},  # lord row absent: only the functional-malefic signal can remain
        functional_malefics=affliction["functional_malefics"],
    )
    assert {s["type"] for s in direct["signals"]} <= {s["type"] for s in ten["signals"]}

    timeline = layers["sade_sati_timeline"]
    assert timeline["audit"]["boundary_precision"] == "bisected_ingress_moment_local_minute"
    assert any(row.get("ending_moment_local") for row in timeline["sadhesati_rows"])
    assert "pl9_display_rows" not in timeline

    assert layers["shadbala_chesta_profiles"]["chart_chesta_profile"]
    assert layers["book_rules"]["routes"] == ["career"]


def test_user_industry_opens_the_industry_gate() -> None:
    from career_analysis import analyze_career

    planets = {"Sun": {"house": 10}, "Mercury": {"house": 10}, "Saturn": {"house": 6}}
    closed = analyze_career(planets, "Aries")
    opened = analyze_career(planets, "Aries", user_industry="软件工程", a10={"sign": "Capricorn"})
    assert closed["career_claim_gate"]["industry_claim_allowed"] is False
    assert opened["career_claim_gate"]["industry_claim_allowed"] is True
    assert opened["career_claim_gate"]["public_backstage_claim_allowed"] is True