path: root/packages/meshbay-node/tests/test_webrtc_transport.py

"""
Integration test: WebRTC DataChannel transport for browser clients.

Phase 9 milestone 9.1 — spike: validate aiortc WebRTC DataChannel works
for MNP protocol exchange (handshake, index_sync, file_request, file_chunk).

Uses local loopback (no STUN/ICE needed for localhost).
"""

import asyncio
import base64
import os
import struct
import time

import jwt
import msgpack
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from aiortc import RTCPeerConnection, RTCSessionDescription

from meshbay_common import MNP_VERSION
from meshbay_common.crypto import (
    generate_gek,
    pk_to_b64,
)
from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes
from meshbay_common.protocol import MNP
from meshbay_node.indexer import DirectoryIndexer
from meshbay_node.transport.webrtc_server import WebRTCTransport


@pytest.fixture
def sk_node():
    return Ed25519PrivateKey.generate()


@pytest.fixture
def sk_hub():
    return Ed25519PrivateKey.generate()


@pytest.fixture
def gek():
    return generate_gek()


@pytest.fixture
def shared_dir(tmp_path):
    d = tmp_path / "shared"
    d.mkdir()
    (d / "test.bin").write_bytes(os.urandom(2048))
    (d / "hello.txt").write_bytes(b"hello webrtc " * 50)
    return d


def _hub_pk_pem(sk_hub):
    return sk_hub.public_key().public_bytes(
        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)


def _make_jwt(sk_hub, groups=None):
    sk_pem = sk_hub.private_bytes(
        serialization.Encoding.PEM,
        serialization.PrivateFormat.PKCS8,
        serialization.NoEncryption(),
    )
    now = int(time.time())
    return jwt.encode({
        "iss": "test-hub", "sub": "user-001",
        "pk_user": "test", "hub_id": "test-hub",
        "jti": "test-jti-webrtc", "iat": now, "exp": now + 3600,
        "groups": groups or [],
    }, sk_pem, algorithm="EdDSA")


def _pack(obj: dict) -> bytes:
    data = msgpack.packb(obj, use_bin_type=True)
    return struct.pack(">I", len(data)) + data


def _unpack(raw: bytes) -> dict:
    length = struct.unpack(">I", raw[:4])[0]
    return msgpack.unpackb(raw[4:4 + length], raw=False)


@pytest.mark.asyncio
async def test_webrtc_datachannel_handshake(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: browser sends MNP handshake, node responds with handshake_ack."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()

    channel = browser_pc.createDataChannel("mnp")

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        received.put_nowait(_unpack(message))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)

    answer_sdp, ice_candidates = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-001")

    answer = RTCSessionDescription(sdp=answer_sdp, type="answer")
    await browser_pc.setRemoteDescription(answer)

    await asyncio.sleep(0.5)

    token = _make_jwt(sk_hub)
    channel.send(_pack({
        "type": MNP.HANDSHAKE,
        "v": MNP_VERSION,
        "token": token,
    }))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == MNP.HANDSHAKE_ACK
    assert msg["v"] == MNP_VERSION
    assert "node_pk" in msg

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: full file transfer — handshake, index, fetch chunk, decrypt."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()

    channel = browser_pc.createDataChannel("mnp")

    @channel.on("open")
    def on_open():
        token = _make_jwt(sk_hub)
        channel.send(_pack({
            "type": MNP.HANDSHAKE,
            "v": MNP_VERSION,
            "token": token,
        }))

    buf = bytearray()

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        buf.extend(message)
        while len(buf) >= 4:
            length = struct.unpack(">I", buf[:4])[0]
            if len(buf) < 4 + length:
                break
            msg_bytes = bytes(buf[4:4 + length])
            del buf[:4 + length]
            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)

    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-002")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    # 1) Handshake ack
    ack = await asyncio.wait_for(received.get(), timeout=5.0)
    assert ack["type"] == MNP.HANDSHAKE_ACK

    # 2) Request index
    channel.send(_pack({"type": MNP.INDEX_SYNC, "v": MNP_VERSION}))
    idx_msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert idx_msg["type"] == MNP.INDEX_SYNC
    assert "entries" in idx_msg
    assert len(idx_msg["entries"]) > 0

    # 3) Request file chunk
    entry = next(e for e in indexer.index.entries if e.name == "test.bin")
    channel.send(_pack({
        "type": MNP.FILE_REQUEST,
        "v": MNP_VERSION,
        "file_id": entry.id,
        "chunk_index": 0,
    }))

    chunk_msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert chunk_msg["type"] == MNP.FILE_CHUNK

    # 4) Verify and decrypt (binary fields — no base64, minimal envelope)
    ct = chunk_msg["ct"]
    nonce = chunk_msg["nonce"]
    file_hash = bytes.fromhex(entry.id)

    ckey = chunk_key_aes(gek, file_hash, 0)
    plaintext = decrypt_chunk_aes(ckey, nonce, ct)

    original = (shared_dir / "test.bin").read_bytes()
    assert plaintext == original

    # 5) Request GEK over DataChannel
    channel.send(_pack({"type": MNP.GEK_REQUEST, "v": MNP_VERSION}))
    gek_msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert gek_msg["type"] == MNP.GEK_RESPONSE
    received_gek = base64.b64decode(gek_msg["gek_b64"])
    assert received_gek == gek

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: invalid JWT is rejected with error."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()

    channel = browser_pc.createDataChannel("mnp")

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        received.put_nowait(_unpack(message))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)

    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-003")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    await asyncio.sleep(0.5)

    channel.send(_pack({
        "type": MNP.HANDSHAKE,
        "v": MNP_VERSION,
        "token": "invalid.jwt.token",
    }))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    assert "JWT" in msg["detail"] or "Invalid" in msg["detail"]

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_request_before_handshake_rejected(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: request without handshake is rejected."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()

    channel = browser_pc.createDataChannel("mnp")

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        received.put_nowait(_unpack(message))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)

    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-004")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    await asyncio.sleep(0.5)

    channel.send(_pack({"type": MNP.INDEX_SYNC, "v": MNP_VERSION}))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    assert "Handshake required" in msg["detail"]

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tmp_path):
    """WebRTC DataChannel: send chat message, then retrieve history."""
    from meshbay_node.chat.store import ChatStore

    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    chat_store = ChatStore(db_path=tmp_path / "chat_test.db")
    await chat_store.open()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["chat_store"] = chat_store

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()
    buf = bytearray()

    channel = browser_pc.createDataChannel("mnp")

    @channel.on("open")
    def on_open():
        token = _make_jwt(sk_hub)
        channel.send(_pack({
            "type": MNP.HANDSHAKE,
            "v": MNP_VERSION,
            "token": token,
        }))

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        buf.extend(message)
        while len(buf) >= 4:
            length = struct.unpack(">I", buf[:4])[0]
            if len(buf) < 4 + length:
                break
            msg_bytes = bytes(buf[4:4 + length])
            del buf[:4 + length]
            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)

    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-chat")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    ack = await asyncio.wait_for(received.get(), timeout=5.0)
    assert ack["type"] == MNP.HANDSHAKE_ACK

    channel.send(_pack({
        "type": MNP.CHAT_MESSAGE,
        "v": MNP_VERSION,
        "payload": "hello from browser",
    }))
    chat_ack = await asyncio.wait_for(received.get(), timeout=5.0)
    assert chat_ack["type"] == "ack"

    await asyncio.sleep(0.2)

    channel.send(_pack({
        "type": MNP.CHAT_HISTORY,
        "v": MNP_VERSION,
        "since": 0,
        "limit": 50,
    }))
    hist = await asyncio.wait_for(received.get(), timeout=5.0)
    assert hist["type"] == MNP.CHAT_HISTORY_RESPONSE
    assert len(hist["messages"]) == 1
    assert hist["messages"][0]["payload"] == "hello from browser"
    assert hist["messages"][0]["sender_id"] == "user-001"

    await chat_store.close()
    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_chat_history_no_store(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: chat history without chat_store returns empty list."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()
    buf = bytearray()

    channel = browser_pc.createDataChannel("mnp")

    @channel.on("open")
    def on_open():
        channel.send(_pack({
            "type": MNP.HANDSHAKE, "v": MNP_VERSION,
            "token": _make_jwt(sk_hub),
        }))

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        buf.extend(message)
        while len(buf) >= 4:
            length = struct.unpack(">I", buf[:4])[0]
            if len(buf) < 4 + length:
                break
            msg_bytes = bytes(buf[4:4 + length])
            del buf[:4 + length]
            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-no-store")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    ack = await asyncio.wait_for(received.get(), timeout=5.0)
    assert ack["type"] == MNP.HANDSHAKE_ACK

    channel.send(_pack({
        "type": MNP.CHAT_HISTORY, "v": MNP_VERSION, "since": 0, "limit": 50,
    }))
    hist = await asyncio.wait_for(received.get(), timeout=5.0)
    assert hist["type"] == MNP.CHAT_HISTORY_RESPONSE
    assert hist["messages"] == []

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path):
    """WebRTC DataChannel: chat message from peer A is broadcast to peer B."""
    from meshbay_node.chat.store import ChatStore

    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    chat_store = ChatStore(db_path=tmp_path / "chat_bc.db")
    await chat_store.open()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )
    transport._ctx["chat_store"] = chat_store

    async def _connect_peer(peer_id, jwt_sub, groups=None):
        pc = RTCPeerConnection()
        q = asyncio.Queue()
        b = bytearray()
        ch = pc.createDataChannel("mnp")

        sk_h_pem = sk_hub.private_bytes(
            serialization.Encoding.PEM,
            serialization.PrivateFormat.PKCS8,
            serialization.NoEncryption(),
        )
        now = int(time.time())
        token = jwt.encode({
            "iss": "test-hub", "sub": jwt_sub,
            "pk_user": "test", "hub_id": "test-hub",
            "jti": f"jti-{peer_id}", "iat": now, "exp": now + 3600,
            "groups": groups or [],
        }, sk_h_pem, algorithm="EdDSA")

        @ch.on("open")
        def on_open():
            ch.send(_pack({
                "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
            }))

        @ch.on("message")
        def on_msg(message):
            if isinstance(message, str):
                message = message.encode()
            b.extend(message)
            while len(b) >= 4:
                length = struct.unpack(">I", b[:4])[0]
                if len(b) < 4 + length:
                    break
                msg_bytes = bytes(b[4:4 + length])
                del b[:4 + length]
                q.put_nowait(msgpack.unpackb(msg_bytes, raw=False))

        offer = await pc.createOffer()
        await pc.setLocalDescription(offer)
        answer_sdp, _ = await transport.handle_offer(
            pc.localDescription.sdp, peer_id)
        await pc.setRemoteDescription(
            RTCSessionDescription(sdp=answer_sdp, type="answer"))

        ack = await asyncio.wait_for(q.get(), timeout=5.0)
        assert ack["type"] == MNP.HANDSHAKE_ACK
        return pc, ch, q

    pc_a, ch_a, q_a = await _connect_peer("peer-A", "user-A")
    pc_b, ch_b, q_b = await _connect_peer("peer-B", "user-B")

    ch_a.send(_pack({
        "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "payload": "hi from A",
    }))

    ack_a = await asyncio.wait_for(q_a.get(), timeout=5.0)
    assert ack_a["type"] == "ack"

    broadcast = await asyncio.wait_for(q_b.get(), timeout=5.0)
    assert broadcast["type"] == MNP.CHAT_MESSAGE
    assert broadcast["sender_id"] == "user-A"
    assert broadcast["payload"] == "hi from A"

    await chat_store.close()
    await pc_a.close()
    await pc_b.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_group_membership_enforced(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: JWT without matching group claim is rejected."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()

    channel = browser_pc.createDataChannel("mnp")

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        received.put_nowait(_unpack(message))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-group-test")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    await asyncio.sleep(0.5)

    token = _make_jwt(sk_hub, groups=["other-group"])
    channel.send(_pack({
        "type": MNP.HANDSHAKE, "v": MNP_VERSION,
        "token": token, "group_id": "my-group",
    }))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    assert "Not a member" in msg["detail"]

    await browser_pc.close()
    await transport.close_all()


@pytest.mark.asyncio
async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: peer removed from _peers dict on session close."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()
    buf = bytearray()

    channel = browser_pc.createDataChannel("mnp")

    @channel.on("open")
    def on_open():
        channel.send(_pack({
            "type": MNP.HANDSHAKE, "v": MNP_VERSION,
            "token": _make_jwt(sk_hub),
        }))

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        buf.extend(message)
        while len(buf) >= 4:
            length = struct.unpack(">I", buf[:4])[0]
            if len(buf) < 4 + length:
                break
            msg_bytes = bytes(buf[4:4 + length])
            del buf[:4 + length]
            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-cleanup")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    ack = await asyncio.wait_for(received.get(), timeout=5.0)
    assert ack["type"] == MNP.HANDSHAKE_ACK

    assert "user-001" in transport._ctx["_peers"]
    assert transport.active_peers == 1

    await transport.close_peer("peer-cleanup")

    assert "user-001" not in transport._ctx["_peers"]
    assert transport.active_peers == 0

    await browser_pc.close()


@pytest.mark.asyncio
async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_dir):
    """WebRTC DataChannel: stream_segment for non-existent file returns error."""
    hub_pk_pem = _hub_pk_pem(sk_hub)
    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
    await indexer.initial_scan()

    transport = WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
        shared_root=shared_dir, index=indexer.index,
        stun_servers=[],
    )

    browser_pc = RTCPeerConnection()
    received = asyncio.Queue()
    buf = bytearray()

    channel = browser_pc.createDataChannel("mnp")

    @channel.on("open")
    def on_open():
        channel.send(_pack({
            "type": MNP.HANDSHAKE, "v": MNP_VERSION,
            "token": _make_jwt(sk_hub),
        }))

    @channel.on("message")
    def on_msg(message):
        if isinstance(message, str):
            message = message.encode()
        buf.extend(message)
        while len(buf) >= 4:
            length = struct.unpack(">I", buf[:4])[0]
            if len(buf) < 4 + length:
                break
            msg_bytes = bytes(buf[4:4 + length])
            del buf[:4 + length]
            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))

    offer = await browser_pc.createOffer()
    await browser_pc.setLocalDescription(offer)
    answer_sdp, _ = await transport.handle_offer(
        browser_pc.localDescription.sdp, "peer-stream")
    await browser_pc.setRemoteDescription(
        RTCSessionDescription(sdp=answer_sdp, type="answer"))

    ack = await asyncio.wait_for(received.get(), timeout=5.0)
    assert ack["type"] == MNP.HANDSHAKE_ACK

    channel.send(_pack({
        "type": MNP.STREAM_SEGMENT, "v": MNP_VERSION,
        "file_id": "nonexistent-file-id",
        "segment_index": 0, "segment_duration": 4,
    }))

    msg = await asyncio.wait_for(received.get(), timeout=5.0)
    assert msg["type"] == "error"
    assert "not found" in msg["detail"].lower()

    await browser_pc.close()
    await transport.close_all()