241 lines
9.7 KiB
Python
241 lines
9.7 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import struct
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from lumaops_backend.config import Settings
|
|
from lumaops_backend.connectors.base import DeviceState, RGBColor
|
|
from lumaops_backend.connectors.openrgb.adapter import OpenRGBAdapter, expand_colors
|
|
from lumaops_backend.connectors.openrgb.protocol import (
|
|
HEADER,
|
|
ModeFlag,
|
|
PacketId,
|
|
pack_color,
|
|
pack_header,
|
|
pack_string,
|
|
parse_header,
|
|
)
|
|
|
|
|
|
def controller_packet(
|
|
*,
|
|
active_mode: int = 0,
|
|
controller_colors: list[RGBColor] | None = None,
|
|
static_color: RGBColor | None = None,
|
|
) -> bytes:
|
|
color = static_color or RGBColor(red=16, green=32, blue=48)
|
|
led_colors = controller_colors or [RGBColor(red=16, green=32, blue=48)] * 2
|
|
direct_mode = (
|
|
pack_string("Direct")
|
|
+ struct.pack("<iI", 0, int(ModeFlag.HAS_PER_LED_COLOR))
|
|
+ struct.pack("<IIIIIIIIII", 0, 0, 0, 0, 0, 0, 0, 0, 0, 1)
|
|
+ struct.pack("<H", 0)
|
|
)
|
|
static_mode = (
|
|
pack_string("Static")
|
|
+ struct.pack("<iI", 7, int(ModeFlag.HAS_BRIGHTNESS | ModeFlag.HAS_MODE_SPECIFIC_COLOR))
|
|
+ struct.pack("<IIIIIIIIII", 0, 0, 0, 255, 1, 1, 0, 128, 0, 1)
|
|
+ struct.pack("<H", 1)
|
|
+ pack_color(color)
|
|
)
|
|
custom_mode = (
|
|
pack_string("Custom")
|
|
+ struct.pack("<iI", 1, int(ModeFlag.HAS_PER_LED_COLOR | ModeFlag.AUTOMATIC_SAVE))
|
|
+ struct.pack("<IIIIIIIIII", 0, 0, 0, 0, 0, 0, 0, 0, 0, 1)
|
|
+ struct.pack("<H", 0)
|
|
)
|
|
zone = (
|
|
pack_string("Main")
|
|
+ struct.pack("<iIIIH", 1, 0, 120, 2, 0)
|
|
+ struct.pack("<H", 0)
|
|
+ struct.pack("<I", 0)
|
|
)
|
|
leds = pack_string("LED 1") + struct.pack("<I", 1) + pack_string("LED 2") + struct.pack("<I", 2)
|
|
body = (
|
|
struct.pack("<i", 0)
|
|
+ pack_string("SDK Fixture")
|
|
+ pack_string("LumaOps")
|
|
+ pack_string("Integration test")
|
|
+ pack_string("1.2.3")
|
|
+ pack_string("SERIAL-1")
|
|
+ pack_string("usb:1-2")
|
|
+ struct.pack("<H", 3)
|
|
+ struct.pack("<i", active_mode)
|
|
+ direct_mode
|
|
+ custom_mode
|
|
+ static_mode
|
|
+ struct.pack("<H", 1)
|
|
+ zone
|
|
+ struct.pack("<H", 2)
|
|
+ leds
|
|
+ struct.pack("<H", 2)
|
|
+ b"".join(pack_color(item) for item in led_colors)
|
|
+ struct.pack("<H", 0)
|
|
+ struct.pack("<I", 1)
|
|
)
|
|
return struct.pack("<I", len(body) + 4) + body
|
|
|
|
|
|
def test_uniform_color_state_survives_a_changed_led_count() -> None:
|
|
color = RGBColor(red=86, green=96, blue=255)
|
|
|
|
assert expand_colors([color] * 6, 23) == [color] * 23
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_real_adapter_handshake_inventory_and_write(tmp_path: Path) -> None:
|
|
seen: list[tuple[int, bytes]] = []
|
|
update_received = asyncio.Event()
|
|
effect_received = asyncio.Event()
|
|
resize_received = asyncio.Event()
|
|
active_mode = 0
|
|
fixture_colors = [RGBColor(red=16, green=32, blue=48)] * 2
|
|
fixture_static_color = RGBColor(red=16, green=32, blue=48)
|
|
|
|
async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
|
|
nonlocal active_mode, fixture_colors, fixture_static_color
|
|
try:
|
|
while True:
|
|
header = parse_header(await reader.readexactly(HEADER.size))
|
|
payload = await reader.readexactly(header.payload_size)
|
|
seen.append((header.packet_id, payload))
|
|
if header.packet_id == PacketId.UPDATE_LEDS:
|
|
count = struct.unpack_from("<H", payload, 4)[0]
|
|
fixture_colors = []
|
|
for index in range(count):
|
|
red, green, blue = struct.unpack_from("<BBB", payload, 6 + index * 4)
|
|
fixture_colors.append(RGBColor(red=red, green=green, blue=blue))
|
|
update_received.set()
|
|
elif header.packet_id == PacketId.UPDATE_ZONE_LEDS:
|
|
zone_index, count = struct.unpack_from("<IH", payload, 4)
|
|
assert zone_index == 0
|
|
fixture_colors = []
|
|
for index in range(count):
|
|
red, green, blue = struct.unpack_from("<BBB", payload, 10 + index * 4)
|
|
fixture_colors.append(RGBColor(red=red, green=green, blue=blue))
|
|
update_received.set()
|
|
elif header.packet_id == PacketId.UPDATE_MODE:
|
|
active_mode = struct.unpack_from("<i", payload, 4)[0]
|
|
if active_mode == 2:
|
|
red, green, blue = struct.unpack_from("<BBB", payload, len(payload) - 4)
|
|
fixture_static_color = RGBColor(red=red, green=green, blue=blue)
|
|
effect_received.set()
|
|
elif header.packet_id == PacketId.SET_CUSTOM_MODE:
|
|
active_mode = 0
|
|
elif header.packet_id == PacketId.RESIZE_ZONE:
|
|
resize_received.set()
|
|
response: bytes | None = None
|
|
if header.packet_id == PacketId.REQUEST_PROTOCOL_VERSION:
|
|
response = struct.pack("<I", 5)
|
|
elif header.packet_id == PacketId.REQUEST_CONTROLLER_COUNT:
|
|
response = struct.pack("<I", 1)
|
|
elif header.packet_id == PacketId.REQUEST_CONTROLLER_DATA:
|
|
response = controller_packet(
|
|
active_mode=active_mode,
|
|
controller_colors=fixture_colors,
|
|
static_color=fixture_static_color,
|
|
)
|
|
if response is not None:
|
|
writer.write(
|
|
pack_header(header.device_index, header.packet_id, len(response)) + response
|
|
)
|
|
await writer.drain()
|
|
except asyncio.IncompleteReadError:
|
|
pass
|
|
finally:
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
|
|
server = await asyncio.start_server(handle, "127.0.0.1", 0)
|
|
port = server.sockets[0].getsockname()[1]
|
|
settings = Settings(
|
|
LUMAOPS_ENV="test",
|
|
auth_enabled=False,
|
|
openrgb_host="127.0.0.1",
|
|
openrgb_port=port,
|
|
openrgb_connect_timeout=1,
|
|
openrgb_command_timeout=1,
|
|
config_dir=tmp_path / "config",
|
|
openrgb_config_dir=tmp_path / "openrgb",
|
|
data_dir=tmp_path / "data",
|
|
logs_dir=tmp_path / "logs",
|
|
database_url=f"sqlite:///{tmp_path / 'test.db'}",
|
|
)
|
|
adapter = OpenRGBAdapter(settings)
|
|
try:
|
|
devices = await adapter.inventory()
|
|
assert len(devices) == 1
|
|
assert devices[0].name == "SDK Fixture"
|
|
state = await adapter.set_state(
|
|
devices[0].external_id,
|
|
DeviceState(
|
|
colors=[RGBColor(red=2, green=4, blue=8)],
|
|
brightness=100,
|
|
mode_index=1,
|
|
),
|
|
)
|
|
assert state.colors == [RGBColor(red=2, green=4, blue=8)] * 2
|
|
assert state.mode == "Direct"
|
|
assert state.mode_index == 0
|
|
assert state.brightness is None
|
|
await adapter.resize_zone(devices[0].external_id, 0, 24)
|
|
await asyncio.wait_for(resize_received.wait(), timeout=1)
|
|
assert [packet_id for packet_id, _payload in seen[:4]] == [
|
|
PacketId.REQUEST_PROTOCOL_VERSION,
|
|
PacketId.SET_CLIENT_NAME,
|
|
PacketId.REQUEST_CONTROLLER_COUNT,
|
|
PacketId.REQUEST_CONTROLLER_DATA,
|
|
]
|
|
assert any(packet_id == PacketId.SET_CUSTOM_MODE for packet_id, _payload in seen)
|
|
assert any(packet_id == PacketId.UPDATE_LEDS for packet_id, _payload in seen)
|
|
assert (PacketId.RESIZE_ZONE, struct.pack("<ii", 0, 24)) in seen
|
|
|
|
zone_start = len(seen)
|
|
zone_color = RGBColor(red=9, green=18, blue=27)
|
|
zone_state = await adapter.set_state(
|
|
devices[0].external_id,
|
|
DeviceState(colors=[zone_color], zone_index=0),
|
|
)
|
|
zone_packets = [packet_id for packet_id, _payload in seen[zone_start:]]
|
|
assert zone_state.colors == [zone_color] * 2
|
|
assert PacketId.UPDATE_ZONE_LEDS in zone_packets
|
|
assert PacketId.SET_CUSTOM_MODE not in zone_packets
|
|
|
|
effect_start = len(seen)
|
|
effect_received.clear()
|
|
effect_state = await adapter.set_state(
|
|
devices[0].external_id,
|
|
DeviceState(
|
|
colors=[RGBColor(red=90, green=45, blue=180)],
|
|
brightness=80,
|
|
mode_index=2,
|
|
),
|
|
)
|
|
await asyncio.wait_for(effect_received.wait(), timeout=1)
|
|
effect_packets = [packet_id for packet_id, _payload in seen[effect_start:]]
|
|
assert effect_state.mode == "Static"
|
|
assert effect_state.mode_index == 2
|
|
assert effect_state.colors == [RGBColor(red=90, green=45, blue=180)]
|
|
assert PacketId.UPDATE_MODE in effect_packets
|
|
assert PacketId.SET_CUSTOM_MODE not in effect_packets
|
|
assert PacketId.UPDATE_LEDS not in effect_packets
|
|
|
|
direct_start = len(seen)
|
|
direct_color = RGBColor(red=24, green=48, blue=96)
|
|
direct_state = await adapter.set_state(
|
|
devices[0].external_id,
|
|
DeviceState(colors=[direct_color]),
|
|
)
|
|
direct_packets = [packet_id for packet_id, _payload in seen[direct_start:]]
|
|
assert direct_state.mode == "Direct"
|
|
assert direct_state.colors == [direct_color] * 2
|
|
assert PacketId.SET_CUSTOM_MODE in direct_packets
|
|
assert PacketId.UPDATE_LEDS in direct_packets
|
|
finally:
|
|
await adapter.stop()
|
|
server.close()
|
|
await server.wait_closed()
|