from __future__ import annotations import unittest from datetime import date, datetime from unittest.mock import patch from scripts.rectification.event_probes import ( MIN_BOUNDARY_DAYS, REFRESH_MIN_BOUNDARY_DAYS, _boundary_windows, 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, } # Relocation age_lo=18 → 2015; asked relocation.2015 blocks 2014–2016. # Relationship age_lo=21 → 2018; asked relationship.2023 / event 2021 leave 2019 open. ADULT_YEAR = 2019 TIMES = ["04:48", "04:53", "04:59", "05:06", "05:07"] ASKED = [ "career.2020", "career.2023", "education.2016", "education.2020", "relationship.2023", "relocation.2015", ] def _varga(asc: int) -> dict: return { "Ascendant": {"sign_idx": asc}, **{name: {"sign_idx": 1} for name in PLANETS}, } def _context(time: str, *, moon: float, d4_asc: int, d9_asc: int, d10_asc: int) -> dict: hour, minute = (int(part) for part in time.split(":")) planets = {**PLANETS, "Moon": moon} return { "candidate_at": datetime(1997, 8, 8, hour, minute), "chart": { "ascendant": {"lon": 10.0, "sign": "Aries"}, "planets": { name: {"house": 4 if name == "Moon" else 10, "lon": lon} for name, lon in planets.items() }, }, "planet_longitudes": dict(planets), "ascendant_index": 0, "varga_charts": { "D4": _varga(d4_asc), "D9": _varga(d9_asc), "D10": _varga(d10_asc), "D5": _varga(1), "D24": _varga(1), "D12": _varga(1), "D7": _varga(1), "D3": _varga(1), }, "arudha_padas": {}, "feature": { "time": time, "ascendant_sign_index": 0, "varga_ascendants": { "D4": d4_asc, "D9": d9_asc, "D10": d10_asc, "D5": 1, "D24": 1, "D12": 1, }, }, } def _gate_events() -> list[dict]: return [ { "id": "00000000-0000-4000-8000-000000000011", "domain": "education", "event_kind": "education_start", "summary": "入学", "date": "2014-09-01", "precision": "month", }, { "id": "00000000-0000-4000-8000-000000000012", "domain": "education", "event_kind": "education_completion", "summary": "毕业", "date": "2017-06-01", "precision": "month", }, { "id": "00000000-0000-4000-8000-000000000013", "domain": "career", "event_kind": "career_entry", "summary": "入职", "date": "2018-07-01", "precision": "month", }, { "id": "00000000-0000-4000-8000-000000000014", "domain": "relationship", "event_kind": "relationship_start", "summary": "相识", "date": "2021-08-01", "precision": "month", }, ] def _built(*, d4_left: int = 0, d4_right: int = 1, d9_left: int = 1, d9_right: int = 1, d10_left: int = 1, d10_right: int = 1) -> dict: early = dict(moon=100.0, d4_asc=d4_left, d9_asc=d9_left, d10_asc=d10_left) late = dict(moon=101.0, d4_asc=d4_right, d9_asc=d9_right, d10_asc=d10_right) return { "static_contexts": [ _context("04:48", **early), _context("04:53", **early), _context("04:59", **early), _context("05:06", **late), _context("05:07", **late), ] } def _score_months(target_year: int, months: set[int], rule: str = "vim_ad_domain_lord"): def fake_score(context: dict, *, birth_date: str, domain: str, year: int, month: int | None = None) -> dict: del birth_date, domain early = str(context.get("feature", {}).get("time") or "")[:5] < "05:00" if year == target_year and month in months: return {"rule_ids": [rule] if early else ["no_domain_activation"]} return {"rule_ids": ["no_domain_activation"]} return fake_score def _probes(*, refresh: bool, built: dict, vim, narayana, score): from scripts.rectification import event_probes as probes_mod request = { "birth_date": "1997-08-08", "events": _gate_events(), "asked_probe_keys": ASKED, "refresh_probes": refresh, } with ( patch.object(probes_mod, "_vim_start_dates", side_effect=vim), patch.object(probes_mod, "_narayana_start_dates", side_effect=narayana), patch.object(probes_mod, "_score_year", side_effect=score), ): return discriminating_event_probes( request, built, scan=window_scan(built), candidate_times=TIMES, representative_time="04:53", today=date(2026, 9, 13), ) class RefreshR3R4Test(unittest.TestCase): def test_default_boundary_threshold_stays_45(self) -> None: self.assertEqual(MIN_BOUNDARY_DAYS, 45) self.assertEqual(REFRESH_MIN_BOUNDARY_DAYS, 30) left = [date(ADULT_YEAR, 3, 1)] right = [date(ADULT_YEAR, 4, 10)] self.assertEqual(_boundary_windows(left, right), []) self.assertEqual( [item.month for item in _boundary_windows(left, right, min_days=30)], [3, 4], ) def test_first_round_drops_a_40_day_same_year_window(self) -> None: def fake_vim(_birth: str, moon: float, _lo: int, _hi: int, **_kwargs: object) -> list[date]: return [date(ADULT_YEAR, 3, 1)] if moon <= 100.0 else [date(ADULT_YEAR, 4, 10)] def fake_narayana(_asc: int, _planets: dict, _birth: str, _lo: int, _hi: int) -> list[date]: return [] score = _score_months(ADULT_YEAR, {3, 4}) built = _built() first = _probes(refresh=False, built=built, vim=fake_vim, narayana=fake_narayana, score=score) refresh = _probes(refresh=True, built=built, vim=fake_vim, narayana=fake_narayana, score=score) self.assertFalse(any(item.get("year") == ADULT_YEAR for item in first)) self.assertTrue(any( item.get("year") == ADULT_YEAR and item.get("source") == "dasha_boundary" for item in refresh ), [item.get("semantic_key") for item in refresh]) def test_refresh_adds_pratyantar_windows_first_round_does_not(self) -> None: def fake_vim(_birth: str, moon: float, _lo: int, _hi: int, *, include_pratyantar: bool = False) -> list[date]: md = [date(2011, 1, 1)] if not include_pratyantar: return md extra = date(ADULT_YEAR, 2, 1) if moon <= 100.0 else date(ADULT_YEAR, 4, 1) return [*md, extra] def fake_narayana(_asc: int, _planets: dict, _birth: str, _lo: int, _hi: int) -> list[date]: return [] score = _score_months(ADULT_YEAR, {2, 4}) built = _built() first = _probes(refresh=False, built=built, vim=fake_vim, narayana=fake_narayana, score=score) refresh = _probes(refresh=True, built=built, vim=fake_vim, narayana=fake_narayana, score=score) self.assertFalse(any(item.get("year") == ADULT_YEAR for item in first)) self.assertTrue( any(item.get("year") == ADULT_YEAR for item in refresh), [item.get("semantic_key") for item in refresh], ) def test_refresh_adds_d9_narayana_when_natal_lagna_matches(self) -> None: def fake_vim(_birth: str, _moon: float, _lo: int, _hi: int, **_kwargs: object) -> list[date]: return [date(2011, 1, 1)] def fake_narayana(asc: int, _planets: dict, _birth: str, _lo: int, _hi: int) -> list[date]: if asc == 1: return [date(ADULT_YEAR, 6, 1)] if asc == 2: return [date(ADULT_YEAR, 7, 21)] return [date(2011, 1, 1)] score = _score_months(ADULT_YEAR, {6, 7}, rule="narayana_ad_domain_lord") built = _built(d4_left=0, d4_right=0, d9_left=1, d9_right=2) first = _probes(refresh=False, built=built, vim=fake_vim, narayana=fake_narayana, score=score) refresh = _probes(refresh=True, built=built, vim=fake_vim, narayana=fake_narayana, score=score) self.assertFalse(any(item.get("year") == ADULT_YEAR for item in first)) self.assertTrue( any(item.get("year") == ADULT_YEAR for item in refresh), [item.get("semantic_key") for item in refresh], ) if __name__ == "__main__": unittest.main()