import io import tarfile from datetime import datetime, timedelta, timezone import pytest from inkwell import backup_file from inkwell.backup_data import NOT_INCLUDED, SETTINGS_NOT_INCLUDED, exported_tables from inkwell.backup_file import ArchiveError, Contents, read_archive, read_rows, safe_media_path, write_archive from inkwell.backup_job import TRIES_PER_DAY, is_due, last_slot from inkwell.models import Base from inkwell.settings import REGISTRY, SECRET_KEY_SETTING AT = datetime(2026, 10, 16, 2, 0, tzinfo=timezone.utc) def _contents(tmp_path, notes=None): media = tmp_path / "media" (media / "ab").mkdir(parents=True, exist_ok=True) (media / "ab" / "photo.png").write_bytes(b"\x89PNG not really") return Contents( tables={ "notes": notes or [{"id": "1", "body": "milk"}, {"id": "2", "body": "eggs"}], "labels": [{"id": "9", "name": "grocery"}], }, files={"ab/photo.png": media / "ab" / "photo.png"}, ) def _members(path) -> dict[str, bytes]: with tarfile.open(path) as tar: return {m.name: tar.extractfile(m).read() for m in tar.getmembers()} # ---- what goes in --------------------------------------------------------------------- def test_every_table_is_backed_up_or_named_as_left_out(): """A table added later must not silently go missing from backups.""" every = {t.name for t in Base.metadata.sorted_tables} backed_up = {t.name for t in exported_tables()} assert set(NOT_INCLUDED) <= every, "NOT_INCLUDED names a table that doesn't exist" assert backed_up | set(NOT_INCLUDED) == every def test_no_credential_ever_enters_a_backup(): secrets = {d.key for d in REGISTRY if d.secret} assert "backup_s3_secret_key" in secrets and "smtp_password" in secrets assert secrets | {SECRET_KEY_SETTING} <= SETTINGS_NOT_INCLUDED # ---- the file --------------------------------------------------------------------------- def test_unchanged_data_writes_the_same_bytes_on_another_day(tmp_path): out1, out2 = tmp_path / "one", tmp_path / "two" out1.mkdir() out2.mkdir() first, _ = write_archive(_contents(tmp_path), out1, AT, app_version="a") second, _ = write_archive(_contents(tmp_path), out2, AT + timedelta(days=1), app_version="b") a, b = _members(first), _members(second) assert a.keys() == b.keys() for name in a: if name != "manifest.json": assert a[name] == b[name], name assert a["manifest.json"] != b["manifest.json"] def test_one_edit_changes_one_line_of_one_member(tmp_path): (tmp_path / "one").mkdir() (tmp_path / "two").mkdir() before, _ = write_archive(_contents(tmp_path), tmp_path / "one", AT, app_version="a") edited = [{"id": "1", "body": "oat milk"}, {"id": "2", "body": "eggs"}] after, _ = write_archive(_contents(tmp_path, edited), tmp_path / "two", AT, app_version="a") a, b = _members(before), _members(after) changed = [n for n in a if a[n] != b[n]] assert changed == ["tables/notes.jsonl"] lines_a, lines_b = a[changed[0]].splitlines(), b[changed[0]].splitlines() assert sum(x != y for x, y in zip(lines_a, lines_b)) == 1 def test_members_come_in_a_fixed_order_with_fixed_headers(tmp_path): (tmp_path / "out").mkdir() path, _ = write_archive(_contents(tmp_path), tmp_path / "out", AT, app_version="a") with tarfile.open(path) as tar: members = tar.getmembers() assert [m.name for m in members] == [ "manifest.json", "tables/labels.jsonl", "tables/notes.jsonl", "files/ab/photo.png", ] assert {(m.mtime, m.uid, m.gid, m.mode) for m in members} == {(0, 0, 0, 0o644)} assert path.name == "inkwell-20261016T020000Z.tar" def test_nothing_appears_under_the_final_name_when_writing_fails(tmp_path): contents = _contents(tmp_path) contents.files["gone.png"] = tmp_path / "media" / "gone.png" # stat() fails (tmp_path / "out").mkdir() with pytest.raises(FileNotFoundError): write_archive(contents, tmp_path / "out", AT, app_version="a") assert list((tmp_path / "out").iterdir()) == [] def test_a_backup_reads_back_as_it_was_written(tmp_path): (tmp_path / "out").mkdir() path, _ = write_archive(_contents(tmp_path), tmp_path / "out", AT, app_version="v1") with open(path, "rb") as f: manifest, members = read_archive(f) got = {} for m in members: got[(m.kind, m.name)] = read_rows(m) if m.kind == "table" else m.fileobj.read() assert manifest["app_version"] == "v1" assert manifest["tables"] == {"labels": 1, "notes": 2} assert got[("table", "notes")][0] == {"body": "milk", "id": "1"} assert got[("file", "ab/photo.png")] == b"\x89PNG not really" def test_a_file_that_is_not_a_backup_is_refused(): buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as tar: tar.addfile(tarfile.TarInfo("other.txt"), io.BytesIO(b"")) buf.seek(0) with pytest.raises(ArchiveError): read_archive(buf) @pytest.mark.parametrize("name", ["../etc/passwd", "/etc/passwd", "a/../../b", ""]) def test_a_file_path_that_leaves_the_media_folder_is_refused(name): with pytest.raises(ArchiveError): safe_media_path(name) # ---- the schedule --------------------------------------------------------------------- def _run(at: datetime, ok: bool) -> dict: return {"started_at": at.isoformat(), "finished_at": (at + timedelta(minutes=1)).isoformat(), "ok": ok} def test_the_slot_is_the_most_recent_scheduled_hour(): assert last_slot(datetime(2026, 10, 16, 1, 59, tzinfo=timezone.utc), 2) == datetime( 2026, 10, 15, 2, 0, tzinfo=timezone.utc ) assert last_slot(AT, 2) == AT def test_due_once_per_slot_and_a_failure_is_tried_again_later_not_at_once(): now = AT + timedelta(minutes=5) assert is_due(now, 2, []) is True assert is_due(now, 2, [_run(AT, ok=True)]) is False # Yesterday's success doesn't count for today. assert is_due(now, 2, [_run(AT - timedelta(days=1), ok=True)]) is True failed = [_run(AT, ok=False)] assert is_due(AT + timedelta(minutes=30), 2, failed) is False assert is_due(AT + timedelta(hours=1, minutes=2), 2, failed) is True # The ladder doubles: 1 h, then 2 h after the second failure. twice = [_run(AT + timedelta(hours=1), ok=False), _run(AT, ok=False)] assert is_due(AT + timedelta(hours=2), 2, twice) is False assert is_due(AT + timedelta(hours=3, minutes=2), 2, twice) is True # And stops for the day after its tries are used. spent = [_run(AT + timedelta(hours=h), ok=False) for h in range(TRIES_PER_DAY)][::-1] assert is_due(AT + timedelta(hours=20), 2, spent) is False def test_the_archive_name_prefix_matches_what_a_destination_counts(): assert backup_file.archive_name(AT).startswith(backup_file.NAME_PREFIX) assert backup_file.archive_name(AT).endswith(backup_file.NAME_SUFFIX)