You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
113 lines
3.9 KiB
113 lines
3.9 KiB
import json |
|
from pathlib import Path |
|
|
|
import pytest |
|
|
|
from voice_transcriptor.services.job_manifest import JobManifestError, JobManifestRepository |
|
|
|
|
|
def chunks() -> list[dict]: |
|
return [ |
|
{ |
|
"index": 0, |
|
"path": "chunks/chunk-00000.m4a", |
|
"source_start_seconds": "0", |
|
"source_end_seconds": "900.125", |
|
"duration_seconds": "900.125", |
|
}, |
|
{ |
|
"index": 1, |
|
"path": "chunks/chunk-00001.m4a", |
|
"source_start_seconds": "885.125", |
|
"source_end_seconds": "1000.5", |
|
"duration_seconds": "115.375", |
|
}, |
|
] |
|
|
|
|
|
def create_manifest(tmp_path: Path) -> tuple[JobManifestRepository, Path]: |
|
job = tmp_path / "job" |
|
repository = JobManifestRepository() |
|
manifest = repository.create( |
|
job, |
|
source={"path": str(tmp_path / "source.mp3"), "size_bytes": 10}, |
|
settings={"model": "gpt-transcribe", "language": "pt-BR"}, |
|
chunks=chunks(), |
|
) |
|
return repository, Path(manifest["manifest_path"]) |
|
|
|
|
|
def test_create_persists_schema_timing_and_pending_statuses(tmp_path: Path) -> None: |
|
repository, path = create_manifest(tmp_path) |
|
|
|
manifest = repository.load(path) |
|
|
|
assert manifest["schema_version"] == 2 |
|
assert manifest["state"] == "preprocessing" |
|
assert manifest["completed_chunks"] == 0 |
|
assert manifest["total_chunks"] == 2 |
|
assert [item["status"] for item in manifest["chunks"]] == ["pending", "pending"] |
|
assert manifest["chunks"][1]["source_start_seconds"] == "885.125" |
|
assert not path.with_suffix(".tmp").exists() |
|
|
|
|
|
def test_create_rejects_chunk_path_outside_job(tmp_path: Path) -> None: |
|
invalid = chunks() |
|
invalid[0]["path"] = "../secret.m4a" |
|
|
|
with pytest.raises(JobManifestError, match="chunk path"): |
|
JobManifestRepository().create(tmp_path / "job", {}, {}, invalid) |
|
|
|
|
|
def test_completed_chunk_is_written_immediately_and_cannot_be_reprocessed(tmp_path: Path) -> None: |
|
repository, path = create_manifest(tmp_path) |
|
|
|
repository.mark_processing(path, 0) |
|
repository.mark_completed(path, 0, "Primeiro trecho") |
|
on_disk = json.loads(path.read_text(encoding="utf-8")) |
|
|
|
assert on_disk["chunks"][0]["status"] == "completed" |
|
assert on_disk["chunks"][0]["transcript"] == "Primeiro trecho" |
|
assert on_disk["completed_chunks"] == 1 |
|
with pytest.raises(JobManifestError, match="completed"): |
|
repository.mark_processing(path, 0) |
|
|
|
|
|
def test_resume_recovers_processing_and_failed_but_preserves_completed(tmp_path: Path) -> None: |
|
repository, path = create_manifest(tmp_path) |
|
repository.mark_processing(path, 0) |
|
repository.mark_completed(path, 0, "Feito") |
|
repository.mark_processing(path, 1) |
|
repository.mark_failed(path, 1, "temporary failure") |
|
|
|
manifest = repository.recover_for_resume(path) |
|
|
|
assert manifest["chunks"][0]["status"] == "completed" |
|
assert manifest["chunks"][0]["transcript"] == "Feito" |
|
assert manifest["chunks"][1]["status"] == "pending" |
|
assert manifest["chunks"][1]["last_error"] is None |
|
|
|
|
|
def test_assemble_transcript_uses_only_completed_chunks_in_index_order(tmp_path: Path) -> None: |
|
repository, path = create_manifest(tmp_path) |
|
repository.mark_processing(path, 1) |
|
repository.mark_completed(path, 1, "Segundo") |
|
repository.mark_processing(path, 0) |
|
repository.mark_completed(path, 0, "Primeiro") |
|
|
|
transcript_path = repository.assemble_transcript(path) |
|
|
|
assert transcript_path.read_text(encoding="utf-8") == "Primeiro\nSegundo\n" |
|
assert repository.load(path)["chunks"][1]["source_end_seconds"] == "1000.5" |
|
|
|
|
|
def test_explicit_reset_is_required_to_clear_completed_work(tmp_path: Path) -> None: |
|
repository, path = create_manifest(tmp_path) |
|
repository.mark_processing(path, 0) |
|
repository.mark_completed(path, 0, "Feito") |
|
|
|
repository.reset_completed(path, indexes=[0]) |
|
|
|
item = repository.load(path)["chunks"][0] |
|
assert item["status"] == "pending" |
|
assert item["transcript"] is None
|
|
|