"""The job queue: shared jobs, order, cancelling and the job history.""" import json from datetime import datetime, timedelta from bindery import config from bindery.jobs import DownloadMode, Job, JobQueue, QueueStatus def job(mode=DownloadMode.NEW, **fields): return Job(mode=mode, **fields) def test_a_waiting_job_is_shared(): queue = JobQueue() first = queue.add(job(story_id=1)) assert queue.add(job(story_id=1)) is first assert queue.add(job(url="https://example.com/s/1")) is not first assert queue.add(job(url="https://example.com/s/1")).url == "https://example.com/s/1" assert queue.add(job(DownloadMode.REBUILD, story_id=1)) is not first assert queue.add(job(story_id=2)) is not first assert queue.add(job(story_id=1, force_fields=["title"])) is not first def test_rechecks_of_other_chapters_are_separate_jobs(): queue = JobQueue() one = queue.add(job(DownloadMode.RECHECK, story_id=1, chapter_ids=[1])) assert queue.add(job(DownloadMode.RECHECK, story_id=1, chapter_ids=[2])) is not one assert queue.add(job(DownloadMode.RECHECK, story_id=1, chapter_ids=[1])) is one def test_uploads_are_separate_jobs(): queue = JobQueue() one = queue.add(job(DownloadMode.IMPORT_FILE, story_id=1, options={"path": "a.epub"})) assert queue.add(job(DownloadMode.IMPORT_FILE, story_id=1, options={"path": "b.epub"})) is not one def test_a_started_job_is_not_shared(): queue = JobQueue() first = queue.add(job(story_id=1)) assert queue.get_next() is first assert queue.add(job(story_id=1)) is not first def test_jobs_run_in_order_and_cancelled_jobs_are_skipped(): queue = JobQueue() a, b, c = (queue.add(job(story_id=n)) for n in (1, 2, 3)) assert queue.cancel(b.id) assert b.status == QueueStatus.CANCELLED assert queue.get_next() is a assert queue.get_next() is c assert queue.get_next() is None def test_a_cancelled_job_is_not_shared(): queue = JobQueue() cancelled = queue.add(job(story_id=1)) queue.cancel(cancelled.id) again = queue.add(job(story_id=1)) assert again is not cancelled and again.status == QueueStatus.QUEUED assert queue.get_next() is again def test_a_cancelled_job_can_expire_before_the_queue_reaches_it(): queue = JobQueue() queue.add(job(story_id=1)) running = queue.get_next() cancelled = queue.add(job(story_id=2)) queue.cancel(cancelled.id) cancelled.updated = datetime.now() - timedelta(seconds=config.JOB_HISTORY + 1) assert cancelled.id not in [j["id"] for j in queue.get_status()] queue.update(running, status=QueueStatus.DONE) later = queue.add(job(story_id=3)) assert queue.get_next() is later def test_only_waiting_jobs_can_be_cancelled(): queue = JobQueue() running = queue.add(job(story_id=1)) queue.update(queue.get_next(), status=QueueStatus.RUNNING) assert not queue.cancel(running.id) assert not queue.cancel("no-such-job") def test_has_active(): queue = JobQueue() first = queue.add(job(story_id=1)) assert queue.has_active(1) and not queue.has_active(2) queue.update(first, status=QueueStatus.RUNNING) assert queue.has_active(1) queue.update(first, status=QueueStatus.DONE) assert not queue.has_active(1) def test_status_is_newest_first_and_filters_by_story(): queue = JobQueue() old = queue.add(job(story_id=1, created=datetime.now() - timedelta(minutes=5))) new = queue.add(job(DownloadMode.REBUILD, story_id=1)) other = queue.add(job(story_id=2)) assert [j["id"] for j in queue.get_status(1)] == [new.id, old.id] assert {j["id"] for j in queue.get_status()} == {old.id, new.id, other.id} data = queue.get_status(2)[0] assert data["mode"] == "new" and data["status"] == "queued" assert data["mode_label"] == "Download new chapters (default)" assert datetime.fromisoformat(data["created"]) def test_finished_jobs_expire(): queue = JobQueue() done = queue.add(job(story_id=1)) waiting = queue.add(job(story_id=2)) long_ago = datetime.now() - timedelta(seconds=config.JOB_HISTORY + 1) queue.update(done, status=QueueStatus.DONE) done.updated = long_ago waiting.updated = long_ago assert [j["id"] for j in queue.get_status()] == [waiting.id] def test_jobs_outlive_a_restart(tmp_path): queue = JobQueue(tmp_path / "db.sqlite") running = queue.add(job(story_id=1)) waiting = queue.add(job(story_id=2, options={"x": [1]})) done = queue.add(job(story_id=3)) cancelled = queue.add(job(story_id=4)) queue.update(queue.get_next(), status=QueueStatus.RUNNING, result={"new": 2}) queue.update(done, status=QueueStatus.DONE) queue.cancel(cancelled.id) again = JobQueue(tmp_path / "db.sqlite") status = {j["id"]: j for j in again.get_status()} assert (status[running.id]["status"], status[running.id]["restarts"]) == ("queued", 1) assert status[running.id]["message"] == "Resumed" assert status[running.id]["result"] == {"new": 2} assert status[done.id]["status"] == "done" and status[cancelled.id]["status"] == "cancelled" # The interrupted job goes first. assert again.get_next().id == running.id nxt = again.get_next() assert (nxt.id, nxt.options, nxt.mode) == (waiting.id, {"x": [1]}, DownloadMode.NEW) assert again.get_next() is None def test_a_job_that_stops_bindery_again_and_again_is_given_up(tmp_path): path = tmp_path / "db.sqlite" queue = JobQueue(path) first = queue.add(job(story_id=1)) queue.update(queue.get_next(), status=QueueStatus.RUNNING) for restarts in (1, 2): queue = JobQueue(path) started = queue.get_next() assert (started.id, started.restarts) == (first.id, restarts) queue.update(started, status=QueueStatus.RUNNING) queue = JobQueue(path) assert queue.get_next() is None (gone,) = queue.get_status() assert gone["status"] == "error" and "3 times" in gone["message"] def test_finished_jobs_leave_the_database(tmp_path, monkeypatch): queue = JobQueue(tmp_path / "db.sqlite") old = queue.add(job(story_id=1)) queue.update(old, status=QueueStatus.DONE) old.updated = datetime.now() - timedelta(seconds=config.JOB_HISTORY + 1) queue.get_status() assert JobQueue(tmp_path / "db.sqlite").get_status() == [] def test_the_api_form_of_a_job_hides_the_path_of_an_upload(): job = Job(mode=DownloadMode.IMPORT_FILE, options={"path": "/config/uploads/x.epub", "name": "x.epub"}) assert job.to_dict()["options"] == {"name": "x.epub"} assert Job.from_record(job.to_record()).options == job.options def test_a_record_of_a_newer_bindery_loads_without_its_unknown_fields(): job = Job(mode=DownloadMode.NEW, story_id=3) record = json.dumps({**json.loads(job.to_record()), "priority": 5}) loaded = Job.from_record(record) assert (loaded.id, loaded.story_id, loaded.mode) == (job.id, 3, DownloadMode.NEW)