tests/test_keep_files.py (3744 bytes)
1 """The server keeps each file of a job: the audio, the book and the subtitles, 2 together in the folder of the job. Nothing deletes them automatically. The 3 operator deletes them by hand.""" 4 5 from __future__ import annotations 6 7 from datetime import timedelta 8 9 from backend import main, runner 10 from backend.db import Job, JobStatus, SessionLocal, utcnow 11 from backend.runner import Paths 12 from backend.settings import settings 13 14 from .conftest import get_job_row, make_job 15 16 SRT = "1\n00:00:00,000 --> 00:00:01,000\nhello\n\n" 17 18 19 def stage_inputs(job_id: str) -> tuple[Paths, list]: 20 """Put an audio file and a book where an upload puts them.""" 21 paths = Paths.for_job(job_id) 22 paths.create() 23 audio = runner.staged_audio_path(job_id, "book.m4b") 24 book = runner.staged_text_path(job_id, "book.epub") 25 audio.write_bytes(b"audio") 26 book.write_bytes(b"book") 27 return paths, [audio, book] 28 29 30 def age(job_id: str, **delta) -> None: 31 with SessionLocal() as s: 32 job = s.get(Job, job_id) 33 job.created_at = utcnow() - timedelta(**delta) 34 job.finished_at = utcnow() - timedelta(**delta) 35 s.commit() 36 37 38 def test_a_finished_server_job_keeps_audio_book_and_subtitles_together(client, monkeypatch): 39 job_id = make_job(client, JobStatus.queued) 40 paths, inputs = stage_inputs(job_id) 41 produced = paths.out / "source.en.srt" 42 43 def fake_aligner(job_id, request, log_path): 44 # Bytes, not text: on Windows write_text writes CRLF line ends. 45 produced.write_bytes(SRT.encode("utf-8")) 46 return 0 47 48 monkeypatch.setattr(runner, "probe_duration", lambda path: 1.0) 49 monkeypatch.setattr(runner, "probe_chapters", lambda path: 0) 50 monkeypatch.setattr(runner, "normalize_for_alignment", lambda job_id, source: source) 51 monkeypatch.setattr(runner, "_stream_aligner", fake_aligner) 52 monkeypatch.setattr(runner.aligner, "locate_output", lambda request: produced) 53 monkeypatch.setattr(settings, "render_video", False) 54 55 runner.run_job(job_id) 56 57 assert get_job_row(job_id).status == JobStatus.succeeded 58 for path in [*inputs, produced]: 59 assert path.exists(), path 60 assert paths.root in path.parents # one folder for the job 61 assert client.get(f"/api/jobs/{job_id}/files/srt").text == SRT 62 63 64 def test_housekeeping_deletes_no_file_and_no_job(client): 65 failed = make_job(client, JobStatus.failed) 66 draft = make_job(client) 67 done = make_job(client, JobStatus.succeeded, with_files=True) 68 staged = {job_id: stage_inputs(job_id)[1] for job_id in (failed, draft, done)} 69 for job_id in staged: 70 age(job_id, days=400) 71 72 main.housekeeping() 73 74 for job_id, inputs in staged.items(): 75 assert get_job_row(job_id) is not None 76 assert all(path.exists() for path in inputs), job_id 77 assert client.get(f"/api/jobs/{done}/files/srt").status_code == 200 78 79 80 def test_deleting_a_job_keeps_its_files(client): 81 job_id = make_job(client) 82 _, inputs = stage_inputs(job_id) 83 assert client.delete(f"/api/jobs/{job_id}").status_code == 200 84 assert get_job_row(job_id) is None 85 assert all(path.exists() for path in inputs) 86 87 88 def test_a_refused_upload_keeps_what_arrived(client, monkeypatch): 89 monkeypatch.setattr(settings, "max_upload_bytes", 10) 90 work = settings.data_dir / "work" 91 before = set(work.iterdir()) if work.exists() else set() 92 93 r = client.post("/api/uploads", files=[ 94 ("files", ("book.epub", b"epub", "application/epub+zip")), 95 ("files", ("book.m4b", b"x" * 100, "audio/mp4")), 96 ]) 97 assert r.status_code == 413 98 99 new = set(work.iterdir()) - before 100 assert len(new) == 1 101 kept = {p.suffix for p in new.pop().rglob("*") if p.is_file()} 102 assert kept == {".epub", ".m4b"}