led-controller/tests/api_server.py

"""Shared FastAPI test server fixture for API endpoint tests."""

from __future__ import annotations

import builtins
import json
import os
import sys
from pathlib import Path
from typing import Any, Dict, Optional

import pytest
from starlette.testclient import TestClient

PROJECT_ROOT = Path(__file__).resolve().parents[1]
SRC_PATH = PROJECT_ROOT / "src"
LIB_PATH = PROJECT_ROOT / "lib"

for p in (str(PROJECT_ROOT), str(LIB_PATH), str(SRC_PATH)):
    if p in sys.path:
        sys.path.remove(p)
    sys.path.insert(0, p)


class DummyBridge:
    def __init__(self):
        self.sent: list[tuple[Any, Optional[str]]] = []

    async def send(self, data: Any, addr: Optional[str] = None):
        if isinstance(data, dict):
            from util.bridge_envelope import (  # noqa: E402
                BROADCAST_MAC,
                build_devices_envelope,
                format_mac_key,
                is_broadcast_mac,
                normalize_mac_key,
            )
            from util.v1_wire import compact_envelope  # noqa: E402

            if data.get("v") == "1" and ("devices" in data or "dv" in data):
                data = compact_envelope(data)
            elif addr is not None:
                s = str(addr).strip().lower()
                if is_broadcast_mac(s):
                    mac_key = BROADCAST_MAC
                else:
                    h = normalize_mac_key(s)
                    mac_key = format_mac_key(h) if h else None
                if mac_key:
                    body = {k: v for k, v in data.items() if k != "v"}
                    data = build_devices_envelope({mac_key: body})
                else:
                    data = json.dumps(data, separators=(",", ":"))
            else:
                data = json.dumps(data, separators=(",", ":"))
        elif isinstance(data, (bytes, bytearray)):
            data = bytes(data).decode(errors="ignore")
        self.sent.append((data, addr))
        return True


def bridge_sent_envelope(bridge: DummyBridge, index: int) -> Dict[str, Any]:
    data, _addr = bridge.sent[index]
    if isinstance(data, dict):
        return data
    return json.loads(data)


def device_body_from_envelope(envelope: Dict[str, Any], mac: str) -> Dict[str, Any]:
    from util.bridge_envelope import format_mac_key, normalize_mac_key  # noqa: E402

    devs = envelope.get("dv") or envelope.get("devices") or {}
    key = format_mac_key(normalize_mac_key(mac))
    return devs[key]


@pytest.fixture(scope="function")
def server(monkeypatch, tmp_path_factory):
    """In-process FastAPI app with isolated db/settings."""
    tmp_root = tmp_path_factory.mktemp("endpoint-tests")
    tmp_db_dir = tmp_root / "db"
    tmp_settings_file = tmp_root / "settings.json"

    for p in (str(SRC_PATH), str(LIB_PATH), str(PROJECT_ROOT)):
        if p in sys.path:
            sys.path.remove(p)
        sys.path.insert(0, p)

    import settings as settings_mod  # noqa: E402

    settings_mod.Settings.SETTINGS_FILE = str(tmp_settings_file)

    import models.model as model_mod  # noqa: E402

    monkeypatch.setattr(model_mod, "_db_dir", lambda: str(tmp_db_dir))

    import models.preset as models_preset  # noqa: E402
    import models.profile as models_profile  # noqa: E402
    import models.group as models_group  # noqa: E402
    import models.zone as models_zone  # noqa: E402
    import models.pallet as models_pallet  # noqa: E402
    import models.scene as models_scene  # noqa: E402
    import models.pattern as models_pattern  # noqa: E402
    import models.sequence as models_sequence  # noqa: E402
    import models.device as models_device  # noqa: E402

    for cls in (
        models_preset.Preset,
        models_profile.Profile,
        models_group.Group,
        models_zone.Zone,
        models_pallet.Palette,
        models_scene.Scene,
        models_pattern.Pattern,
        models_sequence.Sequence,
        models_device.Device,
    ):
        if hasattr(cls, "_instance"):
            delattr(cls, "_instance")

    orig_open = builtins.open

    def patched_open(file, *args, **kwargs):
        if isinstance(file, str):
            if file in {"db/pattern.json", "pattern.json", "/db/pattern.json"}:
                file = str(PROJECT_ROOT / "db" / "pattern.json")
        return orig_open(file, *args, **kwargs)

    monkeypatch.setattr(builtins, "open", patched_open)

    old_cwd = os.getcwd()
    os.chdir(str(SRC_PATH))

    dummy_bridge = DummyBridge()

    try:
        for mod_name in (
            "controllers.preset",
            "controllers.profile",
            "controllers.group",
            "controllers.sequence",
            "controllers.zone",
            "controllers.palette",
            "controllers.scene",
            "controllers.pattern",
            "controllers.settings",
            "controllers.device",
            "controllers.wifi_bridge",
            "fastapi_app",
            "app_factory",
        ):
            sys.modules.pop(mod_name, None)

        from models.transport import set_bridge  # noqa: E402
        from fastapi_app import create_application  # noqa: E402

        set_bridge(dummy_bridge)
        app = create_application(test_mode=True)

        with TestClient(app, raise_server_exceptions=True) as client:
            yield {
                "base_url": "",
                "client": client,
                "bridge": dummy_bridge,
            }
    finally:
        os.chdir(old_cwd)