from __future__ import annotations

import asyncio
import stat
import threading
from pathlib import Path

import numpy as np

from k1link.compute.live_perception import (
    LivePerceptionIngress,
    encode_live_perception_result,
)
from k1link.device_plugins.xgrids_k1.live_perception_shadow import (
    build_live_perception_shadow_router,
    ensure_live_shadow_token,
)


def test_shadow_token_is_stable_and_private(tmp_path: Path) -> None:
    path, token = ensure_live_shadow_token(tmp_path)
    repeated_path, repeated_token = ensure_live_shadow_token(tmp_path)

    assert repeated_path == path
    assert repeated_token == token
    assert stat.S_IMODE(path.stat().st_mode) == 0o600
    assert stat.S_IMODE(path.parent.stat().st_mode) == 0o700


def test_shadow_router_exposes_only_the_exclusive_binary_stream() -> None:
    ingress = LivePerceptionIngress()
    router = build_live_perception_shadow_router(
        ingress,
        "test-plugin",
        bearer_token="x" * 43,
    )
    assert len(router.routes) == 1
    assert router.routes[0].path == (
        "/api/v1/device-plugins/test-plugin/live-perception-shadow"
    )


def test_shadow_router_accepts_only_validated_diagnostic_results_back() -> None:
    ingress = LivePerceptionIngress()
    received: list[bytes] = []
    published = threading.Event()

    def receive_result(payload: bytes) -> bool:
        received.append(payload)
        published.set()
        return True

    router = build_live_perception_shadow_router(
        ingress,
        "test-plugin",
        bearer_token="x" * 43,
        result_receiver=receive_result,
    )
    ingress.begin_session("session-1")
    encoded = encode_live_perception_result(
        frame_index=0,
        source_frame_index=0,
        session_seconds=0.0,
        captured_at_epoch_ns=1,
        image_jpeg=bytes.fromhex("ffd878ffd9"),
        segmentation_mask=np.zeros((600, 800), dtype=np.uint8),
        objects=[],
        delivery={"health": "healthy"},
    )

    class FakeWebSocket:
        def __init__(self) -> None:
            self.headers = {"authorization": f"Bearer {'x' * 43}"}
            self.messages = [
                {"type": "websocket.receive", "bytes": encoded},
                {"type": "websocket.disconnect"},
            ]
            self.sent: list[bytes] = []
            self.accepted = False

        async def accept(self) -> None:
            self.accepted = True

        async def receive(self) -> dict[str, object]:
            await asyncio.sleep(0)
            return self.messages.pop(0)

        async def send_bytes(self, payload: bytes) -> None:
            self.sent.append(payload)

        async def close(self, **_: object) -> None:
            return

    websocket = FakeWebSocket()
    asyncio.run(router.routes[0].endpoint(websocket))  # type: ignore[attr-defined]

    assert websocket.accepted is True
    assert published.is_set()
    assert received == [encoded]