tests/test_local_jobs.py (4342 bytes)
1 """Jobs that run in the visitor's browser: the server only keeps the books.""" 2 3 from __future__ import annotations 4 5 import time 6 from datetime import timedelta 7 8 from backend import api 9 from backend.db import Job, JobStatus, SessionLocal, utcnow 10 11 from .conftest import account_id, checkout_event, get_job_row, post_webhook 12 13 BOOK = { 14 "audio_filename": "book.m4b", "audio_bytes": 123456, "audio_duration_seconds": 3600.0, 15 "text_filename": "book.epub", "language": "ja", 16 } 17 SRT = "1\n00:00:00,000 --> 00:00:01,000\nこんにちは\n\n" 18 19 20 def begin(client, **over): 21 return client.post("/api/local/jobs", json={**BOOK, **over}) 22 23 24 def buy(client, plan="single", email="local@example.com"): 25 sid = f"cs_{time.time_ns()}" 26 assert post_webhook(client, checkout_event(sid, account_id(client), plan, email=email)).status_code == 200 27 28 29 def test_free_job_runs_finishes_and_keeps_its_subtitles(client): 30 r = begin(client) 31 assert r.status_code == 200 32 job = r.json() 33 assert job["local"] is True and job["status"] == "running" and job["tier"] == "free" 34 35 done = client.post(f"/api/local/jobs/{job['id']}/finish", 36 json={"srt": SRT, "filename": "book.ja.srt", "metadata": {"cues": 1}}) 37 assert done.status_code == 200 and done.json()["status"] == "succeeded" 38 kinds = {a["kind"] for a in done.json()["artifacts"]} 39 assert kinds == {"srt", "metadata"} 40 41 got = client.get(f"/api/jobs/{job['id']}/files/srt") 42 assert got.status_code == 200 and got.content.decode("utf-8") == SRT 43 assert job["id"] in {j["id"] for j in client.get("/api/jobs").json()} 44 45 46 def test_browser_jobs_have_no_limit_and_no_price(client): 47 buy(client, email="nolimit@example.com") 48 for n in range(4): 49 r = begin(client, audio_filename=f"book{n}.m4b", audio_bytes=1000 + n) 50 assert r.status_code == 200 and r.json()["tier"] == "free" 51 # The work was done on the visitor's machine: the credit is still there. 52 assert client.get("/api/account").json()["credits"] == 1 53 54 55 def test_reopening_the_same_book_does_not_charge_twice(client): 56 first = begin(client).json() 57 again = begin(client) # closed tab, same files 58 assert again.status_code == 200 and again.json()["id"] == first["id"] 59 with SessionLocal() as s: 60 assert s.query(Job).filter(Job.account_id == account_id(client)).count() == 1 61 62 63 def test_a_failed_browser_job_takes_nothing(client): 64 buy(client, email="failed@example.com") # its own account: an email joins accounts across tests 65 r = begin(client) 66 failed = client.post(f"/api/local/jobs/{r.json()['id']}/fail", json={"error": "GPU lost"}) 67 assert failed.status_code == 200 and failed.json()["status"] == "failed" 68 assert client.get("/api/account").json()["credits"] == 1 69 70 71 def test_finish_is_once_only_and_owner_only(client, second_client): 72 job = begin(client).json() 73 body = {"srt": SRT, "filename": "x.srt"} 74 assert second_client.post(f"/api/local/jobs/{job['id']}/finish", json=body).status_code == 404 75 assert client.post(f"/api/local/jobs/{job['id']}/finish", json=body).status_code == 200 76 assert client.post(f"/api/local/jobs/{job['id']}/finish", json=body).status_code == 409 77 assert client.post(f"/api/local/jobs/{job['id']}/fail", json={}).status_code == 409 78 79 80 def test_filename_cannot_escape_the_job_directory(client): 81 job = begin(client).json() 82 r = client.post(f"/api/local/jobs/{job['id']}/finish", 83 json={"srt": SRT, "filename": "../../../etc/passwd"}) 84 assert r.status_code == 200 85 srt = next(a for a in r.json()["artifacts"] if a["kind"] == "srt") 86 assert srt["filename"] == "passwd" 87 88 89 def test_abandoned_jobs_release_what_they_held(client): 90 job = begin(client).json() 91 with SessionLocal() as s: 92 row = s.get(Job, job["id"]) 93 row.created_at = utcnow() - timedelta(hours=72) 94 s.commit() 95 assert api.expire_stale_local_jobs(s) >= 1 96 assert get_job_row(job["id"]).status == JobStatus.canceled 97 98 99 def test_restart_does_not_queue_browser_jobs(client, monkeypatch): 100 from backend import main 101 102 job = begin(client).json() 103 queued = [] 104 monkeypatch.setattr(main.queue, "enqueue", queued.append) 105 main._requeue_interrupted() 106 assert job["id"] not in queued 107 assert get_job_row(job["id"]).status == JobStatus.running