feat(523): the backup job - an hourly tick that runs at the chosen hour, Back up now, and the last 30 runs

services/backup_job.py reads its config when it fires, so startup reads
nothing (#4181) and a changed hour needs no reschedule. A run is held to
a 60-minute deadline (a hung NFS mount would otherwise hold the lock and
skip every later backup), never runs twice at once, and keeps its outcome
- file, size, duration, or the reason in words - in the bounded
backup_runs setting, which no backup carries. summarise() gives the page
its verdict: overdue when on and no success for 26h.

Admin API: GET/PUT /api/admin/backups, POST /backups/check (tests unsaved
form values, 45s deadline), POST /backups/run (202, background).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/test_services_backup_job.py

@@ -0,0 +1,223 @@
+"""The scheduled backup job (services/backup_job.py) and the admin surface
+over it. No database: admin settings are an in-memory dict, and the export
+and delivery are stand-ins, so what is under test is the job's own contract —
+it records every outcome, keeps a bounded history, runs only at its hour,
+and never runs twice at once.
+"""
+import asyncio
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+import pytest
+
+from scribe.services import backup_destination, backup_job
+from scribe.services.backup_destination import (
+    BackupConfig, Delivered, DestinationError, apply_changes,
+)
+
+NOW = datetime(2026, 10, 15, 2, 0, 0, tzinfo=timezone.utc)
+
+
+class _Settings(dict):
+    async def get(self, key, default=""):
+        return self.get_value(key, default)
+
+    def get_value(self, key, default):
+        return dict.get(self, key, default)
+
+    async def set(self, key, value):
+        self[key] = value
+
+
+def _patched(settings, deliver=None, config=BackupConfig(enabled=True)):
+    async def load_config():
+        return config
+
+    async def load_secret():
+        return (None, False)
+
+    async def export():
+        return {"version": 25, "users": []}
+
+    async def delivered(payload, config, secret, at, **kw):
+        return Delivered("scribe-20261015T020000Z.tar", 1234, "/backups")
+
+    return [
+        patch.object(backup_job, "get_admin_setting", settings.get),
+        patch.object(backup_job, "set_admin_setting", settings.set),
+        patch.object(backup_job, "load_config", load_config),
+        patch.object(backup_job, "load_secret", load_secret),
+        patch.object(backup_job, "deliver", deliver or delivered),
+        patch("scribe.services.backup.export_full_backup", export),
+    ]
+
+
+def _run(coro, patches):
+    for p in patches:
+        p.start()
+    try:
+        return asyncio.run(coro)
+    finally:
+        for p in reversed(patches):
+            p.stop()
+
+
+def test_a_good_run_is_kept_with_its_file_and_size():
+    settings = _Settings()
+    run = _run(backup_job.run_backup("manual"), _patched(settings))
+    assert run["status"] == "ok"
+    assert (run["file"], run["size_bytes"], run["trigger"]) == (
+        "scribe-20261015T020000Z.tar", 1234, "manual",
+    )
+    runs = _run(backup_job.list_runs(), _patched(settings))
+    assert runs == [run]
+
+
+def test_a_failed_run_is_kept_with_its_reason_in_words():
+    async def refuse(*a, **kw):
+        raise DestinationError("/backups isn't there.")
+
+    settings = _Settings()
+    run = _run(backup_job.run_backup("scheduled"), _patched(settings, deliver=refuse))
+    assert run["status"] == "failed"
+    assert run["error"] == "/backups isn't there."
+
+
+def test_an_unexpected_error_is_a_failed_run_not_a_crash():
+    async def boom(*a, **kw):
+        raise RuntimeError("disk on fire")
+
+    settings = _Settings()
+    run = _run(backup_job.run_backup("scheduled"), _patched(settings, deliver=boom))
+    assert run["status"] == "failed"
+    assert "disk on fire" in run["error"]
+
+
+def test_a_run_that_hangs_is_abandoned_at_the_deadline():
+    async def hang(*a, **kw):
+        await asyncio.sleep(10)
+
+    settings = _Settings()
+    with patch.object(backup_job, "RUN_DEADLINE", timedelta(seconds=0.05)):
+        run = _run(backup_job.run_backup("scheduled"), _patched(settings, deliver=hang))
+    assert run["status"] == "failed"
+    assert "didn't finish" in run["error"]
+    assert not backup_job.is_running()
+
+
+def test_the_history_keeps_only_the_newest_runs():
+    settings = _Settings()
+
+    async def many():
+        for _ in range(backup_job.KEEP_RUNS + 5):
+            await backup_job.run_backup("scheduled")
+        return await backup_job.list_runs()
+
+    runs = _run(many(), _patched(settings))
+    assert len(runs) == backup_job.KEEP_RUNS
+    assert runs[0]["started_at"] >= runs[-1]["started_at"]
+
+
+def test_a_second_run_while_one_is_going_is_refused():
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    async def slow(*a, **kw):
+        started.set()
+        await release.wait()
+        return Delivered("x.tar", 1, "/backups")
+
+    async def both():
+        first = asyncio.create_task(backup_job.run_backup("manual"))
+        await started.wait()
+        with pytest.raises(backup_job.BackupBusy):
+            await backup_job.run_backup("manual")
+        release.set()
+        return await first
+
+    run = _run(both(), _patched(_Settings(), deliver=slow))
+    assert run["status"] == "ok"
+
+
+@pytest.mark.parametrize("enabled,hour,ran", [
+    (True, NOW.hour, True),
+    (True, (NOW.hour + 1) % 24, False),
+    (False, NOW.hour, False),
+])
+def test_the_hourly_tick_runs_only_when_on_and_at_its_hour(enabled, hour, ran):
+    calls = []
+
+    async def fake_run(trigger):
+        calls.append(trigger)
+
+    class _Now(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return NOW
+
+    async def config():
+        return BackupConfig(enabled=enabled, hour=hour)
+
+    with patch.object(backup_job, "load_config", config), \
+         patch.object(backup_job, "run_backup", fake_run), \
+         patch.object(backup_job, "datetime", _Now):
+        asyncio.run(backup_job._hourly_check())
+    assert calls == (["scheduled"] if ran else [])
+
+
+# ── The page's verdict ───────────────────────────────────────────────────────
+
+def _ok(at):
+    return {"status": "ok", "finished_at": at.isoformat()}
+
+
+def test_a_recent_success_is_not_overdue():
+    s = backup_job.summarise([_ok(NOW - timedelta(hours=3))], enabled=True, now=NOW)
+    assert s["overdue"] is False
+    assert s["last_success"]["status"] == "ok"
+
+
+def test_failures_after_an_old_success_are_overdue():
+    runs = [{"status": "failed", "finished_at": NOW.isoformat()},
+            _ok(NOW - timedelta(hours=30))]
+    s = backup_job.summarise(runs, enabled=True, now=NOW)
+    assert s["overdue"] is True
+    assert s["last_run"]["status"] == "failed"
+
+
+def test_off_is_never_overdue_and_nothing_run_yet_is_not_late():
+    assert backup_job.summarise([], enabled=False, now=NOW)["overdue"] is False
+    assert backup_job.summarise([], enabled=True, now=NOW)["overdue"] is False
+    failed_only = [{"status": "failed", "finished_at": NOW.isoformat()}]
+    assert backup_job.summarise(failed_only, enabled=True, now=NOW)["overdue"] is True
+
+
+# ── What an admin sends ──────────────────────────────────────────────────────
+
+def test_changes_apply_on_top_of_the_current_config():
+    config = apply_changes(BackupConfig(), {"enabled": True, "hour": "5", "s3_secret": "x"})
+    assert (config.enabled, config.hour, config.destination) == (True, 5, "folder")
+
+
+@pytest.mark.parametrize("data,said", [
+    ({"hour": 24}, "from 0 to 23"),
+    ({"hour": "two"}, "whole number"),
+    ({"destination": "ftp"}, "folder or an S3 bucket"),
+    ({"enabled": "yes"}, "true or false"),
+])
+def test_a_typed_mistake_is_refused_with_the_reason(data, said):
+    with pytest.raises(ValueError, match=said):
+        apply_changes(BackupConfig(), data)
+
+
+def test_routes_and_scheduler_are_wired():
+    from scribe.app import create_app
+    from scribe.routes import admin as admin_routes
+
+    for name in ("get_backups", "update_backups", "check_backups_destination", "run_backup_now"):
+        assert callable(getattr(admin_routes, name))
+    rules = {r.rule for r in create_app().url_map.iter_rules()}
+    assert {"/api/admin/backups", "/api/admin/backups/check", "/api/admin/backups/run"} <= rules
+    assert callable(backup_job.start_backup_scheduler)
+    assert callable(backup_job.stop_backup_scheduler)
+    assert backup_destination.SECRET_SETTING == "backup_s3_secret"