FabledGrimoire/tests/test_services_backup_file.py

"""Unit tests for the backup file (services/backup_file.py).

The file exists to be deduplicated between runs by a tool like restic, so
most of what is tested here is a property of BYTES: what stays identical
when nothing changed, and how little moves when one row does. A round trip
alone would pass for a gzip of one big JSON document, which is exactly the
shape this file replaced.
"""
import io
import tarfile
from datetime import datetime, timezone
from unittest.mock import patch

import pytest

from grimoire.services import backup_file
from grimoire.services.backup_file import (
    ArchiveError, archive_name, is_archive, read_archive, write_archive,
    write_archive_to,
)

AT = datetime(2026, 10, 15, 3, 0, 0, tzinfo=timezone.utc)


def _payload(**over) -> dict:
    payload = {
        "version": 25,
        "scope": "full",
        "exported_at": "2026-10-15T03:00:00+00:00",
        "_security_notice": "hashed passwords inside",
        "_not_included": ["api_keys", "note_embeddings"],
        "users": [{"id": 1, "username": "ada", "role": "admin"}],
        "notes": [
            {"id": n, "user_id": 1, "title": f"note {n}", "body": "x" * 200,
             "tags": ["a", "b"]}
            for n in range(1, 51)
        ],
        "settings": [],
    }
    payload.update(over)
    return payload


def _members(payload: dict) -> dict[str, bytes]:
    buf = io.BytesIO()
    write_archive_to(payload, buf, app_version="2026.10.15.0300")
    buf.seek(0)
    with tarfile.open(fileobj=buf, mode="r:") as tar:
        return {m.name: tar.extractfile(m).read() for m in tar.getmembers()}


def test_the_name_is_dated_in_utc_and_sorts():
    assert archive_name(AT) == "grimoire-20261015T030000Z.tar"
    local = datetime(2026, 10, 14, 23, 0, 0).astimezone()
    assert archive_name(local).endswith("Z.tar")


def test_round_trip_returns_the_payload_restore_takes():
    payload = _payload()
    buf = io.BytesIO()
    write_archive_to(payload, buf, app_version="v")
    buf.seek(0)
    assert read_archive(buf) == payload


def test_notices_stay_in_the_manifest_not_as_sections():
    """`_not_included` is a list, but of table names; it must not become a
    member that a restore would read as rows."""
    names = set(_members(_payload()))
    assert names == {"manifest.json", "notes.jsonl", "settings.jsonl", "users.jsonl"}


def test_members_come_in_one_order_with_fixed_headers():
    buf = io.BytesIO()
    write_archive_to(_payload(), buf, app_version="v")
    buf.seek(0)
    with tarfile.open(fileobj=buf, mode="r:") as tar:
        members = tar.getmembers()
    assert [m.name for m in members] == [
        "manifest.json", "notes.jsonl", "settings.jsonl", "users.jsonl",
    ]
    for m in members:
        assert (m.mtime, m.uid, m.gid, m.uname, m.gname, m.mode) == (
            0, 0, 0, "", "", 0o644,
        )


def test_an_unchanged_payload_gives_identical_sections_on_another_run():
    """The run's own time and the app version live in the manifest and
    nowhere else, so a second run over unchanged data repeats every section
    byte for byte."""
    first = _members(_payload())
    second = _members(_payload(exported_at="2026-10-16T03:00:00+00:00"))
    assert first["manifest.json"] != second["manifest.json"]
    for name in first:
        if name != "manifest.json":
            assert first[name] == second[name], name


def test_key_order_in_a_row_does_not_change_its_bytes():
    rows = _payload()["notes"]
    shuffled = [dict(reversed(list(r.items()))) for r in rows]
    assert _members(_payload(notes=shuffled))["notes.jsonl"] == \
        _members(_payload())["notes.jsonl"]


def test_editing_one_row_changes_one_line_of_one_member():
    before = _members(_payload())
    notes = _payload()["notes"]
    notes[24] = {**notes[24], "body": "edited"}
    after = _members(_payload(notes=notes))

    changed = [n for n in before if before[n] != after[n]]
    assert changed == ["notes.jsonl"]
    old_lines = before["notes.jsonl"].splitlines()
    new_lines = after["notes.jsonl"].splitlines()
    assert len(old_lines) == len(new_lines)
    assert [i for i, (a, b) in enumerate(zip(old_lines, new_lines)) if a != b] == [24]


def test_a_new_row_appends_to_the_end_of_its_member():
    """Rows come in primary-key order, so a table that grows grows at its
    tail: everything already sent is a prefix of the new member."""
    before = _members(_payload())["notes.jsonl"]
    notes = _payload()["notes"] + [{"id": 51, "user_id": 1, "title": "new"}]
    after = _members(_payload(notes=notes))["notes.jsonl"]
    assert after.startswith(before)
    assert after.count(b"\n") == before.count(b"\n") + 1


def test_the_file_is_an_uncompressed_tar():
    buf = io.BytesIO()
    write_archive_to(_payload(), buf, app_version="v")
    raw = buf.getvalue()
    assert raw[257:262] == b"ustar"
    assert b'"title":"note 7"' in raw


def test_is_archive_tells_a_tar_from_the_json_download():
    buf = io.BytesIO()
    write_archive_to(_payload(), buf, app_version="v")
    buf.seek(0)
    assert is_archive(buf)
    assert buf.tell() == 0
    assert not is_archive(io.BytesIO(b'{"version": 25, "users": []}'))


def test_write_archive_lands_under_the_dated_name_only(tmp_path):
    path = write_archive(_payload(), tmp_path, AT, app_version="v")
    assert path == tmp_path / "grimoire-20261015T030000Z.tar"
    assert [p.name for p in tmp_path.iterdir()] == [path.name]
    with open(path, "rb") as f:
        assert read_archive(f) == _payload()


def test_a_failed_write_leaves_nothing_behind(tmp_path):
    """Neither a half backup under the final name (a watcher would count it
    as arrived) nor a stray part file."""
    with patch.object(backup_file, "write_archive_to", side_effect=OSError("disk full")):
        with pytest.raises(OSError):
            write_archive(_payload(), tmp_path, AT, app_version="v")
    assert list(tmp_path.iterdir()) == []


def _rewrite(members: dict[str, bytes]) -> io.BytesIO:
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w", format=tarfile.USTAR_FORMAT) as tar:
        for name, data in members.items():
            info = tarfile.TarInfo(name)
            info.size = len(data)
            tar.addfile(info, io.BytesIO(data))
    buf.seek(0)
    return buf


def test_a_missing_section_is_an_error_not_a_thinner_restore():
    members = _members(_payload())
    del members["notes.jsonl"]
    with pytest.raises(ArchiveError, match="notes.jsonl"):
        read_archive(_rewrite(members))


def test_a_section_cut_short_is_an_error():
    members = _members(_payload())
    members["notes.jsonl"] = b"\n".join(members["notes.jsonl"].splitlines()[:10]) + b"\n"
    with pytest.raises(ArchiveError, match="10 rows"):
        read_archive(_rewrite(members))


def test_a_newer_format_is_refused_by_name():
    members = _members(_payload())
    members["manifest.json"] = members["manifest.json"].replace(
        b'"format_version": 1', b'"format_version": 99',
    )
    with pytest.raises(ArchiveError, match="upgrade"):
        read_archive(_rewrite(members))


def test_an_archive_written_under_the_name_scribe_still_reads():
    """Grimoire was called Scribe; its archives say so in the manifest."""
    members = _members(_payload())
    assert b"grimoire-backup-tar" in members["manifest.json"]
    members["manifest.json"] = members["manifest.json"].replace(
        b"grimoire-backup-tar", b"scribe-backup-tar",
    )
    assert read_archive(_rewrite(members)) == _payload()


def test_another_tools_tar_is_refused():
    members = _members(_payload())
    members["manifest.json"] = members["manifest.json"].replace(
        b"grimoire-backup-tar", b"someone-elses-tar",
    )
    with pytest.raises(ArchiveError, match="not a Grimoire backup"):
        read_archive(_rewrite(members))


def test_garbage_is_not_an_archive():
    with pytest.raises(ArchiveError):
        read_archive(io.BytesIO(b"not a tar at all" * 100))