Panama/tests/quickshell/home_assistant_bridge_test.py

#!/usr/bin/env python3

from __future__ import annotations

import importlib.machinery
import importlib.util
import json
import pathlib
import sys
import threading
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


ROOT = pathlib.Path(__file__).resolve().parents[2]
HELPER = ROOT / "config/dot/quickshell/scripts/panama-home-assistant"

loader = importlib.machinery.SourceFileLoader("panama_home_assistant", str(HELPER))
spec = importlib.util.spec_from_loader(loader.name, loader)
bridge = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = bridge
loader.exec_module(bridge)


class FakeHomeAssistant(BaseHTTPRequestHandler):
    requests: list[dict[str, object]] = []

    def log_message(self, _format: str, *args: object) -> None:
        del args

    def _record(self, body: bytes = b"") -> None:
        self.requests.append(
            {
                "method": self.command,
                "path": self.path,
                "authorization": self.headers.get("Authorization"),
                "body": body,
            }
        )

    def _json(self, value: object, status: int = 200) -> None:
        payload = json.dumps(value).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def do_GET(self) -> None:
        self._record()
        if self.path == "/api/":
            self._json({"message": "API running."})
            return
        if self.path == "/api/states":
            self._json(
                [
                    {
                        "entity_id": "light.hall",
                        "state": "off",
                        "attributes": {"friendly_name": "Hall"},
                    },
                    {
                        "entity_id": "sensor.private",
                        "state": "1",
                        "attributes": {"friendly_name": "Private"},
                    },
                    {
                        "entity_id": "light.kitchen",
                        "state": "on",
                        "attributes": {"friendly_name": "Kitchen"},
                    },
                ]
            )
            return
        self._json({"message": "not found"}, 404)

    def do_POST(self) -> None:
        length = int(self.headers.get("Content-Length", "0"))
        body = self.rfile.read(length)
        self._record(body)
        if self.path == "/api/services/homeassistant/toggle":
            self._json([])
            return
        self._json({"message": "not found"}, 404)


class HomeAssistantBridgeTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.server = ThreadingHTTPServer(("127.0.0.1", 0), FakeHomeAssistant)
        cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
        cls.thread.start()
        cls.base_url = f"http://127.0.0.1:{cls.server.server_port}"

    @classmethod
    def tearDownClass(cls) -> None:
        cls.server.shutdown()
        cls.server.server_close()
        cls.thread.join(timeout=2)

    def setUp(self) -> None:
        FakeHomeAssistant.requests.clear()

    def config(self) -> object:
        return bridge.Config(
            base_url=self.base_url,
            token="fixture-token",
            entity_ids=("light.kitchen", "light.hall"),
        )

    def test_environment_configuration_wins_over_legacy_sources(self) -> None:
        env = {
            "PANAMA_HOME_ASSISTANT_URL": "https://home.example",
            "PANAMA_HOME_ASSISTANT_TOKEN": "fixture-token",
            "PANAMA_HOME_ASSISTANT_ENTITIES": "light.kitchen,light.hall",
        }

        config = bridge.resolve_config(
            env,
            legacy=lambda: (_ for _ in ()).throw(AssertionError("legacy called")),
        )

        self.assertEqual(config.base_url, "https://home.example")
        self.assertEqual(config.entity_ids, ("light.kitchen", "light.hall"))

    def test_snapshot_filters_and_preserves_configured_order(self) -> None:
        snapshot = bridge.collect_snapshot(self.config())

        self.assertTrue(snapshot["ok"])
        self.assertEqual(
            [item["name"] for item in snapshot["entities"]],
            ["Kitchen", "Hall"],
        )
        self.assertNotIn("sensor.private", json.dumps(snapshot))

    def test_snapshot_uses_bearer_authentication(self) -> None:
        bridge.collect_snapshot(self.config())

        request = FakeHomeAssistant.requests[-1]
        self.assertEqual(request["method"], "GET")
        self.assertEqual(request["path"], "/api/states")
        self.assertEqual(request["authorization"], "Bearer fixture-token")

    def test_toggle_rejects_an_unconfigured_entity(self) -> None:
        with self.assertRaisesRegex(ValueError, "entity-not-configured"):
            bridge.ensure_configured(
                "light.office",
                ("light.kitchen",),
            )

    def test_toggle_calls_the_homeassistant_service(self) -> None:
        result = bridge.toggle(self.config(), "light.kitchen")

        self.assertTrue(result["ok"])
        request = FakeHomeAssistant.requests[-1]
        self.assertEqual(request["method"], "POST")
        self.assertEqual(request["path"], "/api/services/homeassistant/toggle")
        self.assertEqual(
            json.loads(request["body"]),
            {"entity_id": "light.kitchen"},
        )

    def test_authentication_error_is_redacted(self) -> None:
        self.assertEqual(
            bridge.public_http_error(401, "sensitive response"),
            "authentication-required",
        )
        self.assertEqual(
            bridge.public_http_error(500, "sensitive response"),
            "request-failed",
        )

    def test_gsettings_array_parser_handles_empty_and_values(self) -> None:
        self.assertEqual(bridge.parse_gsettings_array("@as []"), ())
        self.assertEqual(
            bridge.parse_gsettings_array("['light.kitchen', 'light.hall']"),
            ("light.kitchen", "light.hall"),
        )


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