bvandeusenandClaude Opus 5.5
8feec4dc77
list_backups, check_backup and make_backup are Settings → Backups for an agent, admin only like the page. make_backup runs the page's job in the background and is audited as "<user> via MCP". Restoring stays in the browser. get_overview names a scheduled backup that failed, is two hours past its slot, or has no bucket. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
110 lines
5.8 KiB
Python
110 lines
5.8 KiB
Python
"""The backup tools' pure parts (#5836): when the next backup is due, what the
|
|
overview says about backups, how a check reads to an agent, and their roles.
|
|
|
|
Reading the history, starting one and the audit row run against Postgres in
|
|
tests/integration/test_mcp_backups.py.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from steward.backup.check import AreaCheck, CheckReport, RowDiff, TableCheck
|
|
from steward.mcp.tools import backups, overview
|
|
from steward.models.users import UserRole
|
|
|
|
NOW = datetime(2026, 10, 15, 6, 0, tzinfo=timezone.utc) # three hours after the 03:00 slot
|
|
SLOT = datetime(2026, 10, 15, 3, 0, tzinfo=timezone.utc)
|
|
|
|
BUCKET = {"backup.s3_endpoint": "http://archives:8081", "backup.s3_bucket": "steward",
|
|
"backup.s3_access_key": "AK", "backup.s3_secret_key": "SK"}
|
|
|
|
|
|
def _settings(**over) -> dict:
|
|
return {"backup.enabled": True, "backup.schedule_hour": 3, "backup.schedule_tz": "UTC",
|
|
**BUCKET, **over}
|
|
|
|
|
|
def _entry(at: datetime, ok: bool, error: str = "") -> dict:
|
|
return {"at": at.isoformat(), "ok": ok, "by": "schedule", "name": "steward-x.tar" if ok else "",
|
|
"size": 1000 if ok else 0, "secs": 2.0, "error": error}
|
|
|
|
|
|
def test_the_next_backup_is_tomorrows_slot_an_hour_after_a_failure_or_the_next_tick():
|
|
assert backups.next_backup_at(_settings(**{"backup.enabled": False}), NOW) is None
|
|
done = _settings(**{"backup.last_success_at": (SLOT + timedelta(minutes=5)).isoformat()})
|
|
assert backups.next_backup_at(done, NOW) == SLOT + timedelta(days=1)
|
|
tried = _settings(**{"backup.last_success_at": (SLOT - timedelta(days=1)).isoformat(),
|
|
"backup.last_attempt_at": (NOW - timedelta(minutes=20)).isoformat()})
|
|
assert backups.next_backup_at(tried, NOW) == NOW + timedelta(minutes=40)
|
|
due = _settings(**{"backup.last_success_at": (SLOT - timedelta(days=1)).isoformat()})
|
|
assert backups.next_backup_at(due, NOW) == NOW + timedelta(minutes=15)
|
|
|
|
|
|
def test_the_overview_says_nothing_about_backups_that_are_off_running_or_fine():
|
|
assert overview.backup_findings({"backup.enabled": False}, False, NOW) == []
|
|
failed = _settings(**{"backup.history": [_entry(NOW, False, "refused")]})
|
|
assert overview.backup_findings(failed, True, NOW) == []
|
|
fine = _settings(**{"backup.history": [_entry(SLOT, True)],
|
|
"backup.last_success_at": SLOT.isoformat()})
|
|
assert overview.backup_findings(fine, False, NOW) == []
|
|
assert overview.backup_findings(_settings(), False, NOW) == [] # never run yet: not overdue
|
|
|
|
|
|
def test_backups_on_with_no_bucket_is_a_finding():
|
|
settings = {"backup.enabled": True, "backup.schedule_hour": 3}
|
|
[f] = overview.backup_findings(settings, False, NOW)
|
|
assert f["kind"] == "backup_not_set_up" and f["next"] == "list_backups()"
|
|
|
|
|
|
def test_failed_backups_are_counted_since_the_first_in_a_row():
|
|
good = SLOT - timedelta(days=1)
|
|
history = [_entry(NOW - timedelta(hours=1), False, "Upload of steward-x.tar refused: 500"),
|
|
_entry(SLOT, False, "unreachable"), _entry(good, True)]
|
|
[f] = overview.backup_findings(_settings(**{"backup.history": history,
|
|
"backup.last_success_at": good.isoformat()}),
|
|
False, NOW)
|
|
assert f["kind"] == "backup_failed" and f["severity"] == "warning"
|
|
assert "(2 tries in a row): Upload of steward-x.tar refused: 500" in f["summary"]
|
|
assert f"the newest good one was sent {overview.iso(good)}" in f["summary"]
|
|
assert f["since"] == "2026-10-15T03:00:00Z"
|
|
|
|
|
|
def test_no_backup_two_hours_past_its_slot_is_overdue():
|
|
yesterday = SLOT - timedelta(days=1)
|
|
settings = _settings(**{"backup.history": [_entry(yesterday, True)],
|
|
"backup.last_success_at": yesterday.isoformat()})
|
|
[f] = overview.backup_findings(settings, False, NOW)
|
|
assert f["kind"] == "backup_overdue" and f["since"] == "2026-10-15T03:00:00Z"
|
|
assert overview.backup_findings(settings, False, SLOT + timedelta(minutes=90)) == []
|
|
|
|
|
|
def test_a_check_names_only_the_tables_that_differ_and_keeps_secret_values_out():
|
|
many = [f"host-{i}" for i in range(backups.ROWS_SHOWN + 5)]
|
|
report = CheckReport(
|
|
name="steward-20261015T030000Z.tar", manifest={"created_at": SLOT.isoformat()},
|
|
schema="same", backup_heads=["a1"], here_heads=["a1"], key_same=True, sealed=4, unopened=0,
|
|
areas=[AreaCheck("Settings", [
|
|
TableCheck("app_settings", same=10, differ=[
|
|
RowDiff("backup.schedule_hour", ["value_json"], False, "3", "4"),
|
|
RowDiff("smtp.password", ["value_json"], True)]),
|
|
TableCheck("users", same=2)]),
|
|
AreaCheck("Hosts", [TableCheck("hosts", same=1, only_backup=many)])])
|
|
out = backups.report_dict(report)
|
|
assert out["ok"] is True and out["made_at"] == "2026-10-15T03:00:00Z"
|
|
assert out["changes"] == 2 + len(many) and out["rows"] == 13 + out["changes"]
|
|
settings_area, hosts_area = out["areas"]
|
|
assert [t["table"] for t in settings_area["tables"]] == ["app_settings"]
|
|
plain, secret = settings_area["tables"][0]["differ"]
|
|
assert (plain["backup_value"], plain["here_value"]) == ("3", "4")
|
|
assert secret == {"label": "smtp.password", "columns": ["value_json"], "secret": True}
|
|
hosts = hosts_area["tables"][0]
|
|
assert len(hosts["only_in_backup"]) == backups.ROWS_SHOWN and hosts["not_named"] == 5
|
|
|
|
|
|
def test_backup_tools_are_admin_like_the_page(app):
|
|
mcp = app.mcp_server
|
|
for name, read_only in (("list_backups", True), ("check_backup", True), ("make_backup", False)):
|
|
assert mcp.tool_roles[name] is UserRole.admin, name
|
|
assert mcp._tool_manager.get_tool(name).annotations.read_only_hint is read_only, name
|
|
assert (mcp._tool_manager.get_tool(name).description or "").strip(), name
|