Jyotisha/tests/test_rectification_event_probes.py

from __future__ import annotations

import unittest
from datetime import date, datetime

from scripts.rectification.event_probes import discriminating_event_probes
from scripts.rectification.refinement_packet import window_scan

PLANETS = {
    "Sun": 12.0,
    "Moon": 100.0,
    "Mars": 40.0,
    "Mercury": 20.0,
    "Jupiter": 80.0,
    "Venus": 50.0,
    "Saturn": 200.0,
    "Rahu": 310.0,
    "Ketu": 130.0,
}


def _varga(asc: int, planet_sign: int) -> dict:
    return {
        "Ascendant": {"sign_idx": asc},
        **{name: {"sign_idx": planet_sign} for name in PLANETS},
    }


def _context(
    time: str,
    *,
    d4_asc: int,
    sun_house: int,
    sun_varga_sign: int,
    moon: float = 100.0,
    missing_moon: bool = False,
) -> dict:
    hour, minute = (int(part) for part in time.split(":"))
    planets = {**PLANETS, "Moon": moon}
    natal_planets = {
        name: {"house": sun_house if name != "Moon" else 4, "lon": lon}
        for name, lon in planets.items()
    }
    return {
        "candidate_at": datetime(1997, 8, 8, hour, minute),
        "chart": {"ascendant": {"lon": 10.0, "sign": "Aries"}, "planets": natal_planets},
        "planet_longitudes": {name: lon for name, lon in planets.items() if not (missing_moon and name == "Moon")},
        "ascendant_index": 0,
        "varga_charts": {
            "D4": _varga(d4_asc, sun_varga_sign),
            "D9": _varga(1, 1),
            "D10": _varga(1, 1),
            "D5": _varga(1, 1),
            "D24": _varga(1, 1),
            "D12": _varga(1, 1),
            "D7": _varga(1, 1),
            "D3": _varga(1, 1),
        },
        "arudha_padas": {},
        "feature": {
            "time": time,
            "ascendant_sign_index": 0,
            "varga_ascendants": {"D4": d4_asc, "D9": 1, "D10": 1, "D5": 1},
        },
    }


def _request(**extra: object) -> dict:
    return {
        "birth_date": "1997-08-08",
        "events": [],
        **extra,
    }


class EventProbesTest(unittest.TestCase):
    def test_missing_birth_date_emits_no_probes(self) -> None:
        built = {"static_contexts": [_context("05:13", d4_asc=1, sun_house=4, sun_varga_sign=3)]}
        probes = discriminating_event_probes(
            {"events": []},
            built,
            scan=window_scan(built),
            candidate_times=["05:13"],
            representative_time="05:13",
            today=date(2026, 8, 22),
        )
        self.assertEqual(probes, [])

    def test_known_gaokao_event_asks_quality_not_existence(self) -> None:
        built = {
            "static_contexts": [
                _context("05:13", d4_asc=1, sun_house=10, sun_varga_sign=9),
                _context("05:14", d4_asc=2, sun_house=10, sun_varga_sign=9),
            ]
        }
        probes = discriminating_event_probes(
            _request(events=[{
                "id": "00000000-0000-4000-8000-000000000001",
                "domain": "education",
                "summary": "2015年高考",
                "date": "2015-06-01",
                "precision": "year",
            }]),
            built,
            scan=window_scan(built),
            candidate_times=["05:13", "05:14"],
            representative_time="05:13",
            precision_current="d5_refine",
            today=date(2026, 8, 22),
        )
        self.assertTrue(probes)
        self.assertEqual(probes[0]["source"], "known_event_quality")
        self.assertEqual(probes[0]["role"], "distinguish")
        self.assertEqual(probes[0]["year"], 2015)
        self.assertIn("年份锁定", probes[0]["user_meaning"])
        self.assertIn("请写成", probes[0]["user_meaning"])
        self.assertIn("发挥失常", probes[0]["user_meaning"])
        self.assertNotIn("更像哪一件", probes[0]["user_meaning"])
        self.assertNotIn("05:14", str(probes))
        self.assertNotIn("points", str(probes))

    def test_age_band_fallback_without_full_charts(self) -> None:
        built = {
            "static_contexts": [
                {"feature": {"time": "05:13", "varga_ascendants": {"D4": 1, "D9": 1, "D10": 1}}},
                {"feature": {"time": "05:14", "varga_ascendants": {"D4": 2, "D9": 1, "D10": 1}}},
            ]
        }
        probes = discriminating_event_probes(
            _request(),
            built,
            scan=window_scan(built),
            candidate_times=["05:13", "05:14"],
            representative_time="05:13",
            precision_current="d4_refine",
            today=date(2026, 8, 22),
        )
        self.assertTrue(probes)
        self.assertEqual(probes[0]["source"], "age_band")
        self.assertEqual(probes[0]["role"], "reverse_verify")
        self.assertEqual(probes[0]["domain"], "relocation")
        self.assertEqual(probes[0]["year"], 2018)
        self.assertIn("年份锁定", probes[0]["user_meaning"])
        self.assertIn("请写成", probes[0]["user_meaning"])
        self.assertIn("搬家", probes[0]["user_meaning"])
        self.assertFalse(probes[0]["unique_minute_claim"])
        self.assertNotIn("05:14", probes[0]["user_meaning"])

    def test_d4_activation_difference_asks_move_in_that_year(self) -> None:
        built = {
            "static_contexts": [
                _context("05:13", d4_asc=0, sun_house=4, sun_varga_sign=3),
                _context("05:14", d4_asc=1, sun_house=10, sun_varga_sign=9),
            ]
        }
        probes = discriminating_event_probes(
            _request(),
            built,
            scan=window_scan(built),
            candidate_times=["05:13", "05:14"],
            representative_time="05:13",
            precision_current="d4_refine",
            today=date(2026, 8, 22),
        )
        self.assertTrue(probes)
        row = next(item for item in probes if item["domain"] == "relocation")
        self.assertIn(row["source"], {"dasha_activation", "dasha_boundary"})
        self.assertEqual(row["role"], "reverse_verify")
        self.assertIn("年份锁定", row["user_meaning"])
        self.assertIn("请写成", row["user_meaning"])
        self.assertIn("搬家", row["user_meaning"])
        self.assertIn(str(row["year"]), row["year_label"])
        self.assertNotIn("更像哪一件", row["user_meaning"])
        self.assertNotIn("points", str(row))
        self.assertNotIn("05:13", str(row))
        self.assertEqual(row["tracks"], ["vimshottari", "narayana"])
        self.assertFalse(row["unique_minute_claim"])

    def test_same_calendar_year_shift_is_not_a_boundary_year(self) -> None:
        built = {
            "static_contexts": [
                _context("05:13", d4_asc=1, sun_house=10, sun_varga_sign=9, moon=100.0),
                _context("05:14", d4_asc=1, sun_house=10, sun_varga_sign=9, moon=100.01),
            ]
        }
        probes = discriminating_event_probes(
            _request(),
            built,
            scan=window_scan(built),
            candidate_times=["05:13", "05:14"],
            representative_time="05:13",
            precision_current="d4_refine",
            today=date(2026, 8, 22),
        )
        self.assertTrue(all(item["source"] != "dasha_boundary" for item in probes))

    def test_missing_narayana_inputs_do_not_claim_dasha_year(self) -> None:
        built = {
            "static_contexts": [
                _context("05:13", d4_asc=0, sun_house=4, sun_varga_sign=3, missing_moon=True),
                _context("05:14", d4_asc=1, sun_house=10, sun_varga_sign=9, missing_moon=True),
            ]
        }
        probes = discriminating_event_probes(
            _request(),
            built,
            scan=window_scan(built),
            candidate_times=["05:13", "05:14"],
            representative_time="05:13",
            precision_current="d4_refine",
            today=date(2026, 8, 22),
        )
        self.assertTrue(probes)
        self.assertTrue(all(item["source"] == "age_band" for item in probes))


if __name__ == "__main__":
    unittest.main()