""" Unit tests for the TV power intent validation and state management. Run from project root (venv activated): python -m pytest tests/test_power_intent.py -v """ import sys import os import json import tempfile import unittest from datetime import datetime, timezone, timedelta from unittest.mock import patch, MagicMock # Ensure src/ is importable without running MQTT code sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) from simclient import ( validate_power_intent_payload, write_power_intent_state, _parse_utc_iso, POWER_INTENT_STATE_FILE, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_intent( desired: str = "on", seconds_valid: int = 90, group_id: int = 2, intent_id: str = "test-intent-001", poll_interval: int = 15, offset_issued: timedelta = timedelta(0), ) -> dict: """Build a valid v1 power-intent payload.""" now = datetime.now(timezone.utc) + offset_issued exp = now + timedelta(seconds=seconds_valid) return { "schema_version": "1.0", "intent_id": intent_id, "group_id": group_id, "desired_state": desired, "reason": "active_event" if desired == "on" else "no_active_event", "issued_at": now.strftime("%Y-%m-%dT%H:%M:%SZ"), "expires_at": exp.strftime("%Y-%m-%dT%H:%M:%SZ"), "poll_interval_sec": poll_interval, "active_event_ids": [1] if desired == "on" else [], "event_window_start": now.strftime("%Y-%m-%dT%H:%M:%SZ") if desired == "on" else None, "event_window_end": exp.strftime("%Y-%m-%dT%H:%M:%SZ") if desired == "on" else None, } def _make_stale_intent(desired: str = "on") -> dict: """Build an expired v1 payload (issued_at and expires_at both in the past).""" return { "schema_version": "1.0", "intent_id": "stale-001", "group_id": 2, "desired_state": desired, "reason": "no_active_event", "issued_at": "2026-01-01T00:00:00Z", "expires_at": "2026-01-01T01:30:00Z", "poll_interval_sec": 15, "active_event_ids": [], "event_window_start": None, "event_window_end": None, } # --------------------------------------------------------------------------- # Tests: _parse_utc_iso # --------------------------------------------------------------------------- class TestParseUtcIso(unittest.TestCase): def test_z_suffix(self): dt = _parse_utc_iso("2026-01-15T10:30:00Z") self.assertEqual(dt.tzinfo, timezone.utc) self.assertEqual(dt.year, 2026) self.assertEqual(dt.second, 0) def test_plus00_suffix(self): dt = _parse_utc_iso("2026-01-15T10:30:00+00:00") self.assertEqual(dt.tzinfo, timezone.utc) def test_none_raises(self): with self.assertRaises(Exception): _parse_utc_iso(None) def test_garbage_raises(self): with self.assertRaises(Exception): _parse_utc_iso("not-a-date") # --------------------------------------------------------------------------- # Tests: validate_power_intent_payload — accepted paths # --------------------------------------------------------------------------- class TestValidateAccepted(unittest.TestCase): def test_valid_on_no_group_check(self): intent = _make_intent("on") ok, norm, err = validate_power_intent_payload(intent) self.assertTrue(ok, err) self.assertIsNotNone(norm) self.assertEqual(norm["desired_state"], "on") self.assertEqual(norm["group_id"], 2) def test_valid_off_no_group_check(self): intent = _make_intent("off") ok, norm, err = validate_power_intent_payload(intent) self.assertTrue(ok, err) self.assertEqual(norm["desired_state"], "off") def test_valid_on_with_matching_group_id_str(self): intent = _make_intent("on", group_id=5) ok, norm, err = validate_power_intent_payload(intent, expected_group_id="5") self.assertTrue(ok, err) def test_valid_on_with_matching_group_id_int(self): intent = _make_intent("on", group_id=7) ok, norm, err = validate_power_intent_payload(intent, expected_group_id=7) self.assertTrue(ok, err) def test_normalized_output_contains_required_keys(self): intent = _make_intent("on") ok, norm, _ = validate_power_intent_payload(intent) self.assertTrue(ok) for key in ("intent_id", "desired_state", "issued_at", "expires_at", "group_id", "poll_interval_sec", "active_event_ids", "event_window_start", "event_window_end"): self.assertIn(key, norm, f"missing key: {key}") # --------------------------------------------------------------------------- # Tests: validate_power_intent_payload — rejected paths # --------------------------------------------------------------------------- class TestValidateRejected(unittest.TestCase): def test_missing_required_fields(self): ok, _, err = validate_power_intent_payload({"schema_version": "1.0"}) self.assertFalse(ok) self.assertIn("missing required field", err) def test_wrong_schema_version(self): intent = _make_intent("on") intent["schema_version"] = "2.0" ok, _, err = validate_power_intent_payload(intent) self.assertFalse(ok) self.assertIn("schema_version", err) def test_invalid_desired_state(self): intent = _make_intent("on") intent["desired_state"] = "standby" ok, _, err = validate_power_intent_payload(intent) self.assertFalse(ok) self.assertIn("desired_state", err) def test_group_id_mismatch_string(self): intent = _make_intent("on", group_id=2) ok, _, err = validate_power_intent_payload(intent, expected_group_id="99") self.assertFalse(ok) self.assertIn("group_id mismatch", err) def test_group_id_mismatch_int(self): intent = _make_intent("on", group_id=2) ok, _, err = validate_power_intent_payload(intent, expected_group_id=99) self.assertFalse(ok) self.assertIn("group_id mismatch", err) def test_expired_intent(self): ok, _, err = validate_power_intent_payload(_make_stale_intent("on")) self.assertFalse(ok) self.assertIn("expired", err) def test_expires_before_issued(self): intent = _make_intent("on") # Swap the timestamps so expires < issued intent["expires_at"] = intent["issued_at"] ok, _, err = validate_power_intent_payload(intent) self.assertFalse(ok) def test_zero_poll_interval(self): intent = _make_intent("on", poll_interval=0) ok, _, err = validate_power_intent_payload(intent) self.assertFalse(ok) self.assertIn("poll_interval_sec", err) def test_negative_poll_interval(self): intent = _make_intent("on", poll_interval=-5) ok, _, err = validate_power_intent_payload(intent) self.assertFalse(ok) def test_active_event_ids_not_list(self): intent = _make_intent("on") intent["active_event_ids"] = "not-a-list" ok, _, err = validate_power_intent_payload(intent) self.assertFalse(ok) self.assertIn("active_event_ids", err) def test_missing_intent_id(self): intent = _make_intent("on") del intent["intent_id"] ok, _, err = validate_power_intent_payload(intent) self.assertFalse(ok) self.assertIn("missing required field", err) def test_invalid_issued_at_format(self): intent = _make_intent("on") intent["issued_at"] = "not-a-timestamp" ok, _, err = validate_power_intent_payload(intent) self.assertFalse(ok) self.assertIn("timestamp", err) # --------------------------------------------------------------------------- # Tests: write_power_intent_state atomic write # --------------------------------------------------------------------------- class TestWritePowerIntentState(unittest.TestCase): def test_writes_valid_json(self): data = {"intent_id": "abc", "desired_state": "on", "group_id": 2} with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f: tmp_path = f.name try: with patch("simclient.POWER_INTENT_STATE_FILE", tmp_path): write_power_intent_state(data) with open(tmp_path) as f: loaded = json.load(f) self.assertEqual(loaded["intent_id"], "abc") self.assertEqual(loaded["desired_state"], "on") finally: os.unlink(tmp_path) def test_atomic_write_replaces_existing(self): existing = {"intent_id": "old"} new_data = {"intent_id": "new", "desired_state": "off"} with tempfile.NamedTemporaryFile( suffix=".json", delete=False, mode="w" ) as f: json.dump(existing, f) tmp_path = f.name try: with patch("simclient.POWER_INTENT_STATE_FILE", tmp_path): write_power_intent_state(new_data) with open(tmp_path) as f: loaded = json.load(f) self.assertEqual(loaded["intent_id"], "new") finally: os.unlink(tmp_path) # --------------------------------------------------------------------------- # Tests: display_manager.ProcessHealthState power fields # --------------------------------------------------------------------------- class TestProcessHealthStatePowerFields(unittest.TestCase): def setUp(self): # Import here to avoid triggering display_manager side effects at module level from display_manager import ProcessHealthState self.ProcessHealthState = ProcessHealthState def test_initial_power_fields_are_none(self): h = self.ProcessHealthState() self.assertIsNone(h.power_control_mode) self.assertIsNone(h.power_source) self.assertIsNone(h.last_intent_id) self.assertIsNone(h.last_power_action) self.assertIsNone(h.last_power_at) def test_to_dict_contains_power_control(self): h = self.ProcessHealthState() d = h.to_dict() self.assertIn("power_control", d) pc = d["power_control"] self.assertIn("mode", pc) self.assertIn("source", pc) self.assertIn("last_intent_id", pc) self.assertIn("last_action", pc) self.assertIn("last_power_at", pc) def test_update_power_action_sets_fields(self): h = self.ProcessHealthState() h.power_control_mode = "hybrid" h.update_power_action("on", "mqtt_intent", "intent-xyz") self.assertEqual(h.last_power_action, "on") self.assertEqual(h.power_source, "mqtt_intent") self.assertEqual(h.last_intent_id, "intent-xyz") self.assertIsNotNone(h.last_power_at) def test_update_power_action_without_intent_id(self): h = self.ProcessHealthState() h.update_power_action("off", "local_fallback") self.assertEqual(h.last_power_action, "off") self.assertEqual(h.power_source, "local_fallback") self.assertIsNone(h.last_intent_id) def test_to_dict_reflects_update(self): h = self.ProcessHealthState() h.power_control_mode = "mqtt" h.update_power_action("off", "mqtt_intent", "intent-abc") d = h.to_dict() pc = d["power_control"] self.assertEqual(pc["mode"], "mqtt") self.assertEqual(pc["source"], "mqtt_intent") self.assertEqual(pc["last_intent_id"], "intent-abc") self.assertEqual(pc["last_action"], "off") if __name__ == "__main__": unittest.main()