82f43f75ba
shrink root README into a landing page with a docs map and focused contributor guidance add TV_POWER_RUNBOOK as the canonical TV power rollout and canary runbook add CHANGELOG and move project history out of README-style docs refactor src README into a developer-focused guide (architecture, runtime files, MQTT, debugging) prune redundant older HDMI docs and keep a canonical HDMI_CEC_SETUP path update copilot instructions to a high-signal policy format with strict anti-shadow-README design rules align references across docs to current files, scripts, and TV power behavior
314 lines
12 KiB
Python
314 lines
12 KiB
Python
"""
|
|
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()
|