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) quality = next(item for item in probes if item["source"] == "known_event_quality") self.assertEqual(quality["role"], "distinguish") self.assertEqual(quality["year"], 2015) self.assertIn("年份锁定", quality["user_meaning"]) self.assertIn("请写成", quality["user_meaning"]) self.assertIn("发挥失常", quality["user_meaning"]) self.assertNotIn("更像哪一件", quality["user_meaning"]) self.assertNotIn("05:14", quality["user_meaning"]) 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", row["user_meaning"]) self.assertGreater(row["information_gain"], 0) self.assertTrue(row["expected_outcomes"]) 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)) def test_encoded_exam_quality_does_not_fill_probe_slots(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(events=[ { "id": "00000000-0000-4000-8000-000000000001", "domain": "education", "summary": "2015 年第一次参加高考，发挥失利", "date": "2015-06-01", "precision": "year", }, { "id": "00000000-0000-4000-8000-000000000002", "domain": "education", "summary": "2016 年复读一年后再次参加高考", "date": "2016-06-01", "precision": "year", }, { "id": "00000000-0000-4000-8000-000000000003", "domain": "education", "summary": "2016 年 9 月进入大学", "date": "2016-09-01", "precision": "month", }, ]), 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.assertFalse(any( item["source"] == "known_event_quality" and item["year"] in {2015, 2016} for item in probes )) self.assertTrue(any(item["source"] in {"dasha_activation", "dasha_boundary"} for item in probes)) self.assertTrue(any(item["domain"] == "relocation" for item in probes)) self.assertLessEqual(len(probes), 3) if __name__ == "__main__": unittest.main()