festifun/backend/tests/test_capture.py

"""Live capture tests (Phase 5) — upload safety, the env gate, and the device monitor.

The router is exercised directly on a throwaway app: ``capture_enabled()`` is read at
``api`` import time, so mounting it here is both simpler and keeps the gate test honest
(the real app must NOT expose these routes unless FESTIVAL4D_CAPTURE is set).
"""

from __future__ import annotations

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from festival4d import capture, config


@pytest.fixture()
def client():
    app = FastAPI()
    app.include_router(capture.router)
    config.RAW_DIR.mkdir(parents=True, exist_ok=True)
    with TestClient(app) as c:
        yield c


def _upload(client, name: str, filename: str, data: bytes = b"fake-video-bytes"):
    return client.post(
        "/api/capture/upload",
        data={"device_name": name},
        files={"file": (filename, data, "video/webm")},
    )


# --- the gate ---------------------------------------------------------------------------------

def test_capture_disabled_by_default():
    """FESTIVAL4D_CAPTURE is unset in tests => the real app exposes no upload surface."""
    assert capture.capture_enabled() is False
    from festival4d import api

    paths = {r.path for r in api.app.routes}
    assert "/capture" not in paths
    assert not any(p.startswith("/api/capture") for p in paths)


@pytest.mark.parametrize("val,expected", [("1", True), ("true", True), ("on", True),
                                          ("0", False), ("", False), ("no", False)])
def test_capture_enabled_parsing(monkeypatch, val, expected):
    monkeypatch.setenv("FESTIVAL4D_CAPTURE", val)
    assert capture.capture_enabled() is expected


# --- upload -----------------------------------------------------------------------------------

def test_upload_saves_sanitized_file(client):
    r = _upload(client, "Phone Left!", "clip.webm")
    assert r.status_code == 200, r.text
    body = r.json()
    saved = config.RAW_DIR / body["filename"]
    assert saved.is_file() and saved.read_bytes() == b"fake-video-bytes"
    assert body["filename"].startswith("phone-left-")   # slugified, timestamped
    assert body["filename"].endswith(".webm")
    assert body["bytes"] == len(b"fake-video-bytes")
    saved.unlink()


def test_upload_name_cannot_traverse(client):
    r = _upload(client, "../../etc/passwd", "clip.webm")
    assert r.status_code == 200
    saved = config.RAW_DIR / r.json()["filename"]
    # The slug strips separators entirely; the file lands inside RAW_DIR, nowhere else.
    assert saved.resolve().parent == config.RAW_DIR.resolve()
    assert ".." not in r.json()["filename"]
    saved.unlink()


def test_upload_rejects_bad_container(client):
    r = _upload(client, "cam", "payload.exe")
    assert r.status_code == 400
    assert "unsupported container" in r.json()["detail"]


def test_upload_rejects_empty(client):
    r = _upload(client, "cam", "clip.webm", data=b"")
    assert r.status_code == 400
    assert "empty" in r.json()["detail"]


def test_upload_rejects_oversize(client, monkeypatch):
    monkeypatch.setattr(capture, "MAX_UPLOAD_BYTES", 8)
    r = _upload(client, "cam", "clip.webm", data=b"x" * 64)
    assert r.status_code == 413
    # The partial file must not be left behind for ingest to trip on.
    assert not any(p.name.startswith("cam-") for p in config.RAW_DIR.glob("cam-*.webm"))


# --- monitor ----------------------------------------------------------------------------------

def test_heartbeat_and_devices_monitor(client):
    capture._devices.clear()
    client.post("/api/capture/heartbeat", json={
        "device_id": "d1", "name": "phone-left", "status": "recording",
        "elapsed_s": 3.5, "snapshot": "data:image/jpeg;base64,AAAA",
    })
    client.post("/api/capture/heartbeat", json={"device_id": "d2", "name": "laptop"})
    devs = client.get("/api/capture/devices").json()
    assert [d["name"] for d in devs] == ["laptop", "phone-left"]      # sorted by name
    d1 = next(d for d in devs if d["device_id"] == "d1")
    assert d1["status"] == "recording" and d1["snapshot"].startswith("data:image/jpeg")
    assert "_seen" not in d1                                          # internals not leaked
    capture._devices.clear()


def test_oversized_snapshot_dropped_but_device_kept(client):
    capture._devices.clear()
    client.post("/api/capture/heartbeat", json={
        "device_id": "big", "name": "cam", "snapshot": "data:image/jpeg;base64," + "A" * 400_000,
    })
    devs = client.get("/api/capture/devices").json()
    assert len(devs) == 1 and "snapshot" not in devs[0]
    capture._devices.clear()


def test_stale_devices_pruned(client, monkeypatch):
    capture._devices.clear()
    client.post("/api/capture/heartbeat", json={"device_id": "old", "name": "cam"})
    assert len(client.get("/api/capture/devices").json()) == 1
    monkeypatch.setattr(capture, "DEVICE_STALE_S", -1.0)  # everything is instantly stale
    assert client.get("/api/capture/devices").json() == []
    capture._devices.clear()


def test_capture_page_served(client):
    r = client.get("/capture")
    assert r.status_code == 200
    assert "Festival 4D — Capture" in r.text
    assert "getUserMedia" in r.text        # the page really is the capture app