led-controller/tests/test_bridge_envelope.py

#!/usr/bin/env python3
"""Tests for bridge devices envelope (Pi + espnow-sender downlink)."""

from __future__ import annotations

import json
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))

from util.bridge_envelope import (  # noqa: E402
    BROADCAST_MAC,
    build_devices_envelope,
    build_groups_envelope,
    build_v1_body,
    envelope_payload_size,
    format_mac_key,
    is_broadcast_mac,
    split_v1_body_for_espnow,
    v1_body_size,
)


def test_unicast_mac_keys_per_device():
    from util.driver_delivery import _unicast_mac_keys

    keys = _unicast_mac_keys(["188b0e1560a8", "e8f60a16ea10"])
    assert len(keys) == 2
    assert keys[0] == "18:8b:0e:15:60:a8"
    assert keys[1] == "e8:f6:0a:16:ea:10"
    assert _unicast_mac_keys(["188b0e1560a8"]) == ["18:8b:0e:15:60:a8"]
    assert _unicast_mac_keys(None) == [BROADCAST_MAC]


def test_deliver_json_messages_defaults_broadcast():
    from util.driver_delivery import deliver_json_messages

    class _Sender:
        def __init__(self):
            self.keys = []

        async def send(self, envelope):
            devs = envelope.get("dv") or envelope.get("devices") or {}
            self.keys.extend(devs.keys())
            return True

    async def _run():
        sender = _Sender()
        await deliver_json_messages(
            sender,
            [json.dumps({"v": "1", "select": ["2"]})],
            ["188b0e1560a8", "e8f60a16ea10"],
            None,
        )
        return sender.keys

    keys = __import__("asyncio").run(_run())
    assert keys == [BROADCAST_MAC]


def is_devices_envelope(raw: bytes) -> bool:
    if not raw or raw[0:1] != b"{":
        return False
    try:
        data = json.loads(raw)
    except (ValueError, TypeError):
        return False
    devs = data.get("devices") if isinstance(data, dict) else None
    if devs is None and isinstance(data, dict):
        devs = data.get("dv")
    return isinstance(data, dict) and data.get("v") == "1" and isinstance(devs, dict)


def build_driver_payload(body: dict) -> bytes:
    out = {"v": "1", **{k: body[k] for k in body if k != "v"}}
    raw = json.dumps(out)
    if len(raw) > 250:
        raise ValueError("too large")
    return raw.encode("utf-8")


def test_build_groups_envelope():
    env = build_groups_envelope("e8f60a16ea10", ["5", "18"])
    assert env["v"] == "1"
    key = format_mac_key("e8f60a16ea10")
    devs = env.get("dv") or env.get("devices")
    body = devs[key]
    assert body["sg"] is True
    assert body["g"] == ["5", "18"]


def test_is_broadcast_mac():
    assert is_broadcast_mac("ff:ff:ff:ff:ff:ff")
    assert is_broadcast_mac("ffffffffffff")
    assert not is_broadcast_mac("e8f60a16ea10")


def test_is_devices_envelope():
    env = build_devices_envelope(
        {
            BROADCAST_MAC: build_v1_body(
                presets={"1": {"p": "on", "c": ["#FFFFFF"], "a": True}},
                groups=["5"],
                set_groups=False,
            )
        }
    )
    raw = json.dumps(env).encode("utf-8")
    assert is_devices_envelope(raw)
    assert not is_devices_envelope(b'{"v":"1","s":{}}')


def test_build_driver_payload_size():
    body = build_v1_body(
        presets={"x": {"pattern": "on", "colors": ["#FF0000"], "auto": True}},
        select=["x", 0],
        save=True,
    )
    payload = build_driver_payload(body)
    assert len(payload) <= 250
    data = json.loads(payload)
    assert data["v"] == "1"
    assert data["s"] == ["x", 0]


def test_split_preset_and_select():
    body = build_v1_body(
        presets={
            "2": {
                "p": "on",
                "c": ["#FFFFFF"],
                "bg": "#000000",
                "d": 100,
                "b": 255,
                "a": True,
                "n1": 0,
                "n2": 0,
            }
        },
        select=["2", 0],
        save=True,
    )
    if v1_body_size(body) <= 250:
        chunks = split_v1_body_for_espnow(body)
        assert len(chunks) == 1
    else:
        chunks = split_v1_body_for_espnow(body)
        assert len(chunks) >= 2
        assert all(v1_body_size(c) <= 250 for c in chunks)
        assert "p" in chunks[0]
        assert any("s" in c for c in chunks)


def test_envelope_fits_espnow_limit():
    env = build_devices_envelope(
        {
            BROADCAST_MAC: build_v1_body(
                presets={
                    "2": {
                        "pattern": "on",
                        "colors": ["#FFFFFF"],
                        "auto": True,
                    }
                },
                select=["2"],
            )
        }
    )
    assert envelope_payload_size(env) <= 250