main_backend/tests/unit/test_project_workspace_service.py

from __future__ import annotations

import uuid
from types import SimpleNamespace

import pytest

from cpv3.modules.project_workspaces.schemas import (
    ProjectWorkspaceState,
    build_workspace_state_from_legacy,
)
from cpv3.modules.project_workspaces.service import ProjectWorkspaceService


def test_build_workspace_state_from_legacy_maps_known_fields() -> None:
    source_file_id = uuid.uuid4()
    active_job_id = uuid.uuid4()
    silence_job_id = uuid.uuid4()
    artifact_id = uuid.uuid4()

    workspace = build_workspace_state_from_legacy(
        {
            "wizard": {
                "current_step": "subtitle-revision",
                "primary_file_id": str(source_file_id),
                "active_job_id": str(active_job_id),
                "active_job_type": "TRANSCRIPTION_GENERATE",
                "silence_job_id": str(silence_job_id),
                "transcription_artifact_id": str(artifact_id),
                "silence_settings": {
                    "min_silence_duration_ms": 350,
                    "silence_threshold_db": 21,
                    "padding_ms": 180,
                },
                "unknown_field": "ignored",
            },
            "used_files": [
                {"id": str(source_file_id), "path": "users/test/source.mp4"},
                {"id": "not-a-uuid", "path": "broken"},
            ],
            "unknown_root": {"ignored": True},
        }
    )

    assert workspace.phase == "TRANSCRIPTION"
    assert workspace.source_file_id == source_file_id
    assert workspace.active_job is not None
    assert workspace.active_job.job_id == active_job_id
    assert workspace.active_job.job_type == "TRANSCRIPTION_GENERATE"
    assert workspace.workspace_view.used_file_ids == [source_file_id]
    assert workspace.workspace_view.selected_file_id == source_file_id
    assert workspace.silence.detect_job_id == silence_job_id
    assert workspace.silence.settings.min_silence_duration_ms == 350
    assert workspace.silence.settings.silence_threshold_db == 21
    assert workspace.silence.settings.padding_ms == 180
    assert workspace.transcription.artifact_id == artifact_id
    assert workspace.captions.output_file_id is None


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("job_type", "output_data", "initial_state", "expected"),
    [
        (
            "MEDIA_CONVERT",
            {"file_id": "00000000-0000-4000-a000-000000000101"},
            {
                "phase": "VERIFY",
                "source_file_id": "00000000-0000-4000-a000-000000000001",
                "active_job": {
                    "job_id": "00000000-0000-4000-a000-000000000010",
                    "job_type": "MEDIA_CONVERT",
                },
            },
            {
                "phase": "VERIFY",
                "source_file_id": "00000000-0000-4000-a000-000000000101",
                "active_job": None,
                "current_screen": "verify",
            },
        ),
        (
            "SILENCE_DETECT",
            {
                "silent_segments": [{"start_ms": 100, "end_ms": 220}],
                "duration_ms": 1000,
            },
            {
                "phase": "SILENCE",
                "source_file_id": "00000000-0000-4000-a000-000000000001",
                "active_job": {
                    "job_id": "00000000-0000-4000-a000-000000000011",
                    "job_type": "SILENCE_DETECT",
                },
                "silence": {"status": "DETECTING"},
            },
            {
                "phase": "SILENCE",
                "active_job": None,
                "silence_status": "REVIEWING",
                "current_screen": "fragments",
            },
        ),
        (
            "SILENCE_APPLY",
            {"file_id": "00000000-0000-4000-a000-000000000102"},
            {
                "phase": "SILENCE",
                "source_file_id": "00000000-0000-4000-a000-000000000001",
                "active_job": {
                    "job_id": "00000000-0000-4000-a000-000000000012",
                    "job_type": "SILENCE_APPLY",
                },
                "silence": {"status": "APPLYING"},
            },
            {
                "phase": "TRANSCRIPTION",
                "active_job": None,
                "silence_status": "COMPLETED",
                "current_screen": "transcription-settings",
            },
        ),
        (
            "TRANSCRIPTION_GENERATE",
            {
                "artifact_id": "00000000-0000-4000-a000-000000000103",
                "transcription_id": "00000000-0000-4000-a000-000000000104",
            },
            {
                "phase": "TRANSCRIPTION",
                "source_file_id": "00000000-0000-4000-a000-000000000001",
                "active_job": {
                    "job_id": "00000000-0000-4000-a000-000000000013",
                    "job_type": "TRANSCRIPTION_GENERATE",
                },
                "transcription": {"status": "PROCESSING"},
            },
            {
                "phase": "TRANSCRIPTION",
                "active_job": None,
                "transcription_status": "REVIEWING",
                "current_screen": "subtitle-revision",
            },
        ),
        (
            "CAPTIONS_GENERATE",
            {"file_id": "00000000-0000-4000-a000-000000000105"},
            {
                "phase": "CAPTIONS",
                "source_file_id": "00000000-0000-4000-a000-000000000001",
                "active_job": {
                    "job_id": "00000000-0000-4000-a000-000000000014",
                    "job_type": "CAPTIONS_GENERATE",
                },
                "captions": {"status": "PROCESSING"},
            },
            {
                "phase": "DONE",
                "active_job": None,
                "captions_status": "COMPLETED",
                "current_screen": "caption-result",
            },
        ),
    ],
)
async def test_apply_job_event_advances_workspace_for_done_jobs(
    job_type: str,
    output_data: dict[str, object],
    initial_state: dict[str, object],
    expected: dict[str, object],
) -> None:
    service = ProjectWorkspaceService(session=SimpleNamespace())

    state = ProjectWorkspaceState.model_validate(
        {
            "version": 1,
            "phase": "INGEST",
            "active_job": None,
            "source_file_id": None,
            "workspace_view": {"used_file_ids": [], "selected_file_id": None},
            "silence": {},
            "transcription": {},
            "captions": {},
            **initial_state,
        }
    )

    job = SimpleNamespace(
        id=uuid.UUID(str(state.active_job.job_id)) if state.active_job else uuid.uuid4(),
        project_id=uuid.uuid4(),
        job_type=job_type,
        status="DONE",
        output_data=output_data,
    )

    next_state = service._apply_job_event_to_state(state, job)
    current_screen = service._derive_current_screen(next_state)

    assert next_state.phase == expected["phase"]
    assert next_state.active_job == expected["active_job"]
    assert current_screen == expected["current_screen"]

    if "source_file_id" in expected:
        assert str(next_state.source_file_id) == expected["source_file_id"]
    if "silence_status" in expected:
        assert next_state.silence.status == expected["silence_status"]
    if "transcription_status" in expected:
        assert next_state.transcription.status == expected["transcription_status"]
    if "captions_status" in expected:
        assert next_state.captions.status == expected["captions_status"]