tests/test_retention.py (2605 bytes)
1 """Uploaded books and results do not stay on the server past their time.""" 2 3 from __future__ import annotations 4 5 from datetime import timedelta 6 7 from backend import retention 8 from backend.db import Artifact, Job, JobStatus, SessionLocal, utcnow 9 from backend.runner import Paths 10 from backend.storage import storage 11 12 from .conftest import get_job_row, make_job 13 14 15 def age(job_id: str, **delta) -> None: 16 with SessionLocal() as s: 17 job = s.get(Job, job_id) 18 job.created_at = utcnow() - timedelta(**delta) 19 job.finished_at = utcnow() - timedelta(**delta) 20 s.commit() 21 22 23 def stage_inputs(job_id: str): 24 paths = Paths.for_job(job_id) 25 paths.create() 26 (paths.inp / "book.m4b").write_bytes(b"audio") 27 return paths 28 29 30 def sweep() -> dict: 31 with SessionLocal() as s: 32 return retention.sweep(s) 33 34 35 def test_a_failed_job_keeps_its_upload_for_a_day_and_no_longer(client): 36 job_id = make_job(client, JobStatus.failed) 37 paths = stage_inputs(job_id) 38 39 age(job_id, hours=23) 40 sweep() 41 assert (paths.inp / "book.m4b").exists() # still there for another try 42 43 age(job_id, hours=25) 44 assert sweep()["inputs"] == 1 45 assert not paths.root.exists() 46 assert get_job_row(job_id).status == JobStatus.failed # the list still shows it 47 48 49 def test_an_upload_that_was_never_started_goes_away_with_its_job(client): 50 job_id = make_job(client) # a draft 51 paths = stage_inputs(job_id) 52 age(job_id, hours=25) 53 assert sweep()["drafts"] == 1 54 assert not paths.root.exists() and get_job_row(job_id) is None 55 56 57 def test_results_are_deleted_after_a_week_and_the_job_says_so(client): 58 fresh = make_job(client, JobStatus.succeeded, with_files=True) 59 old = make_job(client, JobStatus.succeeded, with_files=True) 60 age(fresh, days=6) 61 age(old, days=8) 62 63 assert sweep()["results"] == 1 64 assert client.get(f"/api/jobs/{fresh}/files/srt").status_code == 200 65 assert client.get(f"/api/jobs/{old}/files/srt").status_code == 404 66 job = client.get(f"/api/jobs/{old}").json() 67 assert job["artifacts"] == [] and "deleted after 7 days" in job["stage"] 68 with SessionLocal() as s: 69 assert s.query(Artifact).filter(Artifact.job_id == old).count() == 0 70 assert not storage.exists(f"{old}/book.en.srt") 71 72 assert sweep() == {"inputs": 0, "drafts": 0, "results": 0} # and once is enough 73 74 75 def test_running_jobs_are_left_alone(client): 76 job_id = make_job(client, JobStatus.running) 77 paths = stage_inputs(job_id) 78 age(job_id, days=30) 79 sweep() 80 assert (paths.inp / "book.m4b").exists()