Jyotisha/tests/test_event_probes_guided_windows.py

from __future__ import annotations

import unittest
from datetime import date, datetime
from functools import lru_cache

from scripts.active_rectification_event_engine import build_candidate_static_context
from scripts.rectification.event_probes import (
    GUIDED_ANY_DOMAIN,
    GUIDED_COLLECT_LIMIT,
    guided_collect_windows,
)

from tests.test_rectification_event_probes import _context, _gate_events, _multi_layer_window_built, _request

#: Public 1990-01-01 Beijing smoke chart, fictional dated events. A real
#: 20-minute search window (12:00–12:20) run through the engine, so the
#: window contract is asserted against engine output, never a hand-made shape.
REAL_REQUEST = {
    "birth_date": "1990-01-01",
    "start_time": "12:00",
    "end_time": "12:20",
    "lat": 39.9,
    "lon": 116.4,
    "tz": 8,
    "events": [
        {
            "id": "00000000-0000-4000-8000-000000000001",
            "domain": "education",
            "event_kind": "education_start",
            "date_start": "2008-01-01",
            "date_end": "2008-12-31",
            "precision": "year",
            "summary": "入学",
        },
        {
            "id": "00000000-0000-4000-8000-000000000002",
            "domain": "career",
            "event_kind": "career_entry",
            "date_start": "2012-06-01",
            "date_end": "2012-06-30",
            "precision": "month",
            "summary": "入职",
        },
    ],
}
REAL_TIMES = ["12:00", "12:05", "12:10", "12:15", "12:20"]
FROZEN_TODAY = date(2026, 9, 16)


@lru_cache(maxsize=1)
def _real_built() -> dict:
    contexts = []
    for value in REAL_TIMES:
        hour, minute = value.split(":")
        contexts.append(
            build_candidate_static_context(REAL_REQUEST, datetime(1990, 1, 1, int(hour), int(minute)))
        )
    return {"static_contexts": contexts}


def _real_windows(**extra: object) -> list[dict]:
    return guided_collect_windows(
        {**REAL_REQUEST, **extra},
        _real_built(),
        candidate_times=REAL_TIMES,
        today=FROZEN_TODAY,
    )


class GuidedCollectWindowsTests(unittest.TestCase):
    def test_real_engine_twenty_minute_window_emits_windows(self) -> None:
        rows = _real_windows()
        self.assertGreaterEqual(len(rows), 1)
        self.assertLessEqual(len(rows), GUIDED_COLLECT_LIMIT)
        for row in rows:
            self.assertIn("year", row)
            self.assertIn("month_lo", row)
            self.assertIn("month_hi", row)
            self.assertIn("domain", row)
            self.assertIn("split", row)
            self.assertGreaterEqual(row["month_lo"], 1)
            self.assertLessEqual(row["month_hi"], 12)
            self.assertGreaterEqual(row["month_hi"], row["month_lo"])
            self.assertGreaterEqual(row["split"]["left"], 1)
            self.assertGreaterEqual(row["split"]["right"], 1)

    def test_twenty_minute_window_emits_at_least_one_real_dasha_window(self) -> None:
        built = _multi_layer_window_built()
        rows = guided_collect_windows(
            _request(),
            built,
            candidate_times=["04:47", "04:48", "05:00", "05:06", "05:07"],
            today=date(2026, 8, 22),
        )
        self.assertGreaterEqual(len(rows), 1)
        self.assertLessEqual(len(rows), GUIDED_COLLECT_LIMIT)

    def test_windows_without_a_domain_track_are_open(self) -> None:
        """BUG-749: vim / natal narayana boundaries say nothing about domain."""
        built = _multi_layer_window_built()
        rows = guided_collect_windows(
            _request(),
            built,
            candidate_times=["04:47", "04:48", "05:00", "05:06", "05:07"],
            today=date(2026, 8, 22),
        )
        self.assertTrue(rows)
        self.assertTrue(all(row["domain"] == GUIDED_ANY_DOMAIN for row in rows))

    def test_d9_and_d10_tracks_keep_their_fixed_domain(self) -> None:
        rows = _real_windows()
        self.assertTrue(rows)
        self.assertTrue(
            any(row["domain"] in {"career", "relationship"} for row in rows),
            rows,
        )

    def test_declined_domains_fall_back_to_open_windows(self) -> None:
        full = _real_windows()
        self.assertTrue(full)
        target = next(
            row["domain"] for row in full if row["domain"] != GUIDED_ANY_DOMAIN
        )
        blocked = _real_windows(declined_domains=[target])
        self.assertTrue(blocked)
        self.assertFalse(any(row["domain"] == target for row in blocked))

    def test_split_balance_sorts_before_year_distance(self) -> None:
        built = {
            "static_contexts": [
                _context("04:47", d4_asc=0, sun_house=4, sun_varga_sign=3, moon=80.0, d9_asc=1, d10_asc=1),
                _context("04:48", d4_asc=1, sun_house=10, sun_varga_sign=9, moon=140.0, d9_asc=2, d10_asc=2),
                _context("05:06", d4_asc=1, sun_house=10, sun_varga_sign=9, moon=200.0, d9_asc=3, d10_asc=3),
            ],
        }
        rows = guided_collect_windows(
            _request(events=_gate_events()),
            built,
            candidate_times=["04:47", "04:48", "05:06"],
            today=date(2026, 8, 22),
        )
        self.assertGreaterEqual(len(rows), 2)
        balances = [abs(row["split"]["left"] - row["split"]["right"]) for row in rows]
        self.assertEqual(balances, sorted(balances))

    def test_cross_year_split_emits_separate_rows(self) -> None:
        built = {
            "static_contexts": [
                _context("04:50", d4_asc=0, sun_house=4, sun_varga_sign=3, moon=10.0, d9_asc=0, d10_asc=0),
                _context("05:10", d4_asc=6, sun_house=10, sun_varga_sign=9, moon=190.0, d9_asc=6, d10_asc=6),
            ],
        }
        rows = guided_collect_windows(
            _request(),
            built,
            candidate_times=["04:50", "05:10"],
            today=date(2026, 8, 22),
        )
        self.assertTrue(rows, "expected real dasha windows from distant moons")
        for row in rows:
            self.assertGreaterEqual(row["month_hi"], row["month_lo"])


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