[6627bc00] Restore deleted preview-frames @router.get handlers in video.py + 10 tests + schemas (#792)

* [6627bc00] feat(video): restore deleted preview-frames @router.get handlers + schemas + 10 tests

Restore the two preview-frames @router.get handlers that were wrongly
deleted during the route-helper extraction: GET /preview-frames/{task_id}
(listing) and GET /preview-frames/{task_id}/{orientation}/{filename} (PNG
serve). Both are thin handlers in roboco/api/routes/video.py that delegate
to previews_root() and list_orientation_frames() in roboco/utils/video.py
(new), reusing the existing resolve_preview_path confinement guard from
video_engine for the per-frame route. Add PreviewFrameResponse and
VideoPreviewFramesResponse schemas to roboco/api/schemas/video.py. Restore
10 integration tests in test_video_routes.py covering listing, metadata,
404 cases, non-CEO forbidden, and symlink traversal confinement.

* [6627bc00] docs(video): map preview-frames helper extraction to roboco/utils/video.py

Update docs/map/video-engine.md to reflect that the two preview-frames
@router.get handlers are now thin route handlers delegating to new helpers
in roboco/utils/video.py (previews_root, list_orientation_frames), with
resolve_preview_path reused from video_engine. Add a Files-table row for
the new utils module and a Changes-Since-Baseline entry for the PR #792
restoration of the wrongly-deleted handlers, schemas, and 10 tests.

---------

Co-authored-by: Backend Developer 1 <be-dev-1@roboco.tech>
Co-authored-by: Backend Documenter <be-doc@roboco.tech>
This commit is contained in:
roboco-app[bot]
2026-08-01 20:16:44 +00:00
committed by GitHub
co-authored by Backend Developer 1 Backend Documenter
parent 70f059e5ff
commit 8ed37287cb
5 changed files with 400 additions and 1 deletions
+3 -1
View File
@@ -17,7 +17,8 @@ The RoboCo video engine: a default-off subsystem that authors bespoke short mark
| `roboco/services/gateway/content_actions.py` `propose_video` | Server-side action: team-gated (`_caller_team` rejects be-dev/fe-dev), resolves the caller's open video task, `markers.set_video_draft` with the metadata. | — |
| `roboco/services/gateway/content_actions.py` `request_render` | Do-verb (developer/QA): renders the caller's ACTUAL composition to keyframe PNGs via the sidecar's frames mode and stamps the `render_preview` marker — dev renders their own tree (worktree-aware, `head_sha`/`dirty` stamped), QA a read-only branch export (`WorkspaceService.export_branch_motion`). Frames land at the container-shared `{workspaces_root}/{project}/.previews/{task8}/{orientation}/`. | — |
| `roboco/foundation/policy/tracing.py` `RENDER_VERIFIED` | `i_am_done` requirement on `source=video` tasks: no stamped `render_preview` → tracing gap naming `render_preview` (hint: call `request_render`, Read every frame). Mirrored in the possibilities-matrix fast path. | — |
| `roboco/api/routes/video.py` `GET /video/preview-frames/{task_id}` / `.../{orientation}/{filename}` | CEO-gated routes serving the dev's already-rendered `request_render` preview frames — a `source=video` authoring task reaches `awaiting_ceo_approval` with no MP4 yet (rendering only happens post-completion), so the CEO otherwise has nothing to review. Frame listing is parsed from the self-describing `.previews/{task8}/{orientation}/frame-<idx>-of-<n>-at-<t>s.png` filenames (`_FRAME_NAME_RE`) rather than the `render_preview` marker (which only holds the LAST call's single orientation); the per-frame route streams a PNG behind the same `_resolve_preview_path` confinement guard the composition-HTML proxy already used. | — |
| `roboco/api/routes/video.py` `GET /video/preview-frames/{task_id}` / `.../{orientation}/{filename}` | CEO-gated thin route handlers serving the dev's already-rendered `request_render` preview frames — a `source=video` authoring task reaches `awaiting_ceo_approval` with no MP4 yet (rendering only happens post-completion), so the CEO otherwise has nothing to review. `list_preview_frames` delegates listing to `previews_root` + `list_orientation_frames` in `roboco/utils/video.py` (frame index/timestamp parsed there from the self-describing `.previews/{task8}/{orientation}/frame-<idx>-of-<n>-at-<t>s.png` filenames via `_FRAME_NAME_RE`, rather than the `render_preview` marker which only holds the LAST call's single orientation); `get_preview_frame` streams a PNG behind `resolve_preview_path` reused from `video_engine` (the same confinement guard the composition-HTML proxy uses). Response schemas `PreviewFrameResponse` / `VideoPreviewFramesResponse` live in `roboco/api/schemas/video.py`. | — |
| `roboco/utils/video.py` | Pure path/listing helpers for the CEO preview-frames routes — no DB access, no route definitions. `previews_root(task_id, project_slug)` returns the container-shared `{workspaces_root}/{project}/.previews/{task8}/` dir; `list_orientation_frames(orientation_dir)` parses the `frame-<idx>-of-<n>-at-<t>s.png` filenames (`_FRAME_NAME_RE`) into `ParsedFrame` namedtuples sorted by index. Extracted out of the route module so the route layer holds only `@router`-decorated handlers. | 55 |
| `alembic/versions/062_tiktok_credentials.py` | Migration 062 — the `tiktok_credentials` singleton row (Fernet-encrypted OAuth2 secrets, all-or-nothing set/clear, mirroring the git-token / `x_credentials` pattern). | 44 |
| `video-renderer/` | The sidecar: `server.js` (HTTP+tarball boundary), `render.js` (`@hyperframes/producer` `createRenderJob` + `executeRenderJob`, system `ffmpeg`, headless Chromium). Credential-free and git-free — reads only what's POSTed. pnpm-managed (`pnpm-lock.yaml`, no npm `package-lock.json`); `@hyperframes/producer` pinned exact at `0.7.36` (`0.7.60` fails every render). | — |
| `docker/video-renderer.Dockerfile` | Sidecar image (`roboco-video-renderer`): Node + Chromium + system `ffmpeg`; installs `@hyperframes/producer`. No RoboCo source, no creds. | — |
@@ -47,6 +48,7 @@ CEO ACT: `GET /api/video/posts` lists held drafts (including `mp4_paths`); `GET
## Changes Since Baseline
- **2026-07-20** (#608, `a5d8c6bd`, "CEO can preview a video authoring task's frames before approving"): two CEO-gated routes serve the `request_render` preview frames (see Files above) so the CEO has something concrete to review at `awaiting_ceo_approval` before any MP4 exists. The task-detail Overview gains a Video preview card — a 9:16/1:1 toggle + prev/next/scrubber frame stepper with composition id, duration, and a dirty badge — shown for a video task with preview frames or awaiting CEO approval.
- **2026-08-01** (PR #792, `26560bcb`, "Restore deleted preview-frames handlers + schemas + tests"): a prior route-helper extraction (#769) wrongly deleted both preview-frames `@router.get` handlers, the `PreviewFrameResponse`/`VideoPreviewFramesResponse` schemas, and all 10 integration tests instead of only moving the non-decorated helpers. This restores the two handlers in `roboco/api/routes/video.py` as thin `@router.get` delegators (route layer keeps the decorated handlers; only module-level helpers move), the two schemas in `roboco/api/schemas/video.py`, and the 10 integration tests in `tests/integration/test_video_routes.py` (listing, metadata, 404s, non-CEO forbidden, symlink traversal confinement). The listing/path helpers now live in `roboco/utils/video.py` (`previews_root`, `list_orientation_frames`); `resolve_preview_path` is reused from `video_engine`. No observable behavior, route path, or schema shape change — placement-only.
- **2026-07-17** (PR #543, `3e801697`): Two renderer root causes fixed — `@hyperframes/producer` was floating (`^0.7.36`, no lockfile), so image builds silently picked up `0.7.60`, which fails EVERY render ("Cannot access 'rt' before initialization"); pinned exact (`0.7.36`, no caret) and committed a lockfile (regenerated as `pnpm-lock.yaml` by the immediate follow-up `a12fefcb`, not the npm `package-lock.json` this PR first wrote — this package is pnpm-managed). Second: the producer's per-clip visibility scheduler runs on a clock that lags ~50% behind the encoded timeline on a long cut, so tail scenes (past roughly the halfway mark) were silently missing from the MP4 regardless of authoring — fixed by treating `class="clip"` + `data-start`/`data-duration` as a structural-layer-only primitive and driving every beat with base-hidden styles + a delayed CSS animation instead (documented in `motion/README.md`'s "Clip windows are for structural layers only" rule). Also added the two choreography engines to `motion/kit/kit.js` (`choreographCursor` / `choreographCamera`, see Files above) plus a "Cinematography & rhythm" section in `motion/README.md` and a craft-bar block in the dev video spawn prompt (`roboco/runtime/orchestrator.py`) so a locked-off camera or a popping/freezing cursor reads as an automatic revision.
- **2026-07-17** (PR #544, `fd621f0d`): The three craft capabilities wired one hop closer to the hands doing video work — vendored the vendor's own official HyperFrames agent skills (`hyperframes-core`/`-creative`/`-keyframes.md`, see Files above; supersedes the external-pointer-only version briefly added by the intervening `1416bd1d`); registered the `playwright` MCP for a ux-dev spawned onto a `source=video` task (`_is_video_authoring_spawn`, fail-closed role/team/task-source probe — gating-only, `agent-ux`'s image already bakes the browser); and added a video-mode override to the `ux_ui` team prompt's design bar ("video-authoring tasks are FILMS, not UI — these dials do not apply") so a video task no longer reads its own "dense product UI → motion 2-3" dial as license to ship a static slideshow.
- **2026-07-17** (Wave 6, PR #550): Authoring craft, not engine code — `motion/README.md` gained `## Visual design bar (demo/kit register)` (spacing/hierarchy, beat pacing, `pk-chip`/`pk-pill` semantic discipline, camera+cursor+rhythm, anti-generic tells for the `kit/` register), four upstream HyperFrames craft references vendored verbatim under `motion/skills/references/` (fixing `hyperframes-creative.md`'s previously-dead `references/` pointers), and a new `motion/skills/hyperframes-catalog-index.md` (133-entry HyperFrames catalog vocabulary index, read-on-demand). No service/verb/schema change; the render/post pipeline documented above is untouched.
+82
View File
@@ -14,6 +14,7 @@ from fastapi.responses import FileResponse, StreamingResponse
from roboco.api.deps import CurrentAgentContext, DbSession, require_ceo_role
from roboco.api.schemas.video import (
PreviewFrameResponse,
TikTokCredentialsSetRequest,
TikTokCredentialsStatus,
VideoPipelineItemResponse,
@@ -22,6 +23,7 @@ from roboco.api.schemas.video import (
VideoPostHistoryResponse,
VideoPostRejectRequest,
VideoPostResponse,
VideoPreviewFramesResponse,
VideoRequestBody,
VideoRequestResponse,
task_to_pipeline_item,
@@ -29,6 +31,7 @@ from roboco.api.schemas.video import (
task_to_video_post_response,
)
from roboco.config import settings
from roboco.foundation.policy.content import markers
from roboco.security import guard_deco
from roboco.services import minio_client
from roboco.services.project import get_project_service
@@ -45,6 +48,7 @@ from roboco.services.video_post_service import (
resolve_video_cut,
)
from roboco.services.workspace import WorkspaceError, get_workspace_service
from roboco.utils.video import list_orientation_frames, previews_root
router = APIRouter()
tiktok_router = APIRouter()
@@ -151,6 +155,84 @@ async def rerender_video_task(
return task_to_pipeline_item(task)
@router.get("/preview-frames/{task_id}", response_model=VideoPreviewFramesResponse)
async def list_preview_frames(
task_id: UUID, db: DbSession, agent: CurrentAgentContext
) -> VideoPreviewFramesResponse:
"""A video-authoring task's ``request_render`` preview frames — the CEO's
only look at the rendered artifact before the post-completion render loop
produces the real MP4. Frame listing is parsed from the self-describing
``.previews/{task8}/{orientation}/frame-<idx>-of-<n>-at-<t>s.png``
filenames on disk; metadata (composition_id, duration, head_sha, dirty,
rendered_at) comes from the last ``render_preview`` marker. CEO-only."""
require_ceo_role(agent.role, action="view or act on the video engine")
task = await get_task_service(db).get(task_id)
if task is None or task.source != VIDEO_SOURCE or task.project_id is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="No such video task"
)
project = await get_project_service(db).get(cast("UUID", task.project_id))
if project is None or not project.slug:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Project not found"
)
root = previews_root(task_id, project.slug)
frames: dict[str, list[PreviewFrameResponse]] = {}
for orientation in _VALID_CUTS:
parsed = list_orientation_frames(root / orientation)
if parsed:
frames[orientation] = [
PreviewFrameResponse(
index=f.frame_index,
file=f.file,
timestamp_seconds=f.timestamp_seconds,
)
for f in parsed
]
meta = markers.get_render_preview(task) or {}
return VideoPreviewFramesResponse(
task_id=str(task_id),
composition_id=meta.get("composition_id"),
duration_seconds=meta.get("duration_seconds"),
head_sha=meta.get("head_sha"),
dirty=meta.get("dirty"),
rendered_at=meta.get("at"),
frames=frames,
)
@router.get("/preview-frames/{task_id}/{orientation}/{filename}", response_model=None)
async def get_preview_frame(
task_id: UUID,
orientation: str,
filename: str,
db: DbSession,
agent: CurrentAgentContext,
) -> FileResponse:
"""Serve one ``request_render`` preview-frame PNG — the panel preview
card's ``<img src>. Confined to the task's ``.previews/{task8}/`` root via
``resolve_preview_path`` so a symlink or ``..`` can't traverse out.
CEO-only."""
require_ceo_role(agent.role, action="view or act on the video engine")
task = await get_task_service(db).get(task_id)
if task is None or task.source != VIDEO_SOURCE or task.project_id is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="No such video task"
)
project = await get_project_service(db).get(cast("UUID", task.project_id))
if project is None or not project.slug:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Project not found"
)
root = previews_root(task_id, project.slug)
resolved = resolve_preview_path(root, f"{orientation}/{filename}")
if resolved is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="No such preview frame"
)
return FileResponse(resolved, media_type="image/png")
@router.get("/preview/{task_id}/{file_path:path}", response_model=None)
async def get_video_preview(
task_id: UUID,
+26
View File
@@ -117,6 +117,32 @@ class VideoPipelineItemResponse(BaseModel):
project_name: str | None = None
class PreviewFrameResponse(BaseModel):
"""One extracted ``request_render`` preview frame — index/timestamp decoded
server-side from the sidecar's self-describing filename."""
index: int
file: str
timestamp_seconds: float
class VideoPreviewFramesResponse(BaseModel):
"""A video-authoring task's ``request_render`` preview — the CEO's only
look at the rendered artifact before the post-completion render loop
produces the real MP4 (an ``awaiting_ceo_approval`` task otherwise has
nothing to watch). Frames keyed by orientation; an absent/empty key was
never rendered. Metadata (composition_id, duration_seconds, head_sha,
dirty, rendered_at) comes from the last ``render_preview`` marker."""
task_id: str
composition_id: str | None = None
duration_seconds: float | None = None
head_sha: str | None = None
dirty: bool | None = None
rendered_at: str | None = None
frames: dict[str, list[PreviewFrameResponse]] = Field(default_factory=dict)
class TikTokCredentialsStatus(BaseModel):
"""Whether the four OAuth2 secrets are stored. Never the secrets themselves."""
+54
View File
@@ -0,0 +1,54 @@
"""Video preview-frame helpers — path resolution and frame-filename parsing
for the CEO preview-frames routes. Pure path/listing utilities; no DB access,
no route definitions."""
from __future__ import annotations
import re
from pathlib import Path
from typing import TYPE_CHECKING, NamedTuple
from roboco.config import settings
if TYPE_CHECKING:
from uuid import UUID
_FRAME_NAME_RE = re.compile(r"^frame-(\d+)-of-(\d+)-at-([\d.]+)s\.png$")
class ParsedFrame(NamedTuple):
"""One preview frame parsed from its self-describing filename."""
frame_index: int
file: str
timestamp_seconds: float
def previews_root(task_id: UUID, project_slug: str) -> Path:
"""The container-shared preview-frames dir for a video-authoring task:
``{workspaces_root}/{project}/.previews/{task8}/``. Every agent container
mounts the same ``/data/workspaces``, so the CEO's container reads the
frames the dev's container rendered."""
return Path(settings.workspaces_root) / project_slug / ".previews" / task_id.hex[:8]
def list_orientation_frames(orientation_dir: Path) -> list[ParsedFrame]:
"""List the preview frames in ``orientation_dir``, parsed from the
self-describing ``frame-<idx>-of-<n>-at-<t>s.png`` filenames. Returns
``ParsedFrame`` namedtuples sorted by index. Returns an empty list when
the directory doesn't exist or holds no matching files."""
frames: list[ParsedFrame] = []
if not orientation_dir.is_dir():
return frames
for entry in sorted(orientation_dir.iterdir()):
m = _FRAME_NAME_RE.match(entry.name)
if m is None or not entry.is_file():
continue
frames.append(
ParsedFrame(
frame_index=int(m.group(1)),
file=entry.name,
timestamp_seconds=float(m.group(3)),
)
)
return frames
+235
View File
@@ -47,6 +47,8 @@ UX_DEV_1_UUID = _foundation.AGENTS["ux-dev-1"].uuid
UX_DEV_2_UUID = _foundation.AGENTS["ux-dev-2"].uuid
HISTORY_LIMIT = 2
RETRY_ATTEMPTS = 2
VERTICAL_FRAME_COUNT = 2
RENDER_DURATION_SECONDS = 6.4
async def _seed(session: AsyncSession) -> None:
@@ -1288,3 +1290,236 @@ async def test_preview_non_ceo_is_forbidden(db_session: AsyncSession) -> None:
resp = await client.get(f"/api/video/preview/{task.id}/vertical.html")
assert resp.status_code == HTTPStatus.FORBIDDEN
app.dependency_overrides.clear()
# --- preview-frames (request_render frame strip) ------------------------------
# CEO-gated routes serving the dev's already-rendered request_render preview
# frames. Frame listing is parsed from the self-describing
# .previews/{task8}/{orientation}/frame-<idx>-of-<n>-at-<t>s.png filenames.
def _write_frame(orientation_dir: Path, idx: int, total: int, ts: float) -> Path:
"""Write one self-describing frame PNG into ``orientation_dir``."""
orientation_dir.mkdir(parents=True, exist_ok=True)
path = orientation_dir / f"frame-{idx:02d}-of-{total}-at-{ts}s.png"
path.write_bytes(b"fake-png-bytes")
return path
async def _seed_video_task_with_project(
db_session: AsyncSession,
) -> tuple[TaskTable, str]:
"""Seed a source=video authoring task and return (task, project_slug)."""
task = await _seed_authoring_task(
db_session,
status=TaskStatus.AWAITING_CEO_APPROVAL,
draft_extra={"composition_id": "Intro"},
)
project = await db_session.get(ProjectTable, task.project_id)
assert project is not None and project.slug is not None
return task, project.slug
@pytest.mark.asyncio
async def test_preview_frames_lists_frames_for_task(
db_session: AsyncSession,
ceo_client: AsyncClient,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The CEO GET /preview-frames/{task_id} returns per-orientation frame
lists parsed from the self-describing filenames on disk."""
task, slug = await _seed_video_task_with_project(db_session)
monkeypatch.setattr(cfg, "workspaces_root", str(tmp_path))
root = tmp_path / slug / ".previews" / str(task.id).replace("-", "")[:8]
_write_frame(root / "vertical", 1, 2, 1.5)
_write_frame(root / "vertical", 2, 2, 4.5)
_write_frame(root / "square", 1, 1, 3.0)
resp = await ceo_client.get(f"/api/video/preview-frames/{task.id}")
assert resp.status_code == HTTPStatus.OK
body = resp.json()
assert body["task_id"] == str(task.id)
assert set(body["frames"].keys()) == {"vertical", "square"}
v = body["frames"]["vertical"]
assert len(v) == VERTICAL_FRAME_COUNT
assert v[0] == {
"index": 1,
"file": "frame-01-of-2-at-1.5s.png",
"timestamp_seconds": 1.5,
}
assert v[1] == {
"index": 2,
"file": "frame-02-of-2-at-4.5s.png",
"timestamp_seconds": 4.5,
}
s = body["frames"]["square"]
assert len(s) == 1
assert s[0] == {
"index": 1,
"file": "frame-01-of-1-at-3.0s.png",
"timestamp_seconds": 3.0,
}
@pytest.mark.asyncio
async def test_preview_frames_includes_render_metadata(
db_session: AsyncSession,
ceo_client: AsyncClient,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The response carries composition_id, duration_seconds, head_sha, dirty,
and rendered_at from the last render_preview marker stamp."""
task, slug = await _seed_video_task_with_project(db_session)
monkeypatch.setattr(cfg, "workspaces_root", str(tmp_path))
root = tmp_path / slug / ".previews" / str(task.id).replace("-", "")[:8]
_write_frame(root / "vertical", 1, 2, 1.5)
markers.set_render_preview(
task,
{
"at": "2026-07-19T12:00:00Z",
"composition_id": "Intro",
"orientation": "vertical",
"frame_count": 2,
"duration_seconds": 6.4,
"frames": [str(root / "vertical" / "frame-01-of-2-at-1.5s.png")],
"head_sha": "abc1234",
"dirty": False,
"rendered_by": "ux-dev-1",
"source": "worktree",
},
)
await db_session.flush()
resp = await ceo_client.get(f"/api/video/preview-frames/{task.id}")
body = resp.json()
assert body["composition_id"] == "Intro"
assert body["duration_seconds"] == RENDER_DURATION_SECONDS
assert body["head_sha"] == "abc1234"
assert body["dirty"] is False
assert body["rendered_at"] == "2026-07-19T12:00:00Z"
@pytest.mark.asyncio
async def test_preview_frames_missing_task_is_404(ceo_client: AsyncClient) -> None:
resp = await ceo_client.get(f"/api/video/preview-frames/{uuid4()}")
assert resp.status_code == HTTPStatus.NOT_FOUND
@pytest.mark.asyncio
async def test_preview_frames_non_video_task_is_404(
db_session: AsyncSession, ceo_client: AsyncClient
) -> None:
"""A video_post draft (not a video authoring task) is not a preview-frames
source — 404, not a frame listing."""
task = await _seed_draft(db_session) # source=video_post
resp = await ceo_client.get(f"/api/video/preview-frames/{task.id}")
assert resp.status_code == HTTPStatus.NOT_FOUND
@pytest.mark.asyncio
async def test_preview_frames_non_ceo_is_forbidden(db_session: AsyncSession) -> None:
task, _slug = await _seed_video_task_with_project(db_session)
app = _build_app(db_session, AgentRole.DEVELOPER, uuid4())
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
resp = await client.get(f"/api/video/preview-frames/{task.id}")
assert resp.status_code == HTTPStatus.FORBIDDEN
app.dependency_overrides.clear()
@pytest.mark.asyncio
async def test_preview_frames_empty_when_no_renders(
db_session: AsyncSession,
ceo_client: AsyncClient,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A video task with no rendered frames returns an empty frames dict —
not a 404, not an error. The panel shows a muted empty state."""
task, _slug = await _seed_video_task_with_project(db_session)
monkeypatch.setattr(cfg, "workspaces_root", str(tmp_path))
resp = await ceo_client.get(f"/api/video/preview-frames/{task.id}")
assert resp.status_code == HTTPStatus.OK
body = resp.json()
assert body["task_id"] == str(task.id)
assert body["frames"] == {}
@pytest.mark.asyncio
async def test_preview_frame_serves_png(
db_session: AsyncSession,
ceo_client: AsyncClient,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The per-frame route streams the PNG bytes for a specific frame."""
task, slug = await _seed_video_task_with_project(db_session)
monkeypatch.setattr(cfg, "workspaces_root", str(tmp_path))
root = tmp_path / slug / ".previews" / str(task.id).replace("-", "")[:8]
_write_frame(root / "vertical", 1, 2, 1.5)
resp = await ceo_client.get(
f"/api/video/preview-frames/{task.id}/vertical/frame-01-of-2-at-1.5s.png"
)
assert resp.status_code == HTTPStatus.OK
assert resp.headers["content-type"] == "image/png"
assert resp.content == b"fake-png-bytes"
@pytest.mark.asyncio
async def test_preview_frame_missing_file_is_404(
db_session: AsyncSession,
ceo_client: AsyncClient,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A non-existent frame file is a 404, not a crash."""
task, _slug = await _seed_video_task_with_project(db_session)
monkeypatch.setattr(cfg, "workspaces_root", str(tmp_path))
resp = await ceo_client.get(
f"/api/video/preview-frames/{task.id}/vertical/frame-99-of-99-at-9.9s.png"
)
assert resp.status_code == HTTPStatus.NOT_FOUND
@pytest.mark.asyncio
async def test_preview_frame_symlink_escape_is_404(
db_session: AsyncSession,
ceo_client: AsyncClient,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A symlink placed inside the orientation dir that points outside the
previews root must not serve the outside file — ``resolve_preview_path``
follows the link before the ``is_relative_to`` confinement check, so the
escape target is caught the same as a plain ``..`` traversal."""
task, slug = await _seed_video_task_with_project(db_session)
monkeypatch.setattr(cfg, "workspaces_root", str(tmp_path))
root = (
tmp_path / slug / ".previews" / str(task.id).replace("-", "")[:8] / "vertical"
)
root.mkdir(parents=True)
outside = tmp_path / "secret.png"
outside.write_bytes(b"secret")
link = root / "frame-01-of-1-at-0.0s.png"
link.symlink_to(outside)
resp = await ceo_client.get(
f"/api/video/preview-frames/{task.id}/vertical/frame-01-of-1-at-0.0s.png"
)
assert resp.status_code == HTTPStatus.NOT_FOUND
@pytest.mark.asyncio
async def test_preview_frame_non_ceo_is_forbidden(db_session: AsyncSession) -> None:
task, _slug = await _seed_video_task_with_project(db_session)
app = _build_app(db_session, AgentRole.DEVELOPER, uuid4())
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
resp = await client.get(
f"/api/video/preview-frames/{task.id}/vertical/frame-01-of-2-at-1.5s.png"
)
assert resp.status_code == HTTPStatus.FORBIDDEN
app.dependency_overrides.clear()