from __future__ import annotations from copy import deepcopy from pathlib import Path from uuid import uuid4 import pytest from sqlalchemy import select from app.container import build_container from app.core.config import Settings from app.projects.editor_schemas import EditorDocument, EditorSaveRequest, ProjectRenderCreate from app.projects.errors import ( ProjectEditorConflictError, ProjectNotFoundError, ProjectRenderLimitError, ) from app.projects.schemas import ProjectCreate from app.projects.services.render_compiler import compile_render from app.security.models import AuditEvent from app.security.schemas import APIKeyCreate def settings(tmp_path: Path) -> Settings: return Settings( _env_file=None, auth_enabled=True, database_url=f"sqlite+aiosqlite:///{tmp_path / 'security.db'}", social_database_url=f"sqlite+aiosqlite:///{tmp_path / 'social.db'}", social_auto_migrate=True, social_worker_enabled=False, generation_worker_enabled=False, render_worker_enabled=False, social_oauth_encryption_key="test-only-encryption-material", temp_dir=tmp_path / "temp", output_dir=tmp_path / "outputs", whisper_model="tiny", ) def document(project_id: str, asset_id: str) -> EditorDocument: return EditorDocument.model_validate( { "schemaVersion": 1, "projectId": project_id, "timeline": { "timeUnit": "milliseconds", "tracks": [ { "id": "video-1", "type": "video", "name": "Video 1", "order": 0, "muted": False, "locked": False, "visible": True, "clips": [ { "id": "clip-1", "kind": "media", "trackId": "video-1", "assetId": asset_id, "label": "source.mp4", "startMs": 0, "durationMs": 1000, "sourceStartMs": 0, "sourceDurationMs": 1000, "mediaType": "video", "transform": { "x": 0, "y": 0, "scaleX": 1, "scaleY": 1, "rotation": 0, }, "volume": 1, "opacity": 1, "visible": True, "metadata": {}, } ], } ], "transitions": [], "markers": [], }, "renderSettings": {"format": "mp4", "width": 1280, "height": 720, "frameRate": 30}, } ) async def actor(container, name: str): key, secret = await container.api_keys.create( APIKeyCreate( name=name, environment="test", role=None, scopes=[ "projects:read", "projects:create", "projects:update", "jobs:create", "jobs:cancel", ], ), created_by="tests", ) return key, await container.api_keys.authenticate(secret) @pytest.mark.asyncio async def test_editor_revision_isolation_render_idempotency_and_cancellation( tmp_path: Path, ) -> None: container = build_container(settings(tmp_path)) await container.security_database.initialize() try: key_a, actor_a = await actor(container, "A") _, actor_b = await actor(container, "B") project = await container.projects.create( workspace_id=actor_a.workspace_id, user_id=actor_a.user_id, api_key_id=key_a.id, request_id=str(uuid4()), payload=ProjectCreate(name="Studio"), ) request_id = str(uuid4()) output = container.settings.output_dir / request_id output.mkdir(parents=True) source = output / "source.mp4" source.write_bytes(b"test media") asset = await container.assets.register_output( workspace_id=actor_a.workspace_id, user_id=actor_a.user_id, request_id=request_id, path=source, mime_type="video/mp4", project_id=project.id, ) editor_document = document(project.id, asset.id) saved = await container.editor.save( workspace_id=actor_a.workspace_id, user_id=actor_a.user_id, api_key_id=key_a.id, request_id=str(uuid4()), project_id=project.id, payload=EditorSaveRequest(expected_revision=0, schema_version=1, state=editor_document), ) assert saved.revision == 1 with pytest.raises(ProjectEditorConflictError): await container.editor.save( workspace_id=actor_a.workspace_id, user_id=actor_a.user_id, api_key_id=key_a.id, request_id=str(uuid4()), project_id=project.id, payload=EditorSaveRequest( expected_revision=0, schema_version=1, state=editor_document ), ) with pytest.raises(ProjectNotFoundError): await container.editor.get( workspace_id=actor_b.workspace_id, user_id=actor_b.user_id, project_id=project.id, ) render_payload = ProjectRenderCreate( editor_revision=1, output_format="mp4", width=1280, height=720 ) first = await container.renders.create( workspace_id=actor_a.workspace_id, user_id=actor_a.user_id, api_key_id=key_a.id, request_id=str(uuid4()), project_id=project.id, payload=render_payload, idempotency_key="render-1", ) second = await container.renders.create( workspace_id=actor_a.workspace_id, user_id=actor_a.user_id, api_key_id=key_a.id, request_id=str(uuid4()), project_id=project.id, payload=render_payload, idempotency_key="render-1", ) assert first.id == second.id and first.status == "queued" with pytest.raises(ProjectRenderLimitError): await container.renders.create( workspace_id=actor_a.workspace_id, user_id=actor_a.user_id, api_key_id=key_a.id, request_id=str(uuid4()), project_id=project.id, payload=render_payload, idempotency_key="render-2", ) cancelled = await container.renders.cancel( workspace_id=actor_a.workspace_id, user_id=actor_a.user_id, api_key_id=key_a.id, request_id=str(uuid4()), project_id=project.id, render_id=first.id, ) assert cancelled.status == "cancelled" repeated = await container.renders.cancel( workspace_id=actor_a.workspace_id, user_id=actor_a.user_id, api_key_id=key_a.id, request_id=str(uuid4()), project_id=project.id, render_id=first.id, ) assert repeated.status == "cancelled" async with container.security_database.tenant_session( workspace_id=actor_a.workspace_id, user_id=actor_a.user_id, ) as session: cancellation_events = list( ( await session.scalars( select(AuditEvent).where( AuditEvent.entity_id == first.id, AuditEvent.event_type == "project.render_cancelled", ) ) ).all() ) assert len(cancellation_events) == 1 finally: await container.security_database.close() def test_render_compiler_is_deterministic_and_uses_server_paths(tmp_path: Path) -> None: source = tmp_path / "source.mp4" source.write_bytes(b"media") state = document(str(uuid4()), str(uuid4())) asset_id = next(iter(state.asset_ids())) first = compile_render( state, asset_paths={asset_id: (source, "video/mp4")}, width=1280, height=720, frame_rate=30, output_format="mp4", quality="standard", preset="balanced", ) second = compile_render( state, asset_paths={asset_id: (source, "video/mp4")}, width=1280, height=720, frame_rate=30, output_format="mp4", quality="standard", preset="balanced", ) assert first == second assert source in first.args assert first.duration_ms == 1000 assert "yuv420p" in first.args def test_render_compiler_ignores_hidden_timeline_tail(tmp_path: Path) -> None: source = tmp_path / "source.mp4" source.write_bytes(b"media") state = document(str(uuid4()), str(uuid4())) payload = state.model_dump(by_alias=True) hidden_track = deepcopy(payload["timeline"]["tracks"][0]) hidden_track.update({"id": "video-hidden", "name": "Hidden", "order": 1, "visible": False}) hidden_track["clips"][0].update( {"id": "clip-hidden", "trackId": "video-hidden", "startMs": 120_000} ) payload["timeline"]["tracks"].append(hidden_track) state_with_hidden_tail = EditorDocument.model_validate(payload) asset_id = next(iter(state_with_hidden_tail.asset_ids())) plan = compile_render( state_with_hidden_tail, asset_paths={asset_id: (source, "video/mp4")}, width=1280, height=720, frame_rate=30, output_format="webm", quality="high", preset="quality", ) assert plan.duration_ms == 1000 assert plan.args.count(source) == 1 assert "18" in plan.args assert "0" in plan.args