"""Who may see a device's camera: superusers only, or every member too.

``viewers`` on a device covers its ``image`` and
``video`` messages, their bytes, frames on the socket and its live view. A
device credential is bounded by its dashboards instead.
"""

from typing import Any

import pytest
from fastapi.testclient import TestClient

from fluksio.core.config import settings
from fluksio.flow.demand import demand
from tests.api.routes.test_cameras import (
    CAMERA_NODE,
    REPORT,
    _ticket,
    frame,
    wait_for,
)
from tests.api.routes.test_devices import (
    FLOWS,
    _dashboard_with,
    _pair,
    _panels,
    _publish_flow,
)

API = settings.API_V1_STR
PREFIX = f"{API}/devices"
SOCKET = f"{API}/flows/ws"


@pytest.fixture
def camera(client: TestClient, superuser_token_headers: dict[str, str]):
    """``cam1`` with a camera wired in ``cams``, and a frame on it."""
    _panels(client, superuser_token_headers, {"panels": [{"id": "cam1"}]})
    device = _pair(client, superuser_token_headers, "cam1")
    reported = client.put(f"{PREFIX}/cam1/report", headers=device, json=REPORT)
    assert reported.status_code == 200, reported.text
    _publish_flow(client, superuser_token_headers, "cams", [CAMERA_NODE])
    controller = client.app.state.flow_controller
    with client.websocket_connect(SOCKET, headers=device) as ws:
        ws.receive_json()
        for index in range(3):
            ws.send_bytes(frame("cams.camera", b"\xff\xd8 frame %d" % index))
        wait_for(
            lambda: (controller.values().get("cams.camera") or {})
            .get("value", {})
            .get("size")
            == len(b"\xff\xd8 frame 2")
        )
    controller.publish_message("cams.person", True)
    yield device
    client.delete(f"{FLOWS}/cams", headers=superuser_token_headers)
    client.delete(f"{PREFIX}/cam1", headers=superuser_token_headers)


def _viewers(client: TestClient, headers: dict[str, str], viewers: str) -> None:
    current = next(
        one
        for one in client.get(f"{PREFIX}/", headers=headers).json()["data"]
        if one["id"] == "cam1"
    )
    body = {
        key: current[key] for key in ("id", "title", "dashboards", "touch", "version")
    }
    saved = client.put(
        f"{PREFIX}/cam1", headers=headers, json={**body, "viewers": viewers}
    )
    assert saved.status_code == 200, saved.text


def _surfaces(client: TestClient, headers: dict[str, str]) -> dict[str, Any]:
    """What one account gets of the camera on every surface but the socket."""
    controller = client.app.state.flow_controller
    digest = controller.values()["cams.camera"]["value"]["digest"]
    catalog = {
        info["name"]: info
        for info in client.get(f"{API}/messages/", headers=headers).json()["data"]
    }
    return {
        "catalog": catalog["cams.camera"]["value"] is not None,
        "person": catalog["cams.person"]["value"],
        "history": client.get(
            f"{API}/messages/cams.camera/history", headers=headers
        ).status_code,
        "flow_history": client.get(
            f"{FLOWS}/cams/history/camera", headers=headers
        ).status_code,
        "state": "cams.camera"
        in client.get(f"{FLOWS}/cams/state", headers=headers).json()["values"],
        "artifact": client.get(
            f"{API}/artifacts/{digest}", headers=headers
        ).status_code,
        "ice": client.get(f"{PREFIX}/cam1/ice", headers=headers).status_code,
    }


def _live_start(client: TestClient, headers: dict[str, str]) -> str:
    """The reason a viewer's live view of cam1 ends with right away."""
    with client.websocket_connect(_ticket(client, headers)) as ws:
        ws.receive_json()
        ws.send_json(
            {"type": "live", "device": "cam1", "session": "s-access", "action": "start"}
        )
        while True:
            message = ws.receive_json()
            if message.get("type") == "live":
                return str(message.get("reason"))


@pytest.mark.usefixtures("camera")
def test_a_camera_is_for_superusers_until_opened_to_members(
    client: TestClient,
    superuser_token_headers: dict[str, str],
    normal_user_token_headers: dict[str, str],
) -> None:
    member, admin = normal_user_token_headers, superuser_token_headers
    open_to_all = {
        "catalog": True,
        "person": True,
        "history": 200,
        "flow_history": 200,
        "state": True,
        "artifact": 200,
        "ice": 200,
    }
    assert _surfaces(client, admin) == open_to_all
    assert _surfaces(client, member) == {
        "catalog": False,
        # Not a camera: its bool is anyone's.
        "person": True,
        "history": 403,
        "flow_history": 403,
        "state": False,
        "artifact": 403,
        "ice": 403,
    }
    # The frames before the current one too.
    store = client.app.state.artifact_store
    older = store.put([b"\xff\xd8 older"], media_type="image/jpeg", volatile=True)
    client.app.state.flow_controller.publish_message("cams.camera", older)
    newer = store.put([b"\xff\xd8 newer"], media_type="image/jpeg", volatile=True)
    client.app.state.flow_controller.publish_message("cams.camera", newer)
    fetched = client.get(f"{API}/artifacts/{older['digest']}", headers=member)
    assert fetched.status_code == 403
    assert _live_start(client, member) == "forbidden"
    assert _live_start(client, admin) == "offline"

    _viewers(client, admin, "members")
    assert _surfaces(client, member) == open_to_all
    assert _live_start(client, member) == "offline"


@pytest.mark.usefixtures("camera")
def test_a_members_socket_carries_no_frames_of_a_closed_camera(
    client: TestClient,
    superuser_token_headers: dict[str, str],
    normal_user_token_headers: dict[str, str],
    camera: dict[str, str],
) -> None:
    """Neither its values, nor the bytes, nor demand on the device; and the
    socket is given them once the camera is opened."""
    with client.websocket_connect(_ticket(client, normal_user_token_headers)) as ws:
        snapshot = ws.receive_json()
        assert "cams.camera" not in snapshot["values"]
        assert "cams.person" in snapshot["values"]
        ws.send_json({"type": "media", "names": ["cams.camera", "cams.person"]})
        wait_for(lambda: demand.viewed() == {"cams.person"})
        assert demand.ports("cam1") == []

        with client.websocket_connect(SOCKET, headers=camera) as device:
            device.receive_json()
            device.send_bytes(frame("cams.camera", b"\xff\xd8 not for members"))
            client.app.state.flow_controller.publish_message("cams.person", False)
            # The person's flag arrives, the frame before it did not.
            while True:
                message = ws.receive_json()
                names = [
                    event.get("name")
                    for event in message.get("events", [message])
                    if event.get("type") == "message_value"
                ]
                assert "cams.camera" not in names
                if "cams.person" in names:
                    break

        _viewers(client, superuser_token_headers, "members")
        while True:
            message = ws.receive_json()
            if message.get("type") == "snapshot":
                break
        assert "cams.camera" in message["values"]


@pytest.mark.usefixtures("camera")
def test_a_wall_panel_shows_the_camera_its_dashboards_carry(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """Assigned by a superuser, so the panel's own scope is what bounds it."""
    _dashboard_with(
        client,
        superuser_token_headers,
        "panel_cams",
        [
            {
                "id": "w_cam",
                "type": "media",
                "config": {"message": "cams.camera", "dtype": "image"},
            }
        ],
    )
    _panels(
        client,
        superuser_token_headers,
        {"panels": [{"id": "cam1"}, {"id": "hall", "dashboards": ["panel_cams"]}]},
    )
    panel = _pair(client, superuser_token_headers, "hall")
    try:
        value = client.app.state.flow_controller.values()["cams.camera"]["value"]
        fetched = client.get(f"{API}/artifacts/{value['digest']}", headers=panel)
        assert fetched.status_code == 200
        with client.websocket_connect(SOCKET, headers=panel) as ws:
            assert "cams.camera" in ws.receive_json()["values"]
    finally:
        client.delete(f"{PREFIX}/hall", headers=superuser_token_headers)