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"]