thomas
1f110866f5
CI / backend (pull_request) Canceled after 0s
CI / shell (pull_request) Canceled after 0s
CI / frontend (pull_request) Canceled after 0s
CI / arm64-smoke (pull_request) Canceled after 0s
CI / backend (push) Canceled after 0s
CI / shell (push) Canceled after 0s
CI / frontend (push) Canceled after 0s
CI / arm64-smoke (push) Canceled after 0s
119 lines
3.6 KiB
Python
119 lines
3.6 KiB
Python
import struct
|
|
|
|
import pytest
|
|
from app.dmx.backends import (
|
|
ARTNET_HEADER,
|
|
ARTNET_OPCODE_DMX,
|
|
ARTNET_OPCODE_POLL_REPLY,
|
|
ARTNET_PORT,
|
|
ArtNetDmxBackend,
|
|
build_artnet_dmx_packet,
|
|
parse_artnet_poll_reply,
|
|
)
|
|
from app.dmx.frame import DmxFrame
|
|
|
|
|
|
def test_build_artnet_packet_sends_full_512_frame_to_expected_universe() -> None:
|
|
values = [0] * 512
|
|
values[0] = 255
|
|
values[9] = 64
|
|
packet = build_artnet_dmx_packet(2, values, sequence=9)
|
|
|
|
assert packet.startswith(ARTNET_HEADER)
|
|
assert struct.unpack_from("<H", packet, 8)[0] == ARTNET_OPCODE_DMX
|
|
assert packet[12] == 9
|
|
assert packet[14] == 1
|
|
assert packet[15] == 0
|
|
assert struct.unpack_from(">H", packet, 16)[0] == 512
|
|
assert len(packet) == 530
|
|
assert packet[18] == 255
|
|
assert packet[27] == 64
|
|
|
|
|
|
def test_parse_artnet_poll_reply_extracts_node_identity() -> None:
|
|
packet = bytearray(239)
|
|
packet[0:8] = ARTNET_HEADER
|
|
struct.pack_into("<H", packet, 8, ARTNET_OPCODE_POLL_REPLY)
|
|
packet[10:14] = bytes([192, 168, 2, 55])
|
|
packet[18] = 0
|
|
packet[19] = 1
|
|
packet[26 : 26 + len(b"WLED Node\x00")] = b"WLED Node\x00"
|
|
packet[44 : 44 + len(b"Paravega Test Node\x00")] = b"Paravega Test Node\x00"
|
|
struct.pack_into(">H", packet, 172, 4)
|
|
packet[190] = 3
|
|
|
|
node = parse_artnet_poll_reply(bytes(packet))
|
|
|
|
assert node is not None
|
|
assert node.ip == "192.168.2.55"
|
|
assert node.short_name == "WLED Node"
|
|
assert node.long_name == "Paravega Test Node"
|
|
assert node.port_count == 4
|
|
assert node.raw_port_address == 3
|
|
assert node.label == "Paravega Test Node (192.168.2.55)"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artnet_backend_sends_udp_packet_and_updates_status(monkeypatch) -> None:
|
|
sent_packets: list[tuple[bytes, tuple[str, int]]] = []
|
|
|
|
class FakeSocket:
|
|
def setsockopt(self, *_args: object) -> None:
|
|
return
|
|
|
|
def sendto(self, packet: bytes, address: tuple[str, int]) -> None:
|
|
sent_packets.append((packet, address))
|
|
|
|
def close(self) -> None:
|
|
return
|
|
|
|
backend = ArtNetDmxBackend(
|
|
universe=4,
|
|
target_host="192.168.2.77",
|
|
output_port="Paravega stue",
|
|
)
|
|
monkeypatch.setattr(backend, "_socket", FakeSocket())
|
|
frame = DmxFrame(universe=4)
|
|
frame.set_channel(1, 200, "scene")
|
|
frame.set_channel(4, 99, "scene")
|
|
|
|
await backend.send_frame(frame)
|
|
|
|
status = backend.get_status()
|
|
assert status.connected is True
|
|
assert status.degraded is False
|
|
assert status.frames_sent == 1
|
|
assert status.send_errors == 0
|
|
assert status.selected_universe == 4
|
|
assert status.selected_output_port == "Paravega stue"
|
|
assert sent_packets[0][1] == ("192.168.2.77", ARTNET_PORT)
|
|
assert sent_packets[0][0][18] == 200
|
|
assert sent_packets[0][0][21] == 99
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_artnet_backend_marks_degraded_on_socket_error(monkeypatch) -> None:
|
|
class FakeSocket:
|
|
def setsockopt(self, *_args: object) -> None:
|
|
return
|
|
|
|
def sendto(self, _packet: bytes, _address: tuple[str, int]) -> None:
|
|
raise OSError("Network unreachable")
|
|
|
|
def close(self) -> None:
|
|
return
|
|
|
|
backend = ArtNetDmxBackend(universe=1, target_host="192.168.2.90")
|
|
monkeypatch.setattr(backend, "_socket", FakeSocket())
|
|
frame = DmxFrame()
|
|
frame.set_channel(1, 255, "scene")
|
|
|
|
with pytest.raises(RuntimeError, match="Network unreachable"):
|
|
await backend.send_frame(frame)
|
|
|
|
status = backend.get_status()
|
|
assert status.connected is False
|
|
assert status.degraded is True
|
|
assert status.send_errors == 1
|
|
assert status.last_error == "Network unreachable"
|