404 lines
16 KiB
Python
404 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from lumaops_backend.connectors.base import DeviceState, RGBColor
|
|
from lumaops_backend.connectors.mock import MockOpenRGBAdapter
|
|
from lumaops_backend.main import create_app
|
|
from lumaops_backend.services.commands import CommandService
|
|
|
|
|
|
def test_direct_mode_without_explicit_color_preserves_last_nonzero_color() -> None:
|
|
color = RGBColor(red=86, green=96, blue=255)
|
|
effective = CommandService._effective_state(
|
|
{
|
|
"desired_state": DeviceState(power=True, colors=[color]).model_dump(mode="json"),
|
|
"modes": [{"index": 0, "name": "Direct"}],
|
|
},
|
|
DeviceState(power=True, mode_index=0),
|
|
)
|
|
|
|
assert effective.colors == [color]
|
|
assert effective.mode_index == 0
|
|
|
|
|
|
def test_main_device_command_flow(client: TestClient) -> None:
|
|
discovery = client.post("/api/v1/discovery")
|
|
assert discovery.status_code == 200
|
|
assert discovery.json()["found_count"] == 2
|
|
|
|
devices_response = client.get("/api/v1/devices")
|
|
assert devices_response.status_code == 200
|
|
devices = devices_response.json()["items"]
|
|
assert len(devices) == 2
|
|
device = next(item for item in devices if item["fingerprint"] == "mock-mainboard")
|
|
|
|
request = {"state": {"power": True, "colors": [{"red": 20, "green": 40, "blue": 60}]}}
|
|
command = client.post(
|
|
f"/api/v1/devices/{device['id']}/state",
|
|
json=request,
|
|
headers={"Idempotency-Key": "main-flow-1"},
|
|
)
|
|
assert command.status_code == 200
|
|
assert command.json()["status"] == "succeeded"
|
|
duplicate = client.post(
|
|
f"/api/v1/devices/{device['id']}/state",
|
|
json=request,
|
|
headers={"Idempotency-Key": "main-flow-1"},
|
|
)
|
|
assert duplicate.json()["id"] == command.json()["id"]
|
|
|
|
detail = client.get(f"/api/v1/devices/{device['id']}").json()
|
|
assert detail["state"]["colors"][0] == {"red": 20, "green": 40, "blue": 60}
|
|
|
|
resized = client.put(
|
|
f"/api/v1/devices/{device['id']}/zones/1",
|
|
json={"led_count": 24},
|
|
)
|
|
assert resized.status_code == 200
|
|
assert resized.json()["zones"][1]["led_count"] == 24
|
|
assert resized.json()["led_count"] == 32
|
|
|
|
zone_color = {"red": 10, "green": 80, "blue": 160}
|
|
zone_command = client.post(
|
|
f"/api/v1/devices/{device['id']}/zones/1/state",
|
|
json={"state": {"colors": [zone_color]}},
|
|
)
|
|
assert zone_command.status_code == 200
|
|
assert zone_command.json()["zone_index"] == 1
|
|
assert zone_command.json()["led_count"] == 120
|
|
assert client.get(f"/api/v1/devices/{device['id']}").json()["zones"][1]["led_count"] == 120
|
|
|
|
commands = client.get("/api/v1/commands").json()
|
|
audit = client.get("/api/v1/audit").json()
|
|
assert commands["total"] == 2
|
|
assert audit["total"] >= 1
|
|
assert any(item["action"] == "device.resize_zone" for item in audit["items"])
|
|
|
|
|
|
def test_desired_state_survives_connector_restart_and_power_cycle(client: TestClient) -> None:
|
|
client.post("/api/v1/discovery")
|
|
device = next(
|
|
item
|
|
for item in client.get("/api/v1/devices").json()["items"]
|
|
if item["fingerprint"] == "mock-mainboard"
|
|
)
|
|
target = f"/api/v1/devices/{device['id']}"
|
|
color = {"red": 81, "green": 109, "blue": 245}
|
|
|
|
applied = client.post(
|
|
f"{target}/state",
|
|
json={"state": {"power": True, "colors": [color]}},
|
|
)
|
|
assert applied.status_code == 200
|
|
detail = client.get(target).json()
|
|
assert detail["desired_state"]["colors"][0] == color
|
|
|
|
connector = client.app.state.context.registry.get("openrgb-mock")
|
|
assert isinstance(connector, MockOpenRGBAdapter)
|
|
connector._devices["mock-mainboard"].state = DeviceState( # noqa: SLF001
|
|
power=False,
|
|
brightness=72,
|
|
colors=[RGBColor(red=0, green=0, blue=0)] * 12,
|
|
mode="Direct",
|
|
mode_index=0,
|
|
speed=5,
|
|
)
|
|
|
|
commands_before = client.get("/api/v1/commands").json()["total"]
|
|
discovery = client.post("/api/v1/discovery")
|
|
assert discovery.status_code == 200
|
|
restored = client.get(target).json()
|
|
assert restored["state"]["power"] is True
|
|
assert restored["state"]["colors"]
|
|
assert all(entry == color for entry in restored["state"]["colors"])
|
|
commands_after = client.get("/api/v1/commands").json()
|
|
assert commands_after["total"] == commands_before + 1
|
|
assert commands_after["items"][0]["actor"] == "system-restore"
|
|
|
|
assert client.post("/api/v1/discovery").status_code == 200
|
|
assert client.get("/api/v1/commands").json()["total"] == commands_after["total"]
|
|
|
|
assert client.post(f"{target}/state", json={"state": {"power": False}}).status_code == 200
|
|
powered_off = client.get(target).json()
|
|
assert powered_off["state"]["power"] is False
|
|
assert powered_off["desired_state"]["colors"]
|
|
assert all(entry == color for entry in powered_off["desired_state"]["colors"])
|
|
|
|
assert client.post(f"{target}/state", json={"state": {"power": True}}).status_code == 200
|
|
powered_on = client.get(target).json()
|
|
assert powered_on["state"]["power"] is True
|
|
assert powered_on["state"]["colors"]
|
|
assert all(entry == color for entry in powered_on["state"]["colors"])
|
|
|
|
|
|
def test_explicit_colorless_effect_is_not_replaced_by_power_restore(
|
|
client: TestClient,
|
|
) -> None:
|
|
device = client.get("/api/v1/devices").json()["items"][0]
|
|
client.post(
|
|
f"/api/v1/devices/{device['id']}/state",
|
|
json={"state": {"power": True, "colors": [{"red": 24, "green": 48, "blue": 96}]}},
|
|
)
|
|
|
|
effect = client.post(
|
|
f"/api/v1/devices/{device['id']}/state",
|
|
json={"state": {"power": True, "mode_index": 2}},
|
|
)
|
|
|
|
assert effect.status_code == 200
|
|
assert effect.json()["state"]["mode_index"] == 2
|
|
|
|
|
|
def test_startup_reconciles_desired_state_once_before_ready(settings) -> None:
|
|
color = {"red": 24, "green": 96, "blue": 192}
|
|
device_id = ""
|
|
with TestClient(create_app(settings)) as first_client:
|
|
device = next(
|
|
item
|
|
for item in first_client.get("/api/v1/devices").json()["items"]
|
|
if item["fingerprint"] == "mock-mainboard"
|
|
)
|
|
device_id = device["id"]
|
|
applied = first_client.post(
|
|
f"/api/v1/devices/{device_id}/state",
|
|
json={"state": {"power": True, "colors": [color]}},
|
|
)
|
|
assert applied.status_code == 200
|
|
|
|
with TestClient(create_app(settings)) as restarted_client:
|
|
assert restarted_client.get("/health/ready").json() == {"status": "ready"}
|
|
restored = restarted_client.get(f"/api/v1/devices/{device_id}").json()
|
|
assert restored["state"]["colors"]
|
|
assert all(entry == color for entry in restored["state"]["colors"])
|
|
commands = restarted_client.get("/api/v1/commands?limit=100").json()["items"]
|
|
restores = [
|
|
command
|
|
for command in commands
|
|
if command["actor"] == "system-restore" and command["target_id"] == device_id
|
|
]
|
|
assert len(restores) == 1
|
|
assert restores[0]["status"] == "succeeded"
|
|
|
|
|
|
def test_rooms_groups_and_scenes(client: TestClient) -> None:
|
|
client.post("/api/v1/discovery")
|
|
devices = client.get("/api/v1/devices").json()["items"]
|
|
room = client.post("/api/v1/rooms", json={"name": "Werkplek", "sort_order": 1})
|
|
assert room.status_code == 201
|
|
first = devices[0]
|
|
assert (
|
|
client.patch(
|
|
f"/api/v1/devices/{first['id']}", json={"room_id": room.json()["id"]}
|
|
).status_code
|
|
== 200
|
|
)
|
|
group = client.post(
|
|
"/api/v1/groups",
|
|
json={"name": "Bureau", "device_ids": [item["id"] for item in devices]},
|
|
)
|
|
assert group.status_code == 201
|
|
tagged = client.patch(f"/api/v1/devices/{devices[1]['id']}", json={"tags": ["bureau", "rgb"]})
|
|
assert tagged.status_code == 200
|
|
assert tagged.json()["tags"] == ["bureau", "rgb"]
|
|
dynamic_group = client.post(
|
|
"/api/v1/groups",
|
|
json={"name": "Met tag", "dynamic_query": {"tags": ["bureau"], "match": "all"}},
|
|
)
|
|
assert dynamic_group.status_code == 201
|
|
assert [item["id"] for item in dynamic_group.json()["devices"]] == [devices[1]["id"]]
|
|
assert client.get("/api/v1/tags").json()[0]["device_count"] == 1
|
|
scene = client.post(
|
|
"/api/v1/scenes",
|
|
json={
|
|
"name": "Focus",
|
|
"favorite": True,
|
|
"items": [
|
|
{
|
|
"target_type": "group",
|
|
"target_id": group.json()["id"],
|
|
"state": {"power": True, "colors": [{"red": 42, "green": 80, "blue": 180}]},
|
|
}
|
|
],
|
|
},
|
|
)
|
|
assert scene.status_code == 201
|
|
preview = client.get(f"/api/v1/scenes/{scene.json()['id']}/preview").json()
|
|
assert preview["device_count"] == 2
|
|
applied = client.post(
|
|
f"/api/v1/scenes/{scene.json()['id']}/apply", json={"rollback_on_failure": True}
|
|
)
|
|
assert applied.status_code == 200
|
|
assert applied.json()["status"] == "succeeded"
|
|
|
|
duplicate = client.post(f"/api/v1/scenes/{scene.json()['id']}/duplicate", json={})
|
|
assert duplicate.status_code == 201
|
|
assert duplicate.json()["name"] == "Focus (kopie)"
|
|
assert len(duplicate.json()["items"]) == 1
|
|
|
|
exported = client.get(f"/api/v1/scenes/{scene.json()['id']}/export")
|
|
assert exported.status_code == 200
|
|
assert exported.json()["format"] == "lumaops-scene-v1"
|
|
imported = client.post("/api/v1/scenes/import", json=exported.json())
|
|
assert imported.status_code == 201
|
|
assert imported.json()["name"] == "Focus"
|
|
assert client.get("/api/v1/scenes").json()["total"] == 3
|
|
|
|
|
|
def test_physical_dram_family_groups_modules_and_supports_group_commands(
|
|
client: TestClient,
|
|
) -> None:
|
|
devices = client.get("/api/v1/devices").json()["items"]
|
|
device_ids = [device["id"] for device in devices]
|
|
with client.app.state.context.database.connection() as conn:
|
|
conn.execute(
|
|
"UPDATE devices SET device_type='dram', name='Corsair Vengeance RGB Pro SL DDR4', "
|
|
"vendor='Corsair', model='Corsair DRAM RGB Device', state_json=(SELECT state_json "
|
|
"FROM devices ORDER BY controller_index LIMIT 1) WHERE id IN (?, ?)",
|
|
tuple(device_ids),
|
|
)
|
|
|
|
groups = client.get("/api/v1/device-groups")
|
|
assert groups.status_code == 200
|
|
assert len(groups.json()) == 1
|
|
group = groups.json()[0]
|
|
assert group["device_type"] == "dram"
|
|
assert group["module_count"] == 2
|
|
assert group["mixed"] is False
|
|
assert [device["controller_index"] for device in group["devices"]] == [0, 1]
|
|
|
|
color = {"red": 34, "green": 102, "blue": 204}
|
|
applied = client.post(
|
|
f"/api/v1/device-groups/{group['id']}/state",
|
|
json={"state": {"power": True, "colors": [color]}},
|
|
)
|
|
assert applied.status_code == 200
|
|
assert applied.json()["status"] == "succeeded"
|
|
assert [result["status"] for result in applied.json()["results"]] == [
|
|
"succeeded",
|
|
"succeeded",
|
|
]
|
|
for device_id in device_ids:
|
|
detail = client.get(f"/api/v1/devices/{device_id}").json()
|
|
assert detail["desired_state"]["colors"][0] == color
|
|
|
|
|
|
def test_setup_health_backup_and_diagnostics(client: TestClient) -> None:
|
|
setup = client.post("/api/v1/setup/inspect")
|
|
assert setup.status_code == 200
|
|
assert "storage" in setup.json()
|
|
complete = client.post(
|
|
"/api/v1/setup/complete",
|
|
json={"appdata_confirmed": True, "backup_location_confirmed": True},
|
|
)
|
|
assert complete.json()["completed"] is True
|
|
|
|
assert client.get("/health/live").status_code == 200
|
|
assert client.get("/health/ready").status_code == 200
|
|
health = client.get("/api/v1/health")
|
|
assert health.status_code == 200
|
|
|
|
backup = client.post("/api/v1/backups")
|
|
assert backup.status_code == 201
|
|
assert backup.json()["name"].endswith(".db")
|
|
diagnostic = client.get("/api/v1/diagnostics/export")
|
|
assert diagnostic.status_code == 200
|
|
assert diagnostic.content.startswith(b"PK")
|
|
|
|
|
|
def test_setup_complete_accepts_legacy_backup_field(client: TestClient) -> None:
|
|
complete = client.post(
|
|
"/api/v1/setup/complete",
|
|
json={"appdata_confirmed": True, "backup_confirmed": True},
|
|
)
|
|
|
|
assert complete.status_code == 200
|
|
assert complete.json()["completed"] is True
|
|
|
|
|
|
def test_resource_lifecycle_group_capabilities_and_automation_history(
|
|
client: TestClient,
|
|
) -> None:
|
|
client.post("/api/v1/discovery")
|
|
devices = client.get("/api/v1/devices").json()["items"]
|
|
|
|
room = client.post("/api/v1/rooms", json={"name": "Studio", "description": "Boven"}).json()
|
|
updated_room = client.put(
|
|
f"/api/v1/rooms/{room['id']}",
|
|
json={"name": "Werkstudio", "description": "Bovenverdieping"},
|
|
)
|
|
assert updated_room.status_code == 200
|
|
assert updated_room.json()["description"] == "Bovenverdieping"
|
|
client.patch(f"/api/v1/devices/{devices[0]['id']}", json={"room_id": room["id"]})
|
|
assert client.delete(f"/api/v1/rooms/{room['id']}").status_code == 204
|
|
assert client.get(f"/api/v1/devices/{devices[0]['id']}").json()["room_id"] is None
|
|
|
|
group = client.post(
|
|
"/api/v1/groups",
|
|
json={"name": "Alles", "device_ids": [device["id"] for device in devices]},
|
|
).json()
|
|
group_result = client.post(
|
|
f"/api/v1/groups/{group['id']}/state",
|
|
json={"state": {"speed": 3}},
|
|
)
|
|
assert group_result.status_code == 200
|
|
assert {item["status"] for item in group_result.json()["results"]} == {
|
|
"succeeded",
|
|
"skipped",
|
|
}
|
|
|
|
scene = client.post("/api/v1/scenes", json={"name": "Leeg", "items": []}).json()
|
|
automation = client.post(
|
|
"/api/v1/automations",
|
|
json={
|
|
"name": "Avond",
|
|
"description": "Dagelijkse rustige scène",
|
|
"trigger": {"type": "time", "at": "20:00", "weekdays": [0, 2, 4]},
|
|
"actions": [{"type": "scene", "scene_id": scene["id"]}],
|
|
"timezone": "Europe/Brussels",
|
|
"cooldown_seconds": 0,
|
|
"conflict_key": "woonkamer",
|
|
},
|
|
)
|
|
assert automation.status_code == 201
|
|
assert automation.json()["description"] == "Dagelijkse rustige scène"
|
|
run = client.post(f"/api/v1/automations/{automation.json()['id']}/run")
|
|
assert run.status_code == 200
|
|
assert run.json()["status"] == "succeeded"
|
|
history = client.get(f"/api/v1/automations/{automation.json()['id']}/runs").json()
|
|
assert history["total"] == 1
|
|
assert history["items"][0]["status"] == "succeeded"
|
|
|
|
audit = client.get("/api/v1/audit?limit=500").json()["items"]
|
|
assert any(item["action"] == "api.create_room" for item in audit)
|
|
assert any(item["action"] == "api.create_automation" for item in audit)
|
|
|
|
|
|
def test_automation_requests_reject_invalid_schedules_and_actions(client: TestClient) -> None:
|
|
base = {
|
|
"name": "Ongeldig",
|
|
"description": "Validatiecontrole",
|
|
"trigger": {"type": "time", "at": "20:00", "weekdays": [0]},
|
|
"actions": [{"type": "scene", "scene_id": "scene-id"}],
|
|
"timezone": "Europe/Brussels",
|
|
}
|
|
|
|
invalid_time = client.post(
|
|
"/api/v1/automations",
|
|
json={**base, "trigger": {"type": "time", "at": "25:90", "weekdays": [0]}},
|
|
)
|
|
assert invalid_time.status_code == 422
|
|
assert invalid_time.json()["error"]["code"] == "validation_error"
|
|
|
|
missing_scene = client.post(
|
|
"/api/v1/automations",
|
|
json={**base, "actions": [{"type": "scene", "scene_id": None}]},
|
|
)
|
|
assert missing_scene.status_code == 422
|
|
|
|
invalid_timezone = client.post(
|
|
"/api/v1/automations",
|
|
json={**base, "timezone": "Mars/Olympus_Mons"},
|
|
)
|
|
assert invalid_timezone.status_code == 422
|