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 os
import pathlib
import subprocess
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"
QML_SERVICE = ROOT / "config/dot/quickshell/services/HomeAssistant.qml"

# What a Home Assistant would render for the areas template: every area it
# holds, including ones whose entities are not lights, plus entries no template
# should produce but a bridge still has to survive.
TEMPLATE_AREAS = [
    {
        "id": "kitchen",
        "name": "Kitchen",
        "entities": ["light.kitchen", "switch.kettle", "sensor.private"],
    },
    {"id": "hallway", "name": "Hall", "entities": ["light.hall", "light.corner"]},
    {"id": "garage", "name": "Garage", "entities": ["sensor.garage_door"]},
    {"id": 12, "name": "Numeric", "entities": ["light.kitchen"]},
    {"id": "unnamed", "name": "", "entities": ["light.kitchen"]},
    {"id": "not_a_list", "name": "Loose", "entities": "light.kitchen"},
    "not-an-area",
]

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]] = []
    # "rendered" | "malformed" | "unauthorized"
    template_mode: str = "rendered"

    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)

    # Home Assistant answers /api/template with the rendered text, not with a
    # JSON document, so the fixture replies in kind.
    def _text(self, body: str, status: int = 200) -> None:
        payload = body.encode()
        self.send_response(status)
        self.send_header("Content-Type", "text/plain; charset=utf-8")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def _template(self) -> None:
        if self.template_mode == "unauthorized":
            self._json({"message": "Unauthorized: token abc123"}, 401)
            return
        if self.template_mode == "malformed":
            self._text("Error rendering template: UndefinedError")
            return
        self._text(json.dumps(TEMPLATE_AREAS))

    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.kitchen",
                        "state": "on",
                        "attributes": {
                            "friendly_name": "Kitchen",
                            "brightness": 128,
                            "supported_color_modes": ["brightness"],
                        },
                    },
                    {
                        "entity_id": "light.hall",
                        "state": "off",
                        "attributes": {
                            "friendly_name": "Hall",
                            "supported_color_modes": ["color_temp"],
                        },
                    },
                    {"entity_id": "sensor.private", "state": "1", "attributes": {"token": "never-return"}},
                    {
                        "entity_id": "light.malformed",
                        "state": "on",
                        "attributes": "invalid",
                    },
                    {
                        "entity_id": "light.corner",
                        "state": "unavailable",
                        "attributes": {
                            "friendly_name": "Corner",
                            "supported_color_modes": ["brightness"],
                        },
                    },
                ]
            )
            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/template":
            self._template()
            return
        if self.path in {
            "/api/services/homeassistant/toggle",
            "/api/services/light/turn_on",
            "/api/services/light/turn_off",
        }:
            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()
        FakeHomeAssistant.template_mode = "rendered"

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

    def helper_env(self) -> dict[str, str]:
        return {
            "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
            "HOME": os.environ.get("HOME", "/tmp"),
            "PANAMA_HOME_ASSISTANT_URL": self.base_url,
            "PANAMA_HOME_ASSISTANT_TOKEN": "fixture-token",
            "PANAMA_HOME_ASSISTANT_ENTITIES": "light.kitchen",
        }

    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_config_does_not_require_legacy_entity_ids(self) -> None:
        self.assertTrue(
            bridge.Config(
                base_url="https://home.example",
                token="fixture-token",
                entity_ids=(),
            ).configured
        )

    def test_legacy_entity_ids_remain_migration_metadata(self) -> None:
        config = bridge.resolve_config(
            {
                "PANAMA_HOME_ASSISTANT_URL": "https://home.example",
                "PANAMA_HOME_ASSISTANT_TOKEN": "fixture-token",
            },
            legacy=lambda: bridge.Config(
                base_url="https://legacy.example",
                token="legacy-token",
                entity_ids=("light.kitchen",),
            ),
        )

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

    def test_catalog_filters_and_normalizes_discovered_lights(self) -> None:
        catalog = bridge.collect_catalog(self.config())

        self.assertTrue(catalog["ok"])
        self.assertEqual(
            [item["sourceName"] for item in catalog["entities"]],
            ["Kitchen", "Hall", "Corner"],
        )
        self.assertEqual(catalog["entities"][0]["brightnessPct"], 50)
        self.assertEqual(catalog["entities"][1]["brightnessPct"], 0)
        self.assertTrue(all(item["dimmable"] for item in catalog["entities"]))
        rendered = json.dumps(catalog)
        self.assertNotIn("sensor.private", rendered)
        self.assertNotIn("light.malformed", rendered)
        self.assertNotIn("token", rendered)

    def test_snapshot_uses_bearer_authentication(self) -> None:
        bridge.collect_catalog(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_snapshot_remains_a_catalog_compatibility_alias(self) -> None:
        snapshot = bridge.collect_snapshot(self.config())

        self.assertTrue(snapshot["ok"])
        self.assertEqual(
            [item["sourceName"] for item in snapshot["entities"]],
            ["Kitchen", "Hall", "Corner"],
        )

    def test_areas_group_lights_by_home_assistant_area(self) -> None:
        result = bridge.collect_areas(self.config())

        self.assertTrue(result["ok"])
        self.assertEqual(result["error"], "")
        self.assertEqual(
            result["areas"],
            [
                {"id": "kitchen", "name": "Kitchen", "entities": ["light.kitchen"]},
                {
                    "id": "hallway",
                    "name": "Hall",
                    "entities": ["light.hall", "light.corner"],
                },
            ],
        )

    def test_areas_render_the_whole_list_from_one_template_request(self) -> None:
        bridge.collect_areas(self.config())

        request = FakeHomeAssistant.requests[-1]
        self.assertEqual(len(FakeHomeAssistant.requests), 1)
        self.assertEqual(request["method"], "POST")
        self.assertEqual(request["path"], "/api/template")
        self.assertEqual(request["authorization"], "Bearer fixture-token")
        template = json.loads(request["body"])["template"]
        for fragment in ("areas()", "area_name(area)", "area_entities(area)", "tojson"):
            self.assertIn(fragment, template)

    def test_areas_keep_only_lights_and_drop_areas_left_with_none(self) -> None:
        rendered = json.dumps(bridge.collect_areas(self.config()))

        self.assertNotIn("switch.kettle", rendered)
        self.assertNotIn("sensor.private", rendered)
        self.assertNotIn("Garage", rendered)

    def test_areas_discard_malformed_entries_without_losing_the_rest(self) -> None:
        rendered = json.dumps(bridge.collect_areas(self.config()))

        self.assertNotIn("Numeric", rendered)
        self.assertNotIn("unnamed", rendered)
        self.assertNotIn("Loose", rendered)
        self.assertIn("Kitchen", rendered)

    def test_areas_reject_a_template_response_that_is_not_json(self) -> None:
        FakeHomeAssistant.template_mode = "malformed"

        result = bridge.collect_areas(self.config())

        self.assertFalse(result["ok"])
        self.assertEqual(result["error"], "invalid-response")
        self.assertEqual(result["areas"], [])

    def test_areas_redact_an_authentication_failure(self) -> None:
        FakeHomeAssistant.template_mode = "unauthorized"

        result = bridge.collect_areas(self.config())

        self.assertFalse(result["ok"])
        self.assertEqual(result["error"], "authentication-required")
        self.assertNotIn("abc123", json.dumps(result))

    def test_areas_report_not_configured_before_any_request(self) -> None:
        result = bridge.collect_areas(
            bridge.Config(base_url=self.base_url, token="", entity_ids=())
        )

        self.assertEqual(
            result, {"ok": False, "areas": [], "error": "not-configured"}
        )
        self.assertEqual(FakeHomeAssistant.requests, [])

    def test_cli_areas_prints_one_json_object(self) -> None:
        completed = subprocess.run(
            [sys.executable, str(HELPER), "areas"],
            capture_output=True,
            text=True,
            env=self.helper_env(),
            timeout=15,
        )

        self.assertEqual(completed.returncode, 0)
        self.assertEqual(len(completed.stdout.strip().splitlines()), 1)
        self.assertEqual(
            [area["name"] for area in json.loads(completed.stdout)["areas"]],
            ["Kitchen", "Hall"],
        )

    def test_cli_still_rejects_unknown_arguments(self) -> None:
        completed = subprocess.run(
            [sys.executable, str(HELPER), "areas", "kitchen"],
            capture_output=True,
            text=True,
            env=self.helper_env(),
            timeout=15,
        )

        self.assertEqual(completed.returncode, 2)
        self.assertEqual(
            json.loads(completed.stdout), {"ok": False, "error": "invalid-command"}
        )

    def test_action_rejects_an_entity_not_in_the_live_catalog(self) -> None:
        with self.assertRaisesRegex(ValueError, "entity-not-discovered"):
            bridge.toggle(self.config(), "light.office")

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

        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.corner"},
        )

    def test_brightness_uses_turn_on_for_positive_percent(self) -> None:
        bridge.set_brightness(self.config(), "light.kitchen", 62)

        request = FakeHomeAssistant.requests[-1]
        self.assertEqual(request["path"], "/api/services/light/turn_on")
        self.assertEqual(
            json.loads(request["body"]),
            {"entity_id": "light.kitchen", "brightness_pct": 62},
        )

    def test_brightness_zero_uses_turn_off(self) -> None:
        bridge.set_brightness(self.config(), "light.hall", 0)

        request = FakeHomeAssistant.requests[-1]
        self.assertEqual(request["path"], "/api/services/light/turn_off")
        self.assertEqual(
            json.loads(request["body"]),
            {"entity_id": "light.hall"},
        )

    def test_brightness_rejects_invalid_values_before_any_request(self) -> None:
        for value in (-1, 101, 1.5, "bright"):
            with self.subTest(value=value):
                with self.assertRaisesRegex(ValueError, "invalid-brightness"):
                    bridge.set_brightness(self.config(), "light.kitchen", value)
                self.assertEqual(FakeHomeAssistant.requests, [])

    def test_parse_brightness_rejects_invalid_cli_values(self) -> None:
        for value in ("-1", "101", "1.5", "bright"):
            with self.subTest(value=value):
                with self.assertRaisesRegex(ValueError, "invalid-brightness"):
                    bridge.parse_brightness(value)

    def test_qml_refreshes_through_the_catalog_command(self) -> None:
        self.assertIn('command: [root.helperPath, "catalog"]', QML_SERVICE.read_text())

    def test_qml_composes_catalog_and_preferences_for_selected_entities(self) -> None:
        source = QML_SERVICE.read_text()

        self.assertIn(
            "root.catalog = Array.isArray(result.entities) ? result.entities : [];",
            source,
        )
        self.assertIn(
            "const favorites = root.fixtureMode\n"
            "            ? root.fixtureFavorites\n"
            "            : HomePreferences.favorites;",
            source,
        )
        self.assertIn("name: alias || entity.sourceName,", source)
        self.assertIn("root.selectedEntities = nextSelection;", source)

    def test_qml_fetches_areas_alongside_the_catalog(self) -> None:
        source = QML_SERVICE.read_text()

        self.assertIn('command: [root.helperPath, "areas"]', source)
        self.assertIn("if (!areasProc.running)\n            areasProc.running = true;", source)

    def test_qml_composes_rooms_and_tolerates_missing_areas(self) -> None:
        source = QML_SERVICE.read_text()

        self.assertIn("property var areas: []", source)
        self.assertIn(
            "readonly property var rooms: root.composeRooms(root.catalog, root.areas)",
            source,
        )
        self.assertIn('grouped.push({ id: "", name: "Other", lights: orphans });', source)
        # A failed areas fetch empties the grouping and nothing else: no phase,
        # no stale flag, no lastError, so lights keep working without it.
        consume = source.split("function consumeAreas(text: string): void {", 1)[1]
        consume = consume.split("\n    }\n", 1)[0]
        for untouched in ("root.phase", "root.stale", "root.lastError"):
            self.assertNotIn(untouched, consume)

    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()