tests/test_content_studio_phase2.py

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