diff --git a/CHANGELOG.md b/CHANGELOG.md index 8dd2d39..6c128de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,26 @@ Versions follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). --- +## [0.22.0] — 2026-07-10 + +### Added + +**`circuitforge_core.signal_bus`** — generic SSE event publisher for real-time signal streams (MIT, closes #58) + +- `SignalBus` — publish/subscribe bus for real-time events over Server-Sent Events. One instance per service; `publish()` is thread-safe (safe to call from sync producer threads such as OpenCV, Meshtastic, or PyPubSub callbacks) via `loop.call_soon_threadsafe()`. `subscribe(request)` returns a FastAPI `StreamingResponse` — each subscriber gets an independent bounded `asyncio.Queue` (default 100 events) with oldest-drop-on-overflow, plus a 15s keepalive comment to keep proxies from closing idle connections. +- `SignalEvent` — frozen dataclass (`source`, `kind`, `payload`, auto-populated ISO-8601 UTC `timestamp`) with `to_sse()` wire-format serialization. +- New `signal-bus` extra (`fastapi>=0.110`). + +**`circuitforge_core.video.app`** — `POST /caption/upload` endpoint + +Accepts a multipart video file upload, writes it to a temp file, captions it via the configured backend, then deletes the temp file. Lets callers without filesystem access to the video-service node (e.g. a product on one host posting to cf-video on another over an SSH tunnel) caption a video without a shared mount. `video-service` extra now also pulls in `python-multipart`, required by FastAPI's `UploadFile` parsing. + +### Docs + +- `mkdocs.yml` — new blue-grey/cyan palette, wired to a central `docs/stylesheets/theme.css` for consistent theme-aware styling across the site (light and dark mode). + +--- + ## [0.20.0] — 2026-05-05 ### Fixed / Enhanced diff --git a/README.md b/README.md index d06acd5..9e1b167 100644 --- a/README.md +++ b/README.md @@ -42,6 +42,7 @@ pip install circuitforge-core[tts-chatterbox] # Text-to-speech via Chatter pip install circuitforge-core[reranker-qwen3] # Reranking via Qwen3 pip install circuitforge-core[video-service] # Video captioning service (Marlin-2B) pip install circuitforge-core[mqtt] # MQTT broker client +pip install circuitforge-core[signal-bus] # SSE event publisher for real-time signal streams pip install circuitforge-core[meshtastic-service] # Meshtastic mesh radio + MQTT + FastAPI pip install circuitforge-core[memory] # Knowledge graph via mnemo sidecar pip install circuitforge-core[community] # PostgreSQL-backed community store @@ -82,6 +83,7 @@ pip install circuitforge-core[dev] # All dev dependencies | `cloud_session` | Implemented | Cloud session management primitives | | `input` | Implemented | Input handling — MediaPipe gesture recognition | | `job_quality` | Implemented | Job listing quality scoring and signal extraction | +| `signal_bus` | Implemented | Generic SSE event publisher for real-time signal streams | | `vision` | Stub | Vision router (moondream2 / SigLIP dispatch — planned) | | `wizard` | Stub | First-run wizard base class — products subclass `BaseWizard` | | `pipeline` | Stub | Staging queue base — products provide concrete schema | diff --git a/circuitforge_core/signal_bus/__init__.py b/circuitforge_core/signal_bus/__init__.py new file mode 100644 index 0000000..23f92ae --- /dev/null +++ b/circuitforge_core/signal_bus/__init__.py @@ -0,0 +1,36 @@ +"""cf-core signal_bus — generic SSE event publisher for real-time signal streams. + +MIT licensed. Ticket: cf-core #58. + +## Quick start + + from circuitforge_core.signal_bus import SignalBus, SignalEvent + from fastapi import Request + + bus = SignalBus() + + # Producer — safe to call from sync threads (OpenCV, Meshtastic, PyPubSub) + bus.publish(SignalEvent( + source="merlin", + kind="gesture", + payload={"name": "open_palm", "confidence": 0.94}, + )) + + # Consumer — SSE endpoint + @app.get("/events") + async def events(request: Request): + return bus.subscribe(request) + +## Notes + +- Each subscriber gets an independent bounded asyncio.Queue (default: 100 events). +- When full, the oldest event is dropped to make room for the newest. +- publish() is thread-safe via loop.call_soon_threadsafe(). +- Keepalive comment sent every 15 s to keep proxies from closing idle connections. +- Install: no extra dependencies beyond fastapi (already a core dep for service products). +""" + +from circuitforge_core.signal_bus.bus import SignalBus +from circuitforge_core.signal_bus.models import SignalEvent + +__all__ = ["SignalBus", "SignalEvent"] diff --git a/circuitforge_core/signal_bus/bus.py b/circuitforge_core/signal_bus/bus.py new file mode 100644 index 0000000..30c8b56 --- /dev/null +++ b/circuitforge_core/signal_bus/bus.py @@ -0,0 +1,133 @@ +"""SignalBus — generic SSE event publisher for real-time signal streams. + +MIT licensed. +""" +from __future__ import annotations + +import asyncio +import logging +import threading +from typing import AsyncIterator + +from fastapi import Request +from fastapi.responses import StreamingResponse + +from circuitforge_core.signal_bus.models import SignalEvent + +logger = logging.getLogger(__name__) + +_DEFAULT_KEEPALIVE = 15.0 # seconds between SSE keepalive comments + + +def _enqueue_drop_oldest(q: asyncio.Queue, item: str) -> None: + """Put item on queue; if full, drop the oldest entry to make room. + + Designed to run on the event loop thread via call_soon_threadsafe. + """ + if q.full(): + try: + q.get_nowait() + except asyncio.QueueEmpty: + pass + try: + q.put_nowait(item) + except asyncio.QueueFull: + pass # another thread won the race; drop the new item + + +class SignalBus: + """Publish-subscribe bus for real-time signal events over SSE. + + One bus instance per service. Producers call publish() (sync-safe); + consumers mount subscribe() as a FastAPI endpoint. + + Args: + queue_size: Max buffered events per subscriber before oldest is dropped. + """ + + def __init__(self, queue_size: int = 100, keepalive_interval: float = _DEFAULT_KEEPALIVE) -> None: + self._queue_size = queue_size + self._keepalive_interval = keepalive_interval + # int(id(queue)) → queue; guarded by _lock for publish() from sync threads + self._subscribers: dict[int, asyncio.Queue] = {} + self._lock = threading.Lock() + # Captured on first subscribe() call; set from the FastAPI event loop + self._loop: asyncio.AbstractEventLoop | None = None + + # ------------------------------------------------------------------ + # Producer API + # ------------------------------------------------------------------ + + def publish(self, event: SignalEvent) -> None: + """Publish an event to all active subscribers. + + Thread-safe — may be called from sync threads (OpenCV, Meshtastic, + PyPubSub callbacks, etc.). No-op when no subscribers are connected. + """ + loop = self._loop + if loop is None or not loop.is_running(): + return + + sse = event.to_sse() + with self._lock: + subs = list(self._subscribers.values()) + + for q in subs: + loop.call_soon_threadsafe(_enqueue_drop_oldest, q, sse) + + # ------------------------------------------------------------------ + # Consumer API + # ------------------------------------------------------------------ + + def subscribe(self, request: Request) -> StreamingResponse: + """Return an SSE StreamingResponse for a FastAPI endpoint. + + Each caller gets an independent queue. The subscriber is removed + automatically when the client disconnects. + + Usage:: + + @app.get("/events") + async def events(request: Request): + return bus.subscribe(request) + """ + loop = asyncio.get_event_loop() + self._loop = loop + + q: asyncio.Queue = asyncio.Queue(maxsize=self._queue_size) + sub_id = id(q) + with self._lock: + self._subscribers[sub_id] = q + + logger.debug("signal_bus: subscriber %d connected (%d total)", sub_id, len(self._subscribers)) + + async def _generate() -> AsyncIterator[str]: + # Starlette calls aclose() on this generator when the client disconnects, + # injecting GeneratorExit. Polling request.is_disconnected() is unreliable + # in ASGI test transports and adds latency in production. + try: + while True: + try: + chunk = await asyncio.wait_for(q.get(), timeout=self._keepalive_interval) + yield chunk + except asyncio.TimeoutError: + yield ": keepalive\n\n" + finally: + with self._lock: + self._subscribers.pop(sub_id, None) + logger.debug( + "signal_bus: subscriber %d disconnected (%d remaining)", + sub_id, + len(self._subscribers), + ) + + return StreamingResponse(_generate(), media_type="text/event-stream") + + # ------------------------------------------------------------------ + # Introspection + # ------------------------------------------------------------------ + + @property + def subscriber_count(self) -> int: + with self._lock: + return len(self._subscribers) diff --git a/circuitforge_core/signal_bus/models.py b/circuitforge_core/signal_bus/models.py new file mode 100644 index 0000000..846dd70 --- /dev/null +++ b/circuitforge_core/signal_bus/models.py @@ -0,0 +1,42 @@ +"""Data models for the cf-core signal_bus module. + +MIT licensed. +""" +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from datetime import datetime, timezone + + +def _utcnow() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +@dataclass(frozen=True) +class SignalEvent: + """An event published to the signal bus. + + Args: + source: Originating service, e.g. "merlin", "linnet". + kind: Event type, e.g. "gesture", "tone", "alpha_rising". + payload: Arbitrary dict — contents are opaque to the bus. + timestamp: ISO-8601 UTC. Auto-populated if not supplied. + """ + + source: str + kind: str + payload: dict + timestamp: str = field(default_factory=_utcnow) + + def to_sse(self) -> str: + """Serialize to SSE wire format: ``data: {...}\\n\\n``.""" + data = json.dumps( + { + "source": self.source, + "kind": self.kind, + "payload": self.payload, + "timestamp": self.timestamp, + } + ) + return f"data: {data}\n\n" diff --git a/circuitforge_core/video/app.py b/circuitforge_core/video/app.py index b2a3e64..df8627b 100644 --- a/circuitforge_core/video/app.py +++ b/circuitforge_core/video/app.py @@ -32,7 +32,10 @@ import logging import os from typing import Any -from fastapi import FastAPI, HTTPException +import shutil +import tempfile + +from fastapi import FastAPI, HTTPException, UploadFile from pydantic import BaseModel, Field from circuitforge_core.video.backends.base import VideoBackend, make_video_backend @@ -139,6 +142,46 @@ def find(req: FindRequest) -> FindResponse: ) +@app.post("/caption/upload", response_model=CaptionResponse) +async def caption_upload( + file: UploadFile, + max_new_tokens: int = 2048, +) -> CaptionResponse: + """Accept a video file upload, caption it, then delete the temp file. + + Allows callers that don't share a filesystem with this node (e.g. Waxwing + on Heimdall posting to cf-video on Muninn via the SSH tunnel port forward). + """ + if _backend is None: + raise HTTPException(503, detail="backend not initialised") + + suffix = f".{file.filename.rsplit('.', 1)[-1]}" if file.filename and "." in file.filename else ".mkv" + tmp_fd, tmp_path = tempfile.mkstemp(suffix=suffix) + try: + with os.fdopen(tmp_fd, "wb") as dst: + shutil.copyfileobj(file.file, dst) + + result = _backend.caption(tmp_path, max_new_tokens=max_new_tokens) + except Exception as exc: + logging.exception("caption/upload failed") + raise HTTPException(500, detail=str(exc)) from exc + finally: + try: + os.unlink(tmp_path) + except OSError: + pass + + return CaptionResponse( + scene=result.scene, + events=[ + VideoEventOut(start=ev.start, end=ev.end, description=ev.description) + for ev in result.events + ], + caption=result.caption, + model=result.model, + ) + + # ── CLI entry point ─────────────────────────────────────────────────────────── def _parse_args() -> argparse.Namespace: diff --git a/docs/stylesheets/theme.css b/docs/stylesheets/theme.css new file mode 100644 index 0000000..170c031 --- /dev/null +++ b/docs/stylesheets/theme.css @@ -0,0 +1,39 @@ +/* + * circuitforge-core docs theme + * + * Central place for palette tweaks and small layout fixes on top of + * mkdocs-material's blue grey / cyan palette (see mkdocs.yml). Keep + * product-specific overrides out of individual markdown pages — add + * them here so light/dark mode and mobile layout stay consistent + * across the whole site. + */ + +:root { + --cf-accent-strong: #00acc1; /* cyan 600, matches material accent */ + --cf-code-bg-light: #eceff1; /* blue grey 50 */ + --cf-code-bg-dark: #263238; /* blue grey 900 */ +} + +/* Slightly stronger contrast for inline code than material's default, + in both color schemes. */ +[data-md-color-scheme="default"] { + --md-code-bg-color: var(--cf-code-bg-light); +} + +[data-md-color-scheme="slate"] { + --md-code-bg-color: var(--cf-code-bg-dark); +} + +/* Module reference tables can get wide (Module | Status | Description); + let them scroll horizontally on narrow viewports instead of overflowing + the page or squeezing text unreadably small. */ +.md-typeset table:not([class]) { + display: block; + overflow-x: auto; +} + +@media screen and (max-width: 44.9375em) { + .md-typeset table:not([class]) { + font-size: 0.7rem; + } +} diff --git a/mkdocs.yml b/mkdocs.yml index 4766f51..59551c8 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -9,14 +9,14 @@ theme: name: material palette: - scheme: default - primary: deep purple - accent: purple + primary: blue grey + accent: cyan toggle: icon: material/brightness-7 name: Switch to dark mode - scheme: slate - primary: deep purple - accent: purple + primary: blue grey + accent: cyan toggle: icon: material/brightness-4 name: Switch to light mode @@ -78,5 +78,8 @@ nav: - Editable Install Pattern: developer/editable-install.md - BSL vs MIT Boundaries: developer/licensing.md +extra_css: + - stylesheets/theme.css + extra_javascript: - plausible.js diff --git a/pyproject.toml b/pyproject.toml index 6713b7f..e8d37d0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "circuitforge-core" -version = "0.21.0" +version = "0.22.0" description = "Shared scaffold for CircuitForge products (MIT)" requires-python = ">=3.11" dependencies = [ @@ -20,6 +20,9 @@ memory = [ sync = [ "fastapi>=0.110", ] +signal-bus = [ + "fastapi>=0.110", +] community = [ "psycopg2>=2.9", ] @@ -86,6 +89,7 @@ video-service = [ "circuitforge-core[video-marlin]", "fastapi>=0.110", "uvicorn[standard]>=0.29", + "python-multipart>=0.0.9", ] vision-siglip = [ "torch>=2.0", diff --git a/tests/test_signal_bus/__init__.py b/tests/test_signal_bus/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_signal_bus/test_signal_bus.py b/tests/test_signal_bus/test_signal_bus.py new file mode 100644 index 0000000..8e78b7d --- /dev/null +++ b/tests/test_signal_bus/test_signal_bus.py @@ -0,0 +1,192 @@ +"""Tests for SignalBus and SignalEvent.""" +from __future__ import annotations + +import asyncio +import json +from unittest.mock import MagicMock + +import pytest +from fastapi import Request +from fastapi.responses import StreamingResponse + +from circuitforge_core.signal_bus import SignalBus, SignalEvent + + +# --------------------------------------------------------------------------- +# SignalEvent +# --------------------------------------------------------------------------- + +class TestSignalEvent: + def test_sse_format(self) -> None: + ev = SignalEvent(source="merlin", kind="gesture", payload={"name": "open_palm"}, timestamp="2026-01-01T00:00:00Z") + sse = ev.to_sse() + assert sse.startswith("data: ") + assert sse.endswith("\n\n") + + def test_sse_contains_all_fields(self) -> None: + ev = SignalEvent(source="linnet", kind="tone", payload={"label": "sarcasm"}, timestamp="2026-01-01T00:00:00Z") + data = json.loads(ev.to_sse()[len("data: "):].strip()) + assert data["source"] == "linnet" + assert data["kind"] == "tone" + assert data["payload"] == {"label": "sarcasm"} + assert data["timestamp"] == "2026-01-01T00:00:00Z" + + def test_timestamp_auto_populated(self) -> None: + ev = SignalEvent(source="s", kind="k", payload={}) + assert ev.timestamp # not empty + assert "T" in ev.timestamp # ISO format + + def test_frozen(self) -> None: + ev = SignalEvent(source="s", kind="k", payload={}) + with pytest.raises((AttributeError, TypeError)): + ev.source = "other" # type: ignore + + +# --------------------------------------------------------------------------- +# SignalBus — unit (direct queue inspection) +# --------------------------------------------------------------------------- + +class TestSignalBusUnit: + def test_publish_noop_before_subscribe(self) -> None: + bus = SignalBus() + ev = SignalEvent(source="s", kind="k", payload={}) + bus.publish(ev) # must not raise + + def test_subscriber_count_zero_initially(self) -> None: + bus = SignalBus() + assert bus.subscriber_count == 0 + + @pytest.mark.asyncio + async def test_enqueue_drop_oldest_on_overflow(self) -> None: + from circuitforge_core.signal_bus.bus import _enqueue_drop_oldest + q: asyncio.Queue = asyncio.Queue(maxsize=2) + _enqueue_drop_oldest(q, "a") + _enqueue_drop_oldest(q, "b") + _enqueue_drop_oldest(q, "c") # drops "a" + assert q.qsize() == 2 + assert q.get_nowait() == "b" + assert q.get_nowait() == "c" + + @pytest.mark.asyncio + async def test_enqueue_drop_oldest_empty_queue(self) -> None: + from circuitforge_core.signal_bus.bus import _enqueue_drop_oldest + q: asyncio.Queue = asyncio.Queue(maxsize=5) + _enqueue_drop_oldest(q, "x") + assert q.get_nowait() == "x" + + +# --------------------------------------------------------------------------- +# SignalBus — integration via FastAPI TestClient +# --------------------------------------------------------------------------- + +def _make_bus() -> SignalBus: + return SignalBus(keepalive_interval=0.1) + + +def _inject_subscriber(bus: SignalBus, q: "asyncio.Queue[str]") -> None: + """Register a pre-built queue as a subscriber (test helper).""" + bus._loop = asyncio.get_event_loop() + with bus._lock: + bus._subscribers[id(q)] = q + + +class TestSignalBusIntegration: + """Integration tests verifying publish/subscribe queue mechanics directly. + + httpx.ASGITransport cannot cancel an unbounded SSE generator — the transport + buffers outbound chunks but never delivers a disconnect signal to the ASGI + generator, so any test that exits a client.stream() block before the generator + finishes hangs forever. These tests bypass ASGI and inspect the subscriber + queues directly, which is both faster and more reliable. + + The ASGI wiring is a one-liner (StreamingResponse) verified by + test_subscribe_returns_streaming_response below. + """ + + @pytest.mark.asyncio + async def test_event_delivered_to_subscriber_queue(self) -> None: + """publish() enqueues an SSE-formatted string to all subscriber queues.""" + bus = _make_bus() + q: asyncio.Queue = asyncio.Queue(maxsize=10) + _inject_subscriber(bus, q) + + ev = SignalEvent( + source="merlin", kind="gesture", + payload={"name": "open_palm", "confidence": 0.94}, + timestamp="2026-01-01T00:00:00Z", + ) + bus.publish(ev) + await asyncio.sleep(0) # allow call_soon_threadsafe callback to run + + assert not q.empty() + sse = q.get_nowait() + data = json.loads(sse[len("data: "):].strip()) + assert data["source"] == "merlin" + assert data["kind"] == "gesture" + assert data["payload"]["name"] == "open_palm" + + @pytest.mark.asyncio + async def test_multiple_subscribers_each_receive_event(self) -> None: + """Every connected subscriber queue receives its own copy of the event.""" + bus = _make_bus() + q1: asyncio.Queue = asyncio.Queue(maxsize=10) + q2: asyncio.Queue = asyncio.Queue(maxsize=10) + _inject_subscriber(bus, q1) + bus._loop = asyncio.get_event_loop() + with bus._lock: + bus._subscribers[id(q2)] = q2 + + ev = SignalEvent(source="linnet", kind="tone", payload={}, timestamp="T") + bus.publish(ev) + await asyncio.sleep(0) + + assert not q1.empty() + assert not q2.empty() + d1 = json.loads(q1.get_nowait()[len("data: "):].strip()) + d2 = json.loads(q2.get_nowait()[len("data: "):].strip()) + assert d1["kind"] == "tone" + assert d2["kind"] == "tone" + + @pytest.mark.asyncio + async def test_publish_safe_while_subscribed(self) -> None: + """Rapid publish() calls with active subscribers must not raise.""" + bus = _make_bus() + q: asyncio.Queue = asyncio.Queue(maxsize=100) + _inject_subscriber(bus, q) + + errors: list[Exception] = [] + try: + for i in range(20): + bus.publish(SignalEvent(source="s", kind="k", payload={"i": i}, timestamp="T")) + except Exception as exc: + errors.append(exc) + + await asyncio.sleep(0) + assert not errors + assert q.qsize() > 0 + + @pytest.mark.asyncio + async def test_overflow_does_not_block_producer(self) -> None: + """A slow consumer with a tiny queue must not block publish().""" + bus = SignalBus(queue_size=2, keepalive_interval=0.1) + q: asyncio.Queue = asyncio.Queue(maxsize=2) + _inject_subscriber(bus, q) + + for i in range(50): + bus.publish(SignalEvent(source="s", kind="k", payload={"i": i}, timestamp="T")) + + await asyncio.sleep(0) + # Drop-oldest: only the last 2 events survive + assert q.qsize() == 2 + + @pytest.mark.asyncio + async def test_subscribe_returns_streaming_response(self) -> None: + """subscribe() returns a StreamingResponse with text/event-stream media type.""" + bus = _make_bus() + mock_req = MagicMock(spec=Request) + + resp = bus.subscribe(mock_req) + + assert isinstance(resp, StreamingResponse) + assert resp.media_type == "text/event-stream" + assert bus.subscriber_count == 1 diff --git a/tests/test_video/test_app.py b/tests/test_video/test_app.py index 53eacc7..4c56c09 100644 --- a/tests/test_video/test_app.py +++ b/tests/test_video/test_app.py @@ -7,6 +7,8 @@ so a zero-byte placeholder is sufficient. """ from __future__ import annotations +import os + import pytest from fastapi.testclient import TestClient @@ -234,3 +236,116 @@ def test_find_rejects_max_new_tokens_above_max(client, video_file): json={"video_path": video_file, "event": "wave", "max_new_tokens": 99999}, ) assert resp.status_code == 422 + + +# ── /caption/upload ─────────────────────────────────────────────────────────── + + +def test_caption_upload_returns_200(client): + resp = client.post( + "/caption/upload", + files={"file": ("sample.mp4", b"\x00" * 16, "video/mp4")}, + ) + assert resp.status_code == 200 + + +def test_caption_upload_response_matches_caption_shape(client): + data = client.post( + "/caption/upload", + files={"file": ("sample.mp4", b"\x00" * 16, "video/mp4")}, + ).json() + assert isinstance(data["scene"], str) and data["scene"] + assert isinstance(data["events"], list) and len(data["events"]) >= 1 + assert isinstance(data["caption"], str) and data["caption"] + assert isinstance(data["model"], str) + + +def test_caption_upload_preserves_file_extension(client, monkeypatch): + seen_paths: list[str] = [] + original_caption = MockVideoBackend.caption + + def _spy_caption(self, video_path, *, max_new_tokens=2048): + seen_paths.append(video_path) + return original_caption(self, video_path, max_new_tokens=max_new_tokens) + + monkeypatch.setattr(MockVideoBackend, "caption", _spy_caption) + + resp = client.post( + "/caption/upload", + files={"file": ("sample.mkv", b"\x00" * 16, "video/x-matroska")}, + ) + assert resp.status_code == 200 + assert seen_paths and seen_paths[0].endswith(".mkv") + + +def test_caption_upload_defaults_extension_when_filename_has_none(client, monkeypatch): + seen_paths: list[str] = [] + original_caption = MockVideoBackend.caption + + def _spy_caption(self, video_path, *, max_new_tokens=2048): + seen_paths.append(video_path) + return original_caption(self, video_path, max_new_tokens=max_new_tokens) + + monkeypatch.setattr(MockVideoBackend, "caption", _spy_caption) + + resp = client.post( + "/caption/upload", + files={"file": ("sample", b"\x00" * 16, "application/octet-stream")}, + ) + assert resp.status_code == 200 + assert seen_paths and seen_paths[0].endswith(".mkv") + + +def test_caption_upload_removes_temp_file_after_request(client, monkeypatch): + captured_paths: list[str] = [] + original_caption = MockVideoBackend.caption + + def _spy_caption(self, video_path, *, max_new_tokens=2048): + captured_paths.append(video_path) + return original_caption(self, video_path, max_new_tokens=max_new_tokens) + + monkeypatch.setattr(MockVideoBackend, "caption", _spy_caption) + + resp = client.post( + "/caption/upload", + files={"file": ("sample.mp4", b"\x00" * 16, "video/mp4")}, + ) + assert resp.status_code == 200 + assert captured_paths + assert not os.path.exists(captured_paths[0]) + + +def test_caption_upload_503_when_no_backend(client): + video_app._backend = None + resp = client.post( + "/caption/upload", + files={"file": ("sample.mp4", b"\x00" * 16, "video/mp4")}, + ) + assert resp.status_code == 503 + + +def test_caption_upload_500_and_cleans_up_on_backend_error(client, monkeypatch): + captured_paths: list[str] = [] + + def _raise(self, video_path, *, max_new_tokens=2048): + captured_paths.append(video_path) + raise RuntimeError("boom") + + monkeypatch.setattr(MockVideoBackend, "caption", _raise) + + resp = client.post( + "/caption/upload", + files={"file": ("sample.mp4", b"\x00" * 16, "video/mp4")}, + ) + assert resp.status_code == 500 + assert captured_paths + assert not os.path.exists(captured_paths[0]) + + +def test_caption_upload_custom_max_new_tokens(client): + resp = client.post( + "/caption/upload", + params={"max_new_tokens": 512}, + files={"file": ("sample.mp4", b"\x00" * 16, "video/mp4")}, + ) + assert resp.status_code == 200