FabledGrimoire/tests/test_services_backup_destination.py

"""The backup destination (services/backup_destination.py) and the sealing it
keeps the S3 secret under (services/secrets.py).

No database: the admin settings are patched, the folder is a tmp_path, and the
bucket is an httpx.MockTransport. What these pin is the contract a person
meets in Settings — a config that is incomplete says what is missing, a place
that can't take a backup says why, and the secret never leaves sealed.
"""
import asyncio
import io
from datetime import datetime, timezone
from types import SimpleNamespace
from unittest.mock import patch

import httpx
import pytest

from grimoire.services import backup, backup_destination, secrets
from grimoire.services.backup_destination import (
    BackupConfig, DestinationError, check_destination, deliver,
)
from grimoire.services.backup_file import read_archive

AT = datetime(2026, 10, 15, 2, 0, 0, tzinfo=timezone.utc)
PAYLOAD = {
    "version": 25, "scope": "full", "exported_at": AT.isoformat(),
    "users": [{"id": 1, "username": "ada"}], "notes": [],
}
S3 = BackupConfig(
    destination="s3", s3_endpoint="http://archives:8081", s3_bucket="grimoire",
    s3_access_key="AKIA", s3_region="us-east-1",
)


# ── Sealing ──────────────────────────────────────────────────────────────────

def test_a_sealed_secret_opens_and_is_not_the_plain_text():
    token = secrets.seal("wJalrXUtnFEMI/K7MDENG")
    assert "wJalrXUtnFEMI" not in token
    assert secrets.unseal(token) == "wJalrXUtnFEMI/K7MDENG"


def test_another_secret_key_cannot_open_it():
    token = secrets.seal("s3cret")
    with patch.object(secrets.Config, "SECRET_KEY", "a-different-instance"):
        assert secrets.unseal(token) is None


def test_unseal_of_nothing_or_garbage_is_none():
    assert secrets.unseal("") is None
    assert secrets.unseal("not a fernet token") is None


def test_the_secret_setting_never_reaches_a_backup():
    rows = [SimpleNamespace(user_id=1, key=k, value="v") for k in (
        "backup_config", backup_destination.SECRET_SETTING, "theme",
    )]
    keys = [r["key"] for r in backup._setting_rows(rows)]
    assert keys == ["backup_config", "theme"]


def test_load_secret_tells_a_missing_secret_from_one_it_cannot_open():
    async def stored(value):
        async def get(key, default=""):
            return value
        with patch.object(backup_destination, "get_admin_setting", get):
            return await backup_destination.load_secret()

    assert asyncio.run(stored("")) == (None, False)
    assert asyncio.run(stored(secrets.seal("k"))) == ("k", True)
    assert asyncio.run(stored("sealed-elsewhere")) == (None, True)


# ── Config ───────────────────────────────────────────────────────────────────

def test_config_round_trips_and_defaults_to_off():
    assert BackupConfig().enabled is False
    assert BackupConfig.from_dict(S3.to_dict()) == S3


def test_a_damaged_config_still_loads():
    config = BackupConfig.from_dict({
        "hour": "25", "destination": "ftp", "enabled": "yes", "unknown": 1,
        "s3_bucket": "  grimoire  ",
    })
    assert config.hour == backup_destination.DEFAULT_HOUR
    assert config.destination == "folder"
    assert config.enabled is False
    assert config.s3_bucket == "grimoire"


def test_an_incomplete_s3_config_names_what_is_missing():
    said = BackupConfig(destination="s3").problems(has_secret=False)
    assert said == [
        "The S3 address must start with http:// or https://.",
        "The bucket name is missing.",
        "The access key is missing.",
        "The secret key is missing.",
    ]
    assert S3.problems(has_secret=True) == []


def test_a_relative_folder_is_refused():
    assert BackupConfig(folder_path="backups").problems(has_secret=False)


# ── Folder ───────────────────────────────────────────────────────────────────

def test_deliver_to_a_folder_adds_one_dated_file(tmp_path):
    config = BackupConfig(folder_path=str(tmp_path))
    got = asyncio.run(deliver(PAYLOAD, config, None, AT, app_version="v"))
    assert got.name == "grimoire-20261015T020000Z.tar"
    assert [p.name for p in tmp_path.iterdir()] == [got.name]
    assert got.size_bytes == (tmp_path / got.name).stat().st_size
    with open(tmp_path / got.name, "rb") as f:
        assert read_archive(f) == PAYLOAD


def test_a_folder_that_isnt_mounted_says_so(tmp_path):
    config = BackupConfig(folder_path=str(tmp_path / "not-mounted"))
    with pytest.raises(DestinationError, match="isn't there"):
        asyncio.run(deliver(PAYLOAD, config, None, AT, app_version="v"))


def test_checking_a_folder_leaves_nothing_behind(tmp_path):
    (tmp_path / "grimoire-20261015T020000Z.tar").write_bytes(b"x")
    config = BackupConfig(folder_path=str(tmp_path))
    said = asyncio.run(check_destination(config, None))
    assert "holds 1 Grimoire backup" in said
    assert [p.name for p in tmp_path.iterdir()] == ["grimoire-20261015T020000Z.tar"]


def test_a_folder_counts_backups_written_under_the_name_scribe(tmp_path):
    """Grimoire was called Scribe; the destination still holds those files."""
    for name in ("scribe-20261014T020000Z.tar", "grimoire-20261015T020000Z.tar", "notes.txt"):
        (tmp_path / name).write_bytes(b"x")
    said = asyncio.run(check_destination(BackupConfig(folder_path=str(tmp_path)), None))
    assert "holds 2 Grimoire backup" in said


# ── S3 ───────────────────────────────────────────────────────────────────────

def test_deliver_to_s3_puts_the_dated_file_and_keeps_no_copy():
    seen = {}

    def handler(request: httpx.Request):
        seen["method"], seen["url"] = request.method, str(request.url)
        seen["body"] = request.read()
        return httpx.Response(200)

    got = asyncio.run(deliver(PAYLOAD, S3, "secret", AT, app_version="v",
                              transport=httpx.MockTransport(handler)))
    assert seen["method"] == "PUT"
    assert seen["url"] == "http://archives:8081/grimoire/grimoire-20261015T020000Z.tar"
    assert read_archive(io.BytesIO(seen["body"])) == PAYLOAD
    assert got.size_bytes == len(seen["body"])


def test_a_refused_upload_is_a_destination_error_in_words():
    body = b"<Error><Code>AccessDenied</Code><Message>no</Message></Error>"
    transport = httpx.MockTransport(lambda r: httpx.Response(403, content=body))
    with pytest.raises(DestinationError, match="the key isn't allowed to do that"):
        asyncio.run(deliver(PAYLOAD, S3, "secret", AT, app_version="v", transport=transport))


def test_s3_without_its_secret_sends_nothing():
    def handler(request):
        raise AssertionError("nothing should be sent")

    with pytest.raises(DestinationError, match="secret key is missing"):
        asyncio.run(deliver(PAYLOAD, S3, None, AT, app_version="v",
                            transport=httpx.MockTransport(handler)))


def test_checking_s3_only_lists():
    """One listing per name the backups have carried: Grimoire was called
    Scribe, and the bucket still holds those objects."""
    methods = []

    def handler(request):
        methods.append(request.method)
        key = request.url.params["prefix"] + "20261014T020000Z.tar"
        xml = (b'<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'
               b"<Contents><Key>" + key.encode() + b"</Key><Size>9</Size>"
               b"<LastModified>x</LastModified></Contents></ListBucketResult>")
        return httpx.Response(200, content=xml)

    said = asyncio.run(check_destination(S3, "secret", transport=httpx.MockTransport(handler)))
    assert methods == ["GET", "GET"]
    assert "holds 2 Grimoire backup" in said