FabledGrimoire/tests/test_integration_backup_file_restore.py

"""Real-Postgres proof that the scheduled backup's FILE restores: a payload
written as the tar, read back through `grimoire.restore`'s loader, and fed to
the real `restore_full_backup`.

The unit lane already shows the tar hands back the dict it was given. What
only this lane shows is that the dict it hands back is one the restore
accepts — every id remapped to the restored rows, nothing dropped — and that
the command-line restore refuses, on a live instance, a backup that doesn't
record this instance as its source.
"""
import pytest
import pytest_asyncio
from sqlalchemy import select

from grimoire.models import async_session
from grimoire.models.milestone import Milestone
from grimoire.models.note import Note
from grimoire.models.project import Project
from grimoire.models.user import User
from grimoire.restore import load_backup, run
from grimoire.services import backup
from grimoire.services.backup_file import write_archive
from tests.helpers import ensure_user

pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]

OWNER = "backup_file_restore_owner"
RESTORED = "backup_file_restore_restored"


async def _purge(username: str) -> None:
    async with async_session() as s:
        for user in (await s.execute(select(User).where(User.username == username))).scalars().all():
            for model in (Note, Milestone, Project):
                for row in (await s.execute(
                    select(model).where(model.user_id == user.id)
                )).scalars().all():
                    await s.delete(row)
            await s.flush()
            await s.delete(user)
        await s.commit()


@pytest_asyncio.fixture(autouse=True)
async def _no_leftovers():
    """Setup only — see the rule_version roundtrip module for why."""
    await _purge(RESTORED)
    await _purge(OWNER)


@pytest_asyncio.fixture
async def written(tmp_path):
    """A project with a milestone and a task in it, exported with the real
    row builders, renamed so the restore mints a separate user, and written
    as the backup tar. It also carries a `user_profiles` section, as an
    archive from before v27 does."""
    async with async_session() as s:
        user = await ensure_user(s, OWNER)
        project = Project(user_id=user.id, title="Restore drill")
        s.add(project)
        await s.flush()
        milestone = Milestone(user_id=user.id, project_id=project.id, title="Plan")
        s.add(milestone)
        await s.flush()
        note = Note(user_id=user.id, project_id=project.id, milestone_id=milestone.id,
                    title="Prove the restore", body="From the tar, not the JSON.",
                    status="todo", note_type="note", task_kind="work")
        s.add(note)
        await s.commit()
        payload = {
            "version": backup.BACKUP_VERSION,
            "scope": "full",
            "users": backup._user_rows([user]),
            "projects": backup._project_rows([project]),
            "milestones": backup._milestone_rows([milestone]),
            "notes": backup._note_rows([note]),
            "user_profiles": [{"user_id": user.id, "display_name": "Drill Owner",
                               "tone": "technical", "interests": ["backups"]}],
        }
    payload["users"][0]["username"] = RESTORED
    from datetime import datetime, timezone
    path = write_archive(payload, tmp_path, datetime.now(timezone.utc), app_version="test")
    yield path
    await _purge(RESTORED)
    await _purge(OWNER)


async def test_the_tar_restores_with_every_id_remapped(written):
    await backup.restore_full_backup(load_backup(written))
    async with async_session() as s:
        user = (await s.execute(select(User).where(User.username == RESTORED))).scalars().one()
        project = (await s.execute(select(Project).where(Project.user_id == user.id))).scalars().one()
        milestone = (await s.execute(select(Milestone).where(Milestone.user_id == user.id))).scalars().one()
        note = (await s.execute(select(Note).where(Note.user_id == user.id))).scalars().one()
    assert project.title == "Restore drill"
    assert (note.project_id, note.milestone_id) == (project.id, milestone.id)
    assert note.body == "From the tar, not the JSON."
    assert note.status == "todo"


async def test_the_command_refuses_a_backup_from_no_known_instance(written):
    """This database is live and the backup names no instance it came from,
    so it can't be rolled back to: the command stops before restoring
    anything, even with --yes."""
    assert await run(written, yes=True) == 3
    async with async_session() as s:
        restored = (await s.execute(select(User).where(User.username == RESTORED))).scalars().all()
    assert restored == []