Jyotisha/tests/test_tajika_mudda.py

"""Mudda Dasha from the natal Moon nakshatra, independent of the year lord (BUG-1216).

Astrologer round-5 ruling 2026-10-03, Q7: first lord = (completed years +
natal Moon nakshatra number 1..27 - 2) mod 9; remainder 1 Sun, 2 Moon,
3 Mars, 4 Rahu, 5 Jupiter, 6 Saturn, 7 Mercury, 8 Ketu, 0 Venus.
"""

from __future__ import annotations

import ast
import hashlib
import inspect
import subprocess
import sys
from datetime import datetime
from pathlib import Path

import pytest

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

from scripts import tajika  # noqa: E402
from scripts.tajika_mudda import annual_mudda_dasha  # noqa: E402

REMAINDER_TO_LORD = {1: "Sun", 2: "Moon", 3: "Mars", 4: "Rahu", 5: "Jupiter", 6: "Saturn", 7: "Mercury", 8: "Ketu", 0: "Venus"}
VIMSHOTTARI = ["Ketu", "Venus", "Sun", "Moon", "Mars", "Rahu", "Jupiter", "Saturn", "Mercury"]
NAK = 360.0 / 27.0
START = datetime(2026, 3, 1, 10, 30, 0)
UPSTREAM = Path("/workspace/yinduzhanxing")
UPSTREAM_SHA = "c7117f92"


@pytest.mark.parametrize("years", [0, 1, 7, 26, 45, 80])
@pytest.mark.parametrize("nakshatra", list(range(1, 28)))
def test_first_lord_follows_the_round5_formula(years: int, nakshatra: int) -> None:
    moon = (nakshatra - 1) * NAK + NAK * 0.4
    result = annual_mudda_dasha(annual_start=START, completed_years=years, natal_moon_longitude=moon)
    expected = REMAINDER_TO_LORD[(years + nakshatra - 2) % 9]
    assert result["status"] == "partial_verified"
    assert result["natal_moon_nakshatra_number"] == nakshatra
    assert result["starting_lord"] == expected
    assert result["periods"][0]["lord"] == expected


def test_record_fields_and_year_window() -> None:
    moon = 3 * NAK + NAK * 0.25  # nakshatra 4, 75% of the first lord left
    result = annual_mudda_dasha(annual_start=START, completed_years=30, natal_moon_longitude=moon)
    for key in ("annual_start", "annual_end", "annual_duration_days", "completed_years",
                "natal_moon_longitude", "balance_profile", "year_length_basis"):
        assert key in result, key
    assert result["balance_profile"] == "moon_at_birth_time"
    assert result["year_length_basis"] == "solar_year_365_25"
    assert result["annual_duration_days"] == 365.25
    assert result["independent_of_year_lord"] is True
    periods = result["periods"]
    # Nine lords in Vimshottari order, cropped to the year: the running
    # remainder of the first lord, eight full lords, then the first lord again
    # up to the year end.
    first = VIMSHOTTARI.index(result["starting_lord"])
    assert [row["lord"] for row in periods] == [VIMSHOTTARI[(first + i) % 9] for i in range(10)]
    assert periods[0]["start"] == START.isoformat(sep=" ")
    assert periods[-1]["end"] == result["annual_end"]
    assert sum(row["months"] for row in periods) == pytest.approx(12.0, abs=0.05)
    assert periods[0]["duration_days"] == pytest.approx(periods[0]["full_duration_days"] * 0.75, rel=1e-9)
    for left, right in zip(periods, periods[1:]):
        assert left["end"] == right["start"]


def test_360_day_year_is_named_separately() -> None:
    solar = annual_mudda_dasha(annual_start=START, completed_years=30, natal_moon_longitude=100.0)
    savana = annual_mudda_dasha(annual_start=START, completed_years=30, natal_moon_longitude=100.0, year_length="savana_360")
    assert savana["year_length_basis"] == "savana_360"
    assert savana["annual_duration_days"] == 360.0
    assert savana["starting_lord"] == solar["starting_lord"]
    assert savana["annual_end"] != solar["annual_end"]
    assert annual_mudda_dasha(annual_start=START, completed_years=30, natal_moon_longitude=100.0, year_length="x")["status"] == "blocked"


def test_blocked_without_inputs() -> None:
    assert annual_mudda_dasha(annual_start=None, completed_years=3, natal_moon_longitude=10.0)["status"] == "blocked"
    assert annual_mudda_dasha(annual_start=START, completed_years=None, natal_moon_longitude=10.0)["status"] == "blocked"
    assert annual_mudda_dasha(annual_start=START, completed_years=3, natal_moon_longitude=None)["status"] == "blocked"


FICTIONAL = {"birth": {"date": "1990-01-01", "time": "12:00:00", "latitude": 39.9, "longitude": 116.4, "utc_offset": "+08:00"},
             "settings": {"ayanamsa": "raman"}}


def test_changing_only_the_year_lord_leaves_mudda_unchanged(monkeypatch) -> None:
    """Natal nakshatra, completed years, year start, year length and balance
    mode fixed; only the year lord changes -> the Mudda table is identical."""
    import tajika_year_lord as year_lord_module  # the name solar_return imports
    from scripts.solar_return import solar_return_full_report

    def run(forced_lord):
        real = year_lord_module.select_annual_year_lord

        def fake(sr, **kwargs):
            data = real(sr, **kwargs)
            if forced_lord:
                data["year_lord"] = forced_lord
            return data

        monkeypatch.setattr(year_lord_module, "select_annual_year_lord", fake)
        return solar_return_full_report(1990, 1, 1, 12, 0, 39.9, 116.4, 8.0, 2026, ayanamsa_name="raman")

    baseline = run(None)
    tables = []
    for lord in ("Sun", "Saturn", "Venus"):
        report = run(lord)
        assert report["year_lord"]["year_lord"] == lord
        tables.append(report["mudda_dasha"])
    assert baseline["mudda_dasha"]["status"] == "partial_verified"
    for table in tables:
        assert table == baseline["mudda_dasha"]


def test_both_producers_give_the_same_mudda_and_the_card_uses_its_dates() -> None:
    from scripts.annual_tajika_pack import build_annual_tajika_pack
    from scripts.consultation_native_layers import _mudda_segments

    pack = build_annual_tajika_pack({**FICTIONAL, "target_year": 2026})
    field = pack["mudda_dasha"]
    assert field["status"] == "partial_verified"
    values = [row.get("value") or row.get("data") for row in field["producers_checked"]]
    data = field["data"]
    assert data["independent_of_year_lord"] is True
    windows = pack["monthly_windows"]
    assert [row["lord"] for row in windows] == [row["lord"] for row in data["periods"]]
    assert all(row["start"] and row["end"] for row in windows)
    start = datetime.fromisoformat(data["annual_start"])
    end = datetime.fromisoformat(data["annual_end"])
    rows = _mudda_segments(data["periods"], start, end, data["annual_start"])
    assert [row["lord"] for row in rows] == [row["lord"] for row in data["periods"]]
    assert rows[0]["start"] == start.strftime("%Y-%m-%d")
    assert rows[-1]["end"] == end.strftime("%Y-%m-%d")
    assert values  # producers recorded


def test_cmd_tajika_mudda_matches_the_solar_return_producer() -> None:
    from types import SimpleNamespace

    from scripts.jyotish_engine import cmd_tajika
    from scripts.solar_return import solar_return_full_report

    args = SimpleNamespace(year=1990, month=1, day=1, hour=12, minute=0, second=0, lat=39.9, lon=116.4, tz=8.0,
                           age=36, mode="all", ayanamsa="raman", node_mode="mean")
    tajika_report = cmd_tajika(args)
    solar = solar_return_full_report(1990, 1, 1, 12, 0, 39.9, 116.4, 8.0, 2026, ayanamsa_name="raman")
    assert tajika_report["mudda_dasha"] == solar["mudda_dasha"]


def test_no_product_path_calls_the_year_lord_started_mudda() -> None:
    callers = [path.name for path in sorted((ROOT / "scripts").glob("*.py"))
               if path.name != "tajika.py" and "calc_mudda_dasha(" in path.read_text(encoding="utf-8")]
    assert callers == []


def _function_source(source: str, name: str) -> str:
    tree = ast.parse(source)
    node = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == name)
    lines = source.splitlines()
    return "\n".join(lines[node.lineno - 1:node.end_lineno])


# sha256 of the function text as ported from upstream c7117f92.
PORTED_SHA256 = {
    "calc_mudda_with_dasha_balance": "8569dd862cf575ac8a670115206c68cd79c5b51bf53113c07bd0e21b07f0560f",
}


@pytest.mark.parametrize("name", sorted(PORTED_SHA256))
def test_ported_function_is_verbatim(name: str) -> None:
    ours = _function_source(inspect.getsource(tajika), name)
    assert hashlib.sha256(ours.encode("utf-8")).hexdigest() == PORTED_SHA256[name]
    if not (UPSTREAM / ".git").exists():
        pytest.skip("upstream checkout not on this machine")
    upstream = subprocess.run(["git", "show", f"{UPSTREAM_SHA}:scripts/tajika.py"], cwd=UPSTREAM,
                              capture_output=True, text=True, check=False)
    if upstream.returncode != 0:
        pytest.skip("upstream commit not available")
    assert ours == _function_source(upstream.stdout, name)