hms-mediaengine/tests/integration/test_ipc_connection.py

"""Integrationstests IPC-Verbindung (PLAN.md §6.2, §29.2).

Echter TCP-Loopback (kein Mock): Handshake, Heartbeat, Snapshot/Delta,
Idempotenz, Version-Mismatch, Re-Sync nach Reconnect.
"""

from __future__ import annotations

import asyncio

from hms_protocol import (
    Envelope,
    HandshakeInfo,
    IpcClient,
    IpcServer,
    MessageType,
)


def _run(coro):
    return asyncio.run(coro)


async def _connected_pair(
    server_caps: dict | None = None,
    client_caps: dict | None = None,
):
    """Startet Server+Client und führt den Handshake durch."""
    server = IpcServer()
    if server_caps is not None:
        server.capabilities = server_caps
    port = await server.start()
    client = IpcClient(port)
    if client_caps is not None:
        client.capabilities = client_caps
    info = await client.connect()
    return server, client, info


# ---------- Bindung ausschließlich Loopback (§6.2) ----------


def test_server_rejects_non_loopback_host() -> None:
    try:
        IpcServer(host="0.0.0.0")
        raise AssertionError("0.0.0.0 muss abgelehnt werden")
    except ValueError:
        pass


def test_client_rejects_non_loopback_host() -> None:
    try:
        IpcClient(port=1234, host="192.168.1.5")
        raise AssertionError("externe IP muss abgelehnt werden")
    except ValueError:
        pass


# ---------- Handshake (§6.2) ----------


def test_handshake_exchanges_capabilities() -> None:
    async def impl() -> None:
        server, client, info = await _connected_pair(
            server_caps={"backend": "d3d11", "max_layers": 8},
            client_caps={"role": "control_core"},
        )
        try:
            assert isinstance(info, HandshakeInfo)
            assert info.peer_name == "hms-renderer"
            assert info.peer_capabilities == {"backend": "d3d11", "max_layers": 8}
            assert info.protocol_version == 1
        finally:
            await client.disconnect()
            await server.stop()

    _run(impl())


def test_client_rejects_version_mismatch() -> None:
    async def impl() -> None:
        server = IpcServer()
        port = await server.start()
        client = IpcClient(port)
        try:
            # manipulierter Handshake mit falscher Protokollversion
            from hms_protocol import encode_frame

            reader, writer = await asyncio.open_connection("127.0.0.1", port)
            writer.write(
                encode_frame(
                    {
                        "protocol_version": 99,
                        "message_id": "x",
                        "type": "command",
                        "revision": 0,
                        "monotonic_timestamp_ns": 0,
                        "payload": {"action": "hello", "name": "evil", "capabilities": {}},
                    }
                )
            )
            await writer.drain()
            # Server antwortet mit ERROR VERSION_MISMATCH
            from hms_protocol import read_frame_async

            reply = await read_frame_async(reader)
            assert reply["type"] == "error"
            assert reply["payload"]["code"] == "VERSION_MISMATCH"
            writer.close()
        finally:
            await client.disconnect()
            await server.stop()

    _run(impl())


# ---------- Nachrichtenübertragung ----------


def test_snapshot_and_delta_delivery() -> None:
    async def impl() -> None:
        server, client, _ = await _connected_pair()
        try:
            # Server sendet Snapshot
            snap = Envelope(
                type=MessageType.SNAPSHOT,
                revision=5,
                payload={"values": {"master/intensity": 1.0}},
            )
            await server.send(snap)
            received = await asyncio.wait_for(client.receive(), timeout=2.0)
            assert received is not None
            assert received.type is MessageType.SNAPSHOT
            assert received.revision == 5
            assert received.payload["values"]["master/intensity"] == 1.0

            # danach Delta mit höherer Revision
            delta = Envelope(
                type=MessageType.EVENT,
                revision=6,
                payload={"changes": {"master/intensity": 0.5}},
            )
            await server.send(delta)
            received2 = await asyncio.wait_for(client.receive(), timeout=2.0)
            assert received2 is not None
            assert received2.revision == 6
            assert received2.payload["changes"]["master/intensity"] == 0.5
        finally:
            await client.disconnect()
            await server.stop()

    _run(impl())


def test_bidirectional_commands_and_acks() -> None:
    async def impl() -> None:
        server, client, _ = await _connected_pair()
        try:
            # Client → Server: Command
            cmd = Envelope(
                type=MessageType.COMMAND,
                revision=10,
                payload={"action": "parameter.set", "path": "master/intensity", "value": 0.7},
            )
            await client.send(cmd)
            received = await asyncio.wait_for(server.receive(), timeout=2.0)
            assert received is not None
            assert received.type is MessageType.COMMAND
            assert received.payload["action"] == "parameter.set"

            # Server → Client: Ack
            ack = Envelope(
                type=MessageType.ACK,
                revision=11,
                payload={"ack_for": received.message_id, "status": "ok"},
            )
            await server.send(ack)
            ack_received = await asyncio.wait_for(client.receive(), timeout=2.0)
            assert ack_received is not None
            assert ack_received.type is MessageType.ACK
            assert ack_received.payload["ack_for"] == received.message_id
        finally:
            await client.disconnect()
            await server.stop()

    _run(impl())


# ---------- Heartbeat (§6.2: mindestens alle 500 ms) ----------


def test_heartbeat_flows_bidirectionally() -> None:
    async def impl() -> None:
        server, client, _ = await _connected_pair()
        try:
            await asyncio.sleep(0.7)  # > Heartbeat-Intervall
            # Client empfängt Server-Heartbeat
            got_server_hb = False
            for _ in range(4):
                msg = await asyncio.wait_for(client.receive(), timeout=1.5)
                if msg is not None and msg.type is MessageType.HEARTBEAT:
                    got_server_hb = True
                    break
            assert got_server_hb, "Client musste einen Heartbeat empfangen"
            assert client.peer_alive, "Server-Heartbeat muss peer_alive setzen"

            # Server empfängt Client-Heartbeat
            await asyncio.sleep(0.1)
            got_client_hb = False
            for _ in range(6):
                msg = await asyncio.wait_for(server.receive(), timeout=1.5)
                if msg is not None and msg.type is MessageType.HEARTBEAT:
                    got_client_hb = True
                    break
            assert got_client_hb, "Server musste einen Heartbeat empfangen"
            assert server.peer_alive
        finally:
            await client.disconnect()
            await server.stop()

    _run(impl())


# ---------- Re-Sync nach Reconnect (§6.2, §29.2) ----------


def test_reconnect_requires_new_snapshot() -> None:
    async def impl() -> None:
        server = IpcServer()
        port = await server.start()
        try:
            # erste Verbindung: Snapshot Revision 3
            client1 = IpcClient(port)
            await client1.connect()
            await server.send(
                Envelope(type=MessageType.SNAPSHOT, revision=3, payload={"values": {}})
            )
            snap1 = await asyncio.wait_for(client1.receive(), timeout=2.0)
            assert snap1 is not None and snap1.revision == 3
            await client1.disconnect()

            # zweite Verbindung: neuer Snapshot (Re-Sync) mit Revision 4
            client2 = IpcClient(port)
            info2 = await client2.connect()  # noqa: F841 – Handshake reicht
            await server.send(
                Envelope(type=MessageType.SNAPSHOT, revision=4, payload={"values": {"a": 1}})
            )
            snap2 = await asyncio.wait_for(client2.receive(), timeout=2.0)
            assert snap2 is not None
            assert snap2.revision == 4  # Deltas erst nach erfolgreichem Re-Sync
            await client2.disconnect()
        finally:
            await server.stop()

    _run(impl())


def test_receive_returns_none_on_disconnect() -> None:
    async def impl() -> None:
        server, client, _ = await _connected_pair()
        await client.disconnect()
        msg = await asyncio.wait_for(server.receive(), timeout=2.0)
        assert msg is None  # Verbindung weg → sauberes None statt Exception
        await server.stop()

    _run(impl())