107 lines
3.8 KiB
Python
107 lines
3.8 KiB
Python
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()
|