import unittest from datetime import datetime, timezone from scheduler.db_utils import ( build_group_power_intent_body, compute_group_power_intent_basis, compute_group_power_intent_fingerprint, ) class PowerIntentUtilsTests(unittest.TestCase): def test_no_events_results_in_off(self): now = datetime(2026, 3, 31, 10, 0, 0, tzinfo=timezone.utc) basis = compute_group_power_intent_basis(events=[], group_id=7, now_utc=now) self.assertEqual(basis["group_id"], 7) self.assertEqual(basis["desired_state"], "off") self.assertEqual(basis["reason"], "no_active_event") self.assertIsNone(basis["event_window_start"]) self.assertIsNone(basis["event_window_end"]) def test_active_event_results_in_on(self): now = datetime(2026, 3, 31, 10, 5, 0, tzinfo=timezone.utc) events = [ { "id": 101, "group_id": 2, "start": "2026-03-31T10:00:00+00:00", "end": "2026-03-31T10:30:00+00:00", } ] basis = compute_group_power_intent_basis(events=events, group_id=2, now_utc=now) self.assertEqual(basis["desired_state"], "on") self.assertEqual(basis["reason"], "active_event") self.assertEqual(basis["event_window_start"], "2026-03-31T10:00:00Z") self.assertEqual(basis["event_window_end"], "2026-03-31T10:30:00Z") self.assertEqual(basis["active_event_ids"], [101]) def test_adjacent_events_are_merged_without_off_blip(self): now = datetime(2026, 3, 31, 10, 30, 0, tzinfo=timezone.utc) events = [ { "id": 1, "group_id": 3, "start": "2026-03-31T10:00:00+00:00", "end": "2026-03-31T10:30:00+00:00", }, { "id": 2, "group_id": 3, "start": "2026-03-31T10:30:00+00:00", "end": "2026-03-31T11:00:00+00:00", }, ] basis = compute_group_power_intent_basis(events=events, group_id=3, now_utc=now) self.assertEqual(basis["desired_state"], "on") self.assertEqual(basis["event_window_start"], "2026-03-31T10:00:00Z") self.assertEqual(basis["event_window_end"], "2026-03-31T11:00:00Z") def test_true_gap_results_in_off(self): now = datetime(2026, 3, 31, 10, 31, 0, tzinfo=timezone.utc) events = [ { "id": 1, "group_id": 4, "start": "2026-03-31T10:00:00+00:00", "end": "2026-03-31T10:30:00+00:00", }, { "id": 2, "group_id": 4, "start": "2026-03-31T10:35:00+00:00", "end": "2026-03-31T11:00:00+00:00", }, ] basis = compute_group_power_intent_basis(events=events, group_id=4, now_utc=now) self.assertEqual(basis["desired_state"], "off") self.assertEqual(basis["reason"], "no_active_event") def test_fingerprint_is_stable_for_same_semantics(self): basis = { "schema_version": "1.0", "group_id": 9, "desired_state": "on", "reason": "active_event", "event_window_start": "2026-03-31T10:00:00Z", "event_window_end": "2026-03-31T10:30:00Z", "active_event_ids": [12, 7], } body_a = build_group_power_intent_body(basis, poll_interval_sec=15) body_b = build_group_power_intent_body(basis, poll_interval_sec=15) fingerprint_a = compute_group_power_intent_fingerprint(body_a) fingerprint_b = compute_group_power_intent_fingerprint(body_b) self.assertEqual(fingerprint_a, fingerprint_b) if __name__ == "__main__": unittest.main()