hms-mediaengine/tests/unit/test_artnet_packets.py

"""Unit-Tests Art-Net-Pakete gegen die offizielle Spezifikation (PLAN.md §16).

Verifizierte Referenzwerte aus der Art-Net-4-Spezifikation:
- ID 'Art-Net\\0', OpCode little-endian, ProtVer 14 high-byte-first
- ArtDMX: 0x5000, Header 18 Bytes, Länge gerade, 2..512
- ArtPoll: 0x2000, 14 Bytes
- ArtPollReply: 0x2100, 210 Bytes, Style 0x02 StMedia, Port 0x1936
"""

from __future__ import annotations

import struct

import pytest
from hms_artnet.packets import (
    ARTNET_ID,
    UDP_PORT,
    build_artpoll_reply,
    build_dmx,
    build_poll,
    parse_dmx,
    parse_poll,
    parse_poll_reply,
)

# ---------- Header ----------

def test_artnet_id_is_8_bytes_with_null() -> None:
    assert ARTNET_ID == b"Art-Net\x00"
    assert len(ARTNET_ID) == 8


def test_dmx_opcode_little_endian() -> None:
    packet = build_dmx(0, b"\x00\x00")
    assert packet[8:10] == b"\x00\x50"  # 0x5000 low byte first


def test_protocol_version_14_high_byte_first() -> None:
    packet = build_dmx(0, b"\x00\x00")
    assert packet[10:12] == b"\x00\x0e"  # ProtVer 14


# ---------- ArtDMX ----------

def test_dmx_roundtrip() -> None:
    data = bytes(range(64))
    packet = build_dmx(universe=5, data=data, sequence=7, physical=1)
    parsed = parse_dmx(packet)
    assert parsed is not None
    assert parsed.universe == 5
    assert parsed.sequence == 7
    assert parsed.physical == 1
    assert parsed.data == data


def test_dmx_length_is_even_and_header_18_bytes() -> None:
    packet = build_dmx(0, b"\x01\x02\x03")  # ungerade → aufgerundet
    (length,) = struct.unpack_from(">H", packet, 16)
    assert length == 4  # auf gerade aufgerundet
    assert len(packet) == 18 + length


def test_dmx_universe_encoding_net_and_subuni() -> None:
    # universe = Net<<8 | SubUni; Net 7 Bit, SubUni 8 Bit
    packet = build_dmx(universe=(3 << 8) | 0x42, data=b"\x00\x00")
    assert packet[14] == 0x42  # SubUni
    assert packet[15] == 0x03  # Net
    parsed = parse_dmx(packet)
    assert parsed is not None
    assert parsed.universe == (3 << 8) | 0x42


def test_dmx_rejects_short_data() -> None:
    with pytest.raises(ValueError):
        build_dmx(0, b"\x00")  # unter 2 Bytes
    with pytest.raises(ValueError):
        build_dmx(0, b"\x00" * 513)  # über 512


def test_dmx_rejects_invalid_universe() -> None:
    with pytest.raises(ValueError):
        build_dmx(0x8000, b"\x00\x00")  # 15 Bit max


def test_parse_dmx_rejects_garbage() -> None:
    assert parse_dmx(b"") is None
    assert parse_dmx(b"\x00" * 10) is None
    wrong_opcode = ARTNET_ID + struct.pack("<H", 0x9999) + b"\x00\x0e" + b"\x00" * 10
    assert parse_dmx(wrong_opcode) is None


def test_parse_dmx_rejects_old_protocol_version() -> None:
    packet = bytearray(build_dmx(0, b"\x00\x00"))
    packet[10:12] = b"\x00\x0c"  # ProtVer 12
    assert parse_dmx(bytes(packet)) is None


def test_parse_dmx_rejects_truncated_payload() -> None:
    packet = build_dmx(0, b"\x00" * 64)
    assert parse_dmx(packet[:-32]) is None  # abgeschnittene Daten


# ---------- ArtPoll ----------

def test_poll_is_14_bytes_and_roundtrips() -> None:
    packet = build_poll(talk_to_me=0x02, priority=0x0A)
    assert len(packet) == 14
    parsed = parse_poll(packet)
    assert parsed is not None
    assert parsed.talk_to_me == 0x02
    assert parsed.priority == 0x0A


def test_poll_opcode() -> None:
    assert build_poll()[8:10] == b"\x00\x20"  # 0x2000 low byte first


def test_parse_poll_accepts_extended_packets() -> None:
    packet = build_poll() + b"\x00" * 10  # größere Pakete müssen akzeptiert werden
    assert parse_poll(packet) is not None


def test_parse_poll_rejects_wrong_opcode() -> None:
    packet = build_dmx(0, b"\x00\x00")
    assert parse_poll(packet) is None


# ---------- ArtPollReply ----------

def test_pollreply_exactly_210_bytes() -> None:
    reply = build_artpoll_reply(
        ip=b"\xc0\xa8\x01\x2a",
        short_name="HMS ME",
        long_name="HMS MediaEngine Render Node",
    )
    assert len(reply) == 210


def test_pollreply_roundtrip_as_media_server() -> None:
    reply = build_artpoll_reply(
        ip=b"\xc0\xa8\x01\x2a",
        short_name="HMS ME",
        long_name="HMS MediaEngine Render Node A",
        node_report="Media Server Ready",
        mac=b"\xde\xad\xbe\xef\x00\x01",
    )
    info = parse_poll_reply(reply)
    assert info is not None
    assert info.ip == "192.168.1.42"
    assert info.short_name == "HMS ME"
    assert info.long_name == "HMS MediaEngine Render Node A"
    assert info.style == 0x02  # StMedia (Media Server, §3.4)
    assert info.mac == b"\xde\xad\xbe\xef\x00\x01"
    assert info.bind_index == 1


def test_pollreply_port_is_6454() -> None:
    reply = build_artpoll_reply(ip=b"\x7f\x00\x00\x01", short_name="x", long_name="y")
    (port,) = struct.unpack_from(">H", reply, 14)
    assert port == UDP_PORT == 0x1936


def test_pollreply_node_report_format() -> None:
    reply = build_artpoll_reply(
        ip=b"\x7f\x00\x00\x01",
        short_name="x",
        long_name="y",
        report_code=0x0000,
        error_count=3,
    )
    info = parse_poll_reply(reply)
    assert info is not None
    assert info.node_report.startswith("#0000 [0003]")  # Format '#hhhh [hhhh] text'


def test_pollreply_names_null_terminated_and_truncated() -> None:
    reply = build_artpoll_reply(
        ip=b"\x7f\x00\x00\x01",
        short_name="S" * 40,   # > 17 → auf 17 gekürzt
        long_name="L" * 100,   # > 63 → auf 63 gekürzt
    )
    info = parse_poll_reply(reply)
    assert info is not None
    assert info.short_name == "S" * 17
    assert info.long_name == "L" * 63


def test_pollreply_rejects_short_packets() -> None:
    assert parse_poll_reply(b"\x00" * 100) is None