hms-mediaengine/_entwicklung/tests/integration/test_control_server.py

"""Integrationstests Control Core (PLAN.md §6.1B, §23, §36 Nr. 9).

FastAPI-REST + WebSocket mit derselben Parameter-Engine, die auch
Art-Net bedient (§11: eine autoritative Instanz). Node-Identität und
Cluster-Registry sind verkabelt (§6.3).
"""

from __future__ import annotations

import uuid

import pytest
from fastapi.testclient import TestClient
from hms_control_server import create_app
from hms_domain import NodeIdentity, NodeRole


@pytest.fixture()
def client() -> TestClient:
    return TestClient(create_app())


@pytest.fixture()
def named_client() -> TestClient:
    """App mit injizierter produktionsähnlicher Identität."""
    identity = NodeIdentity.ephemeral(
        "Show Server A", frozenset({NodeRole.RENDER_NODE, NodeRole.COORDINATOR})
    )
    return TestClient(create_app(identity=identity))


def _payload(path: str, value: float) -> dict:
    return {"parameter_path": path, "value": value}


@pytest.fixture()
def path() -> str:
    return f"composition/{uuid.uuid4()}/layer/{uuid.uuid4()}/opacity"


# ---------- Health/Capabilities/Identity ----------


def test_health(client: TestClient) -> None:
    r = client.get("/api/v1/system/health")
    assert r.status_code == 200
    assert r.json()["status"] == "ok"
    assert r.json()["phase"] == 5
    assert r.json()["node_id"]  # Identität in Health sichtbar


def test_identity_reports_roles_without_secrets(named_client: TestClient) -> None:
    r = named_client.get("/api/v1/system/identity")
    assert r.status_code == 200
    body = r.json()
    assert body["display_name"] == "Show Server A"
    assert body["roles"] == ["COORDINATOR", "RENDER_NODE"]
    assert body["renders_locally"] is True
    assert body["is_coordinator"] is True
    # keine vertraulichen Felder (§27.1)
    blob = str(body).lower()
    for forbidden in ("token", "secret", "password"):
        assert forbidden not in blob


def test_two_apps_get_distinct_dev_identities() -> None:
    """Ohne injizierte Identität erhält jede App-Instanz eine eigene ID
    (Dev-Modus; Showbetrieb nutzt persistente Identität über den Launcher)."""
    a = TestClient(create_app()).get("/api/v1/system/identity").json()
    b = TestClient(create_app()).get("/api/v1/system/identity").json()
    assert a["node_id"] != b["node_id"]


def test_capabilities_reported_without_fake_tier(client: TestClient) -> None:
    r = client.get("/api/v1/system/capabilities")
    assert r.status_code == 200
    assert r.json()["tier"] is None  # ungeprüft (CPU-only), kein Fake


def test_diagnostics_reports_not_connected(client: TestClient) -> None:
    r = client.get("/api/v1/diagnostics")
    assert r.status_code == 200
    assert r.json()["renderer"] == "not_connected"
    assert r.json()["node_id"]


# ---------- Cluster-Registry (§6.3) ----------


def test_self_node_listed_in_cluster_nodes(named_client: TestClient) -> None:
    r = named_client.get("/api/v1/cluster/nodes")
    assert r.status_code == 200
    body = r.json()
    self_entry = next(n for n in body["nodes"] if n["node_id"] == body["self"])
    assert self_entry["display_name"] == "Show Server A"
    assert set(self_entry["roles"]) == {"RENDER_NODE", "COORDINATOR"}
    assert self_entry["health"] in ("online", "offline")  # ohne Heartbeat offline


# ---------- Commands (§23.2) ----------


def test_parameter_set_command_ack(client: TestClient, path: str) -> None:
    r = client.post("/api/v1/commands", json={"payload": _payload(path, 0.75)})
    assert r.status_code == 200
    body = r.json()
    assert body["status"] == "ack"
    assert body["effective"] == pytest.approx(0.75)


def test_duplicate_command_id_is_idempotent(client: TestClient, path: str) -> None:
    cmd = {"command_id": str(uuid.uuid4()), "payload": _payload(path, 0.5)}
    first = client.post("/api/v1/commands", json=cmd).json()
    second = client.post("/api/v1/commands", json=cmd).json()
    assert first["status"] == "ack"
    assert second.get("duplicate") is True
    assert second["result"]["revision"] == first["revision"]


def test_revision_conflict_returns_409(client: TestClient, path: str) -> None:
    r = client.post(
        "/api/v1/commands",
        json={"payload": _payload(path, 0.5), "expected_revision": 999},
    )
    assert r.status_code == 409
    assert r.json()["detail"]["error"] == "REVISION_CONFLICT"


def test_invalid_parameter_path_returns_400(client: TestClient) -> None:
    r = client.post(
        "/api/v1/commands",
        json={"payload": _payload("../evil/path", 1.0)},
    )
    assert r.status_code == 400


def test_unknown_command_type_returns_400(client: TestClient) -> None:
    r = client.post("/api/v1/commands", json={"type": "renderer.draw", "payload": {}})
    assert r.status_code == 400


def test_parameters_endpoint_reflects_state(client: TestClient, path: str) -> None:
    client.post("/api/v1/commands", json={"payload": _payload(path, 0.42)})
    r = client.get("/api/v1/parameters")
    assert r.status_code == 200
    assert r.json()["values"][path] == pytest.approx(0.42)
    assert r.json()["revision"] >= 1


# ---------- WebSocket (§23.3) ----------


def test_websocket_snapshot_on_connect(client: TestClient, path: str) -> None:
    client.post("/api/v1/commands", json={"payload": _payload(path, 0.33)})
    with client.websocket_connect("/ws") as ws:
        snap = ws.receive_json()
        assert snap["type"] == "snapshot"
        assert snap["values"][path] == pytest.approx(0.33)


def test_websocket_receives_parameter_updates(client: TestClient, path: str) -> None:
    with client.websocket_connect("/ws") as ws:
        ws.receive_json()  # Initial-Snapshot
        client.post("/api/v1/commands", json={"payload": _payload(path, 0.9)})
        event = ws.receive_json()
        assert event["type"] == "parameter.update"
        assert event["parameter_path"] == path
        assert event["value"] == pytest.approx(0.9)