from uuid import uuid4 from app.core.dependencies import app_state from app.main import app from fastapi.testclient import TestClient def test_health_endpoint() -> None: with TestClient(app) as client: response = client.get("/api/v1/health") assert response.status_code == 200 payload = response.json() assert payload["status"] == "ok" assert payload["simulator_enabled"] is True def test_trigger_endpoint_returns_202() -> None: with TestClient(app) as client: response = client.post( "/api/v1/triggers/raid", json={"eventType": "raid", "displayName": "Tester", "eventId": "evt-1"}, ) assert response.status_code == 202 assert response.json()["accepted"] is True def test_dmx_status_exposes_backend_counters() -> None: with TestClient(app) as client: response = client.get("/api/v1/dmx/status") assert response.status_code == 200 payload = response.json() assert payload["backend"] in {"simulator", "ola", "artnet"} assert payload["selected_universe"] == 1 assert "selected_output_port" in payload assert payload["frames_sent"] >= 0 assert payload["send_errors"] >= 0 assert "last_successful_frame" in payload def test_dmx_output_config_can_be_saved_and_read_back() -> None: with TestClient(app) as client: original = client.get("/api/v1/dmx/config").json() response = client.put( "/api/v1/dmx/config", json={ "backend": "artnet", "universe": 7, "output_port": "WLED stue", "target_host": "192.168.2.50", }, ) assert response.status_code == 200 payload = response.json() assert payload["backend"] == "artnet" assert payload["universe"] == 7 assert payload["output_port"] == "WLED stue" assert payload["target_host"] == "192.168.2.50" read_back = client.get("/api/v1/dmx/config") assert read_back.status_code == 200 assert read_back.json()["backend"] == "artnet" assert read_back.json()["universe"] == 7 restore_response = client.put("/api/v1/dmx/config", json=original) assert restore_response.status_code == 200 def test_dmx_artnet_discover_endpoint_returns_nodes(monkeypatch) -> None: async def fake_discover(timeout_s: float) -> dict[str, object]: assert timeout_s == 0.25 return { "count": 1, "items": [ { "ip": "192.168.2.60", "short_name": "WLED", "long_name": "WLED Node", "label": "WLED Node (192.168.2.60)", "net": 0, "sub_switch": 0, "port_count": 1, "raw_port_address": 0, } ], } monkeypatch.setattr(app_state.dmx, "discover_artnet", fake_discover) with TestClient(app) as client: response = client.post("/api/v1/dmx/artnet/discover?timeout_s=0.25") assert response.status_code == 200 payload = response.json() assert payload["count"] == 1 assert payload["items"][0]["ip"] == "192.168.2.60" def test_home_assistant_config_can_be_saved_and_mapping_created(monkeypatch) -> None: with TestClient(app) as client: config_response = client.put( "/api/v1/integrations/home-assistant/config", json={ "enabled": True, "base_url": "http://ha.local:8123", "token": "secret-token", "default_universe": 10, }, ) assert config_response.status_code == 200 assert config_response.json()["enabled"] is True assert config_response.json()["base_url"] == "http://ha.local:8123" assert config_response.json()["has_token"] is True assert "token" not in config_response.json() assert config_response.json()["token_mask"] == "********" create_response = client.post( "/api/v1/integrations/home-assistant/mappings", json={ "name": "HA RGB", "universe": 10, "start_address": 1, "fixture_type": "rgb", "entity_id": "light.bar_rgb", "rate_limit_hz": 5, "deadband": 2, "fade_ms": 150, "invert_channel": False, "min_value": 0, "max_value": 255, "enabled": True, "master_dimmer": True, }, ) assert create_response.status_code == 200 mapping = create_response.json() assert mapping["universe"] == 10 assert mapping["channel_span"] == 4 async def fake_test_mapping(mapping_id: int) -> dict[str, object]: return {"status": "sent", "mapping_id": mapping_id} monkeypatch.setattr(app_state.home_assistant, "test_mapping", fake_test_mapping) test_response = client.post(f"/api/v1/integrations/home-assistant/mappings/{mapping['id']}/test") assert test_response.status_code == 200 assert test_response.json()["status"] == "sent" delete_response = client.delete(f"/api/v1/integrations/home-assistant/mappings/{mapping['id']}") assert delete_response.status_code == 200 def test_home_assistant_connection_endpoint_returns_status(monkeypatch) -> None: async def fake_test_connection() -> dict[str, object]: return { "reachable": True, "auth_ok": True, "ha_version": "2026.7.0", "last_error": None, "last_successful_call_at": "2026-07-23T12:45:00+02:00", "last_connection_success_at": "2026-07-23T12:45:00+02:00", } monkeypatch.setattr(app_state.home_assistant, "test_connection", fake_test_connection) with TestClient(app) as client: response = client.post("/api/v1/integrations/home-assistant/test-connection") assert response.status_code == 200 payload = response.json() assert payload["reachable"] is True assert payload["auth_ok"] is True assert payload["ha_version"] == "2026.7.0" def test_bpm_config_can_be_saved_and_read_back() -> None: with TestClient(app) as client: original = client.get("/api/v1/bpm/config").json()["preferred_device"] response = client.put( "/api/v1/bpm/config", json={"preferred_device": "alsa:plughw:CARD=SB,DEV=0"}, ) assert response.status_code == 200 assert response.json()["preferred_device"] == "alsa:plughw:CARD=SB,DEV=0" status_response = client.get("/api/v1/bpm/status") assert status_response.status_code == 200 assert status_response.json()["selected_device"] == "alsa:plughw:CARD=SB,DEV=0" restore_response = client.put( "/api/v1/bpm/config", json={"preferred_device": original}, ) assert restore_response.status_code == 200 def test_restart_service_endpoint_returns_scheduler_result(monkeypatch) -> None: async def fake_restart_service() -> dict[str, object]: return { "accepted": True, "action": "restart-service", "status": "scheduled", "detail": "Servicegenstart planlagt.", } monkeypatch.setattr(app_state.system, "restart_service", fake_restart_service) with TestClient(app) as client: response = client.post("/api/v1/system/restart-service") assert response.status_code == 200 assert response.json()["accepted"] is True assert response.json()["action"] == "restart-service" def test_reboot_host_endpoint_returns_scheduler_result(monkeypatch) -> None: async def fake_reboot_host() -> dict[str, object]: return { "accepted": True, "action": "reboot-host", "status": "scheduled", "detail": "Hostgenstart planlagt.", } monkeypatch.setattr(app_state.system, "reboot_host", fake_reboot_host) with TestClient(app) as client: response = client.post("/api/v1/system/reboot-host") assert response.status_code == 200 assert response.json()["accepted"] is True assert response.json()["action"] == "reboot-host" def test_delete_scene_releases_active_layer() -> None: with TestClient(app) as client: app_state.scenes.active_slug = None app_state.engine.layers.clear() existing_scenes = client.get("/api/v1/scenes").json()["items"] for scene in existing_scenes: if scene["slug"] == "brr3-test": client.delete(f"/api/v1/scenes/{scene['id']}") create_response = client.post( "/api/v1/scenes", json={ "name": "BRR-3 test", "slug": "brr3-test", "priority": 10, "values": [{"channel": 10, "value": 255, "precedence": "htp", "source": "scene"}], }, ) assert create_response.status_code == 200 activate_response = client.post("/api/v1/scenes/brr3-test/activate") assert activate_response.status_code == 200 assert "scene:brr3-test" in app_state.engine.layers scene_id = create_response.json()["id"] delete_response = client.delete(f"/api/v1/scenes/{scene_id}") assert delete_response.status_code == 200 assert delete_response.json() == {"deleted": "brr3-test"} assert "scene:brr3-test" not in app_state.engine.layers assert app_state.scenes.active_slug is None def test_imported_fixture_persists_and_can_be_patched() -> None: universe = 101 unique_key = f"brr3-{uuid4().hex[:8]}" payload = { "manufacturer": "Eurolite", "model": "BRR-3 Test", "categories": ["Color Changer"], "source": {"manufacturer_key": "eurolite", "fixture_key": unique_key}, "modes": [ { "key": "6ch", "channels": [ {"key": "Red", "display_name": "Red", "precedence": "ltp", "resolution": 8}, {"key": "Green", "display_name": "Green", "precedence": "ltp", "resolution": 8}, {"key": "Blue", "display_name": "Blue", "precedence": "ltp", "resolution": 8}, {"key": "White", "display_name": "White", "precedence": "ltp", "resolution": 8}, {"key": "Amber", "display_name": "Amber", "precedence": "ltp", "resolution": 8}, {"key": "UV", "display_name": "UV", "precedence": "ltp", "resolution": 8}, ], } ], } with TestClient(app) as client: import_response = client.post("/api/v1/fixtures/import-file", json=payload) assert import_response.status_code == 200 fixture = import_response.json() fixture_id = fixture["id"] assert fixture["manufacturer"] == "Eurolite" assert fixture["modes"][0]["channel_count"] == 6 list_response = client.get("/api/v1/fixtures") assert list_response.status_code == 200 assert any(item["id"] == fixture_id for item in list_response.json()["items"]) validate_response = client.post( "/api/v1/patch/validate", json={ "universe": universe, "definition_id": fixture_id, "mode_key": "6ch", "start_address": 10, }, ) assert validate_response.status_code == 200 assert validate_response.json()["valid"] is True assert validate_response.json()["range"] == "10-15" create_patch_response = client.post( "/api/v1/patch", json={ "universe": universe, "name": "Eurolite venstre", "definition_id": fixture_id, "mode_key": "6ch", "start_address": 10, "enabled": True, "group_names": ["front", "wash"], "position": {"x": 20, "y": 35, "z": 8, "rotation": 15}, }, ) assert create_patch_response.status_code == 201 patch = create_patch_response.json() assert patch["start_address"] == 10 assert patch["end_address"] == 15 assert patch["group_names"] == ["front", "wash"] assert patch["position"]["x"] == 20 conflict_response = client.post( "/api/v1/patch", json={ "universe": universe, "name": "Eurolite højre", "definition_id": fixture_id, "mode_key": "6ch", "start_address": 12, "enabled": True, "group_names": [], "position": {"x": 70, "y": 35, "z": 8, "rotation": 345}, }, ) assert conflict_response.status_code == 409 delete_patch_response = client.delete(f"/api/v1/patch/{patch['id']}") assert delete_patch_response.status_code == 200 delete_fixture_response = client.delete(f"/api/v1/fixtures/{fixture_id}") assert delete_fixture_response.status_code == 204 def test_import_ofl_style_fixture_file_supports_mode_channels() -> None: payload = { "$schema": "https://raw.githubusercontent.com/OpenLightingProject/open-fixture-library/master/schemas/fixture.json", "manufacturer": "Eurolite", "name": "LED Bar-3 HCL Bar", "categories": ["Color Changer"], "source": { "manufacturer_key": "eurolite", "fixture_key": f"led-bar-3-hcl-bar-{uuid4().hex[:8]}", }, "availableChannels": { "Dimmer": {"capability": {"type": "Intensity"}}, "Strobe": { "capabilities": [ {"dmxRange": [0, 9], "type": "NoFunction"}, {"dmxRange": [10, 255], "type": "ShutterStrobe"}, ] }, "Red": {"capability": {"type": "ColorIntensity", "color": "Red"}}, "Green": {"capability": {"type": "ColorIntensity", "color": "Green"}}, "Blue": {"capability": {"type": "ColorIntensity", "color": "Blue"}}, "White": {"capability": {"type": "ColorIntensity", "color": "White"}}, "Amber": {"capability": {"type": "ColorIntensity", "color": "Amber"}}, "UV": {"capability": {"type": "ColorIntensity", "color": "UV"}}, }, "modes": [ {"name": "6ch", "channels": ["Red", "Green", "Blue", "White", "Amber", "UV"]}, {"name": "10ch", "channels": ["Dimmer", "Strobe", "Red", "Green", "Blue", "White", "Amber", "UV"]}, ], } with TestClient(app) as client: response = client.post("/api/v1/fixtures/import-file", json=payload) assert response.status_code == 200 fixture = response.json() assert fixture["manufacturer"] == "Eurolite" assert fixture["model"] == "LED Bar-3 HCL Bar" assert fixture["source"]["manufacturer_key"] == "eurolite" assert fixture["modes"][0]["channels"][0]["key"] == "Red" assert fixture["modes"][1]["channels"][0]["precedence"] == "htp" assert fixture["modes"][1]["channels"][1]["capabilities"][1]["type"] == "ShutterStrobe" delete_fixture_response = client.delete(f"/api/v1/fixtures/{fixture['id']}") assert delete_fixture_response.status_code == 204 def test_live_mixer_returns_and_updates_patched_fixture() -> None: universe = 102 unique_key = f"live-mixer-{uuid4().hex[:8]}" payload = { "manufacturer": "Eurolite", "model": "LED Bar-3 HCL Bar", "categories": ["Color Changer"], "source": {"manufacturer_key": "eurolite", "fixture_key": unique_key}, "modes": [ { "key": "10-channel", "channels": [ {"key": "Dimmer", "display_name": "Dimmer", "precedence": "htp", "resolution": 8}, {"key": "Strobe", "display_name": "Strobe", "precedence": "ltp", "resolution": 8}, {"key": "Red", "display_name": "Red", "precedence": "ltp", "resolution": 8}, ], } ], } with TestClient(app) as client: fixture = client.post("/api/v1/fixtures/import-file", json=payload).json() patch = client.post( "/api/v1/patch", json={ "universe": universe, "name": "Eurolite live", "definition_id": fixture["id"], "mode_key": "10-channel", "start_address": 10, "enabled": True, "group_names": [], "position": {"x": 40, "y": 50, "z": 5, "rotation": 0}, }, ).json() mixer_response = client.get("/api/v1/live/mixer") assert mixer_response.status_code == 200 assert any(item["patch_id"] == patch["id"] for item in mixer_response.json()["items"]) update_response = client.put( f"/api/v1/live/mixer/{patch['id']}", json={"values": {"1": 200, "3": 128}}, ) assert update_response.status_code == 200 mixer_item = next(item for item in update_response.json()["items"] if item["patch_id"] == patch["id"]) dimmer = next(channel for channel in mixer_item["channels"] if channel["index"] == 1) red = next(channel for channel in mixer_item["channels"] if channel["index"] == 3) assert dimmer["value"] == 200 assert red["absolute_channel"] == 12 clear_response = client.delete(f"/api/v1/live/mixer/{patch['id']}") assert clear_response.status_code == 200 assert client.delete(f"/api/v1/patch/{patch['id']}").status_code == 200 assert client.delete(f"/api/v1/fixtures/{fixture['id']}").status_code == 204 def test_live_mixer_returns_and_updates_home_assistant_mapping() -> None: with TestClient(app) as client: create_response = client.post( "/api/v1/integrations/home-assistant/mappings", json={ "name": "Loft bunker", "universe": 10, "start_address": 1, "fixture_type": "switch", "entity_id": "light.loft_bunker", "rate_limit_hz": 5, "deadband": 2, "fade_ms": 0, "invert_channel": False, "min_value": 0, "max_value": 255, "enabled": True, "master_dimmer": False, }, ) assert create_response.status_code == 200 mapping = create_response.json() mixer_response = client.get("/api/v1/live/mixer") assert mixer_response.status_code == 200 mixer_item = next( item for item in mixer_response.json()["items"] if item["source_type"] == "home_assistant" and item["source_id"] == mapping["id"] ) assert mixer_item["entity_id"] == "light.loft_bunker" assert mixer_item["universe"] == 10 assert mixer_item["channel_count"] == 1 update_response = client.put( f"/api/v1/live/mixer/home-assistant/{mapping['id']}", json={"values": {"1": 255}}, ) assert update_response.status_code == 200 updated_item = next( item for item in update_response.json()["items"] if item["source_type"] == "home_assistant" and item["source_id"] == mapping["id"] ) assert updated_item["channels"][0]["value"] == 255 assert updated_item["channels"][0]["absolute_channel"] == 1 clear_response = client.delete(f"/api/v1/live/mixer/home-assistant/{mapping['id']}") assert clear_response.status_code == 200 delete_response = client.delete(f"/api/v1/integrations/home-assistant/mappings/{mapping['id']}") assert delete_response.status_code == 200 def test_scene_group_target_resolves_fixture_channels() -> None: universe = 103 unique_key = f"scene-group-{uuid4().hex[:8]}" payload = { "manufacturer": "Eurolite", "model": "LED Bar-3 HCL Bar", "categories": ["Color Changer"], "source": {"manufacturer_key": "eurolite", "fixture_key": unique_key}, "modes": [ { "key": "10-channel", "channels": [ {"key": "Dimmer", "display_name": "Dimmer", "precedence": "htp", "resolution": 8}, {"key": "Strobe", "display_name": "Strobe", "precedence": "ltp", "resolution": 8}, {"key": "Blue", "display_name": "Blue", "precedence": "ltp", "resolution": 8}, ], } ], } with TestClient(app) as client: fixture = client.post("/api/v1/fixtures/import-file", json=payload).json() patch_left = client.post( "/api/v1/patch", json={ "universe": universe, "name": "Bar venstre", "definition_id": fixture["id"], "mode_key": "10-channel", "start_address": 1, "enabled": True, "group_names": ["synk-a", "front"], "position": {"x": 20, "y": 30, "z": 5, "rotation": 10}, }, ).json() patch_right = client.post( "/api/v1/patch", json={ "universe": universe, "name": "Bar højre", "definition_id": fixture["id"], "mode_key": "10-channel", "start_address": 11, "enabled": True, "group_names": ["synk-a", "front"], "position": {"x": 80, "y": 30, "z": 5, "rotation": 350}, }, ).json() scene_response = client.post( "/api/v1/scenes", json={ "name": "Front blue", "slug": "front-blue", "priority": 20, "values": [], "targets": [ { "target_type": "group", "group_name": "synk-a", "values": [ {"attribute": "Dimmer", "value": 200}, {"attribute": "Blue", "value": 128}, ], } ], }, ) assert scene_response.status_code == 200 activate_response = client.post("/api/v1/scenes/front-blue/activate") assert activate_response.status_code == 200 layer = app_state.engine.layers["scene:front-blue"] assert layer.values_by_universe[universe][1] == 200 assert layer.values_by_universe[universe][3] == 128 assert layer.values_by_universe[universe][11] == 200 assert layer.values_by_universe[universe][13] == 128 scene_id = scene_response.json()["id"] assert client.delete(f"/api/v1/scenes/{scene_id}").status_code == 200 assert client.delete(f"/api/v1/patch/{patch_left['id']}").status_code == 200 assert client.delete(f"/api/v1/patch/{patch_right['id']}").status_code == 200 assert client.delete(f"/api/v1/fixtures/{fixture['id']}").status_code == 204