"""Integration: Restore on Settings → Backups (#5817). A chosen backup is checked and held; nothing starts until the person types the word, an older backup also needs Replace everything ticked, a newer one has no button, and only an admin gets this far. The restore itself is restoring.py's (stubbed here; tests/core/test_restoring.py covers it).""" from __future__ import annotations import asyncio import io import os import time import uuid import pytest pytestmark = pytest.mark.integration CSRF = {"X-CSRF-Token": "test-token"} @pytest.fixture def app(): if not os.environ.get("STEWARD_DATABASE_URL"): pytest.skip("needs Postgres") from steward.app import create_app return create_app(testing=False) async def _client(app, role): from steward.models.users import User, UserRole uid, tag = str(uuid.uuid4()), uuid.uuid4().hex[:8] async with app.db_sessionmaker() as s: async with s.begin(): s.add(User(id=uid, username=f"rs-{tag}", email=f"rs-{tag}@test", password_hash="x", role=UserRole(role), is_active=True)) client = app.test_client() async with client.session_transaction() as sess: sess.update(user_id=uid, user_role=role, username=f"rs-{tag}", _csrf="test-token", login_at=time.time()) return client def _report(schema="same", unopened=0): from steward.backup.check import AreaCheck, CheckReport, TableCheck return CheckReport(name="steward-20261015T030000Z.tar", manifest={"created_at": "2026-10-15T03:00:00+00:00"}, schema=schema, backup_heads=["a"], here_heads=["a"], key_same=True, sealed=1, unopened=unopened, areas=[AreaCheck("Settings", [TableCheck("app_settings", same=4)])]) @pytest.fixture def stubs(monkeypatch, tmp_path): from steward.backup import job, restoring seen = {"started": [], "schema": "same", "unopened": 0} async def check_file(app, path, name=""): seen["checked"] = path.read_bytes() return _report(seen["schema"], seen["unopened"]) def start_restore(app, path, name, *, username, replace_all=False, owned_dir=None): seen["started"].append({"path": path, "name": name, "replace_all": replace_all, "owned_dir": owned_dir}) return True monkeypatch.setattr(restoring, "SAFETY_DIR", tmp_path) monkeypatch.setattr(job, "check_file", check_file) monkeypatch.setattr(restoring, "start_restore", start_restore) return seen async def _upload(client, name="steward-20261015T030000Z.tar"): from quart.datastructures import FileStorage return await client.post("/settings/backups/restore/file/", headers=CSRF, files={ "file": FileStorage(io.BytesIO(b"a backup"), filename=name)}) def _token(page: str) -> str: import re return re.search(r'name="token" value="([0-9a-f]{32})"', page).group(1) def test_nothing_starts_until_the_word_is_typed(app, stubs): async def _go(): admin = await _client(app, "admin") page = await (await _upload(admin)).get_data(as_text=True) assert stubs["checked"] == b"a backup" assert "Matches what is in place" in page and "Type restore to confirm" in page token = _token(page) wrong = await admin.post("/settings/backups/restore/start/", headers=CSRF, form={"token": token, "confirm": "yes"}) assert wrong.status_code == 200 and not stubs["started"] assert "Type restore to confirm." in await wrong.get_data(as_text=True) go = await admin.post("/settings/backups/restore/start/", headers=CSRF, form={"token": token, "confirm": " Restore "}) assert go.status_code == 302 and go.headers["Location"].endswith("/settings/backups/") [started] = stubs["started"] assert started["name"] == "steward-20261015T030000Z.tar" and not started["replace_all"] assert started["path"].read_bytes() == b"a backup" and started["owned_dir"].name == token gone = await admin.post("/settings/backups/restore/start/", headers=CSRF, form={"token": "0" * 32, "confirm": "restore"}) assert "no longer held" in await gone.get_data(as_text=True) asyncio.run(_go()) def test_an_older_backup_also_needs_replace_everything(app, stubs): stubs["schema"] = "older" async def _go(): admin = await _client(app, "admin") page = await (await _upload(admin)).get_data(as_text=True) assert 'name="replace_all"' in page and "replaces everything" in page token = _token(page) await admin.post("/settings/backups/restore/start/", headers=CSRF, form={"token": token, "confirm": "restore"}) assert not stubs["started"] await admin.post("/settings/backups/restore/start/", headers=CSRF, form={"token": token, "confirm": "restore", "replace_all": "on"}) assert [s["replace_all"] for s in stubs["started"]] == [True] asyncio.run(_go()) def test_a_backup_that_cant_go_in_has_no_button(app, stubs): async def _go(): admin = await _client(app, "admin") stubs["schema"] = "newer" newer = await (await _upload(admin)).get_data(as_text=True) stubs["schema"], stubs["unopened"] = "same", 1 unopened = await (await _upload(admin)).get_data(as_text=True) return newer, unopened newer, unopened = asyncio.run(_go()) assert "Update Steward, then restore it." in newer and 'name="token"' not in newer assert "wouldn't come back" in unopened and 'name="token"' not in unopened def test_only_an_admin_restores(app, stubs): async def _go(): operator = await _client(app, "operator") return (await _upload(operator)).status_code, (await operator.post( "/settings/backups/restore/", headers=CSRF, form={"name": "steward-20261015T030000Z.tar"}) ).status_code assert asyncio.run(_go()) == (403, 403) assert not stubs["started"] def test_the_backups_page_offers_restore_and_says_when_one_failed(app, stubs, monkeypatch): from steward.backup import restoring monkeypatch.setattr(restoring, "STATE", restoring.Restoring( name="steward-20261014T030000Z.tar", error="The restore was undone, nothing changed: x")) async def _go(): admin = await _client(app, "admin") return await (await admin.get("/settings/backups/")).get_data(as_text=True) page = asyncio.run(_go()) assert 'formaction="/settings/backups/restore/file/"' in page assert "The restore of steward-20261014T030000Z.tar" in page and "nothing changed: x" in page