bvandeusenandClaude Opus 5.5
1942a7a336
CI & Build / Python lint (push) Successful in 4s
CI & Build / Plugin hooks (push) Successful in 13s
CI & Build / TypeScript typecheck (push) Successful in 55s
CI & Build / integration (push) Successful in 1m9s
CI & Build / Python tests (push) Successful in 1m55s
CI & Build / Build & push image (push) Successful in 1m36s
export_full_backup orders every query by primary key: an UPDATE moves a row to the end of the heap, so an unordered export shifted every byte after an edited row. services/backup_file.py writes the payload as one dated tar per run - manifest.json plus <section>.jsonl, sorted keys, fixed member headers - atomically via a dot-prefixed .part file, and reads it back into the dict restore_full_backup takes, refusing a missing or short section. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
"""Real-Postgres check that the full export is ORDERED, so the backup file
|
|
deduplicates between runs (milestone 523).
|
|
|
|
Only Postgres shows this. An unordered select returns rows in heap order, and
|
|
an UPDATE writes the row's new version at the end of the heap — so before
|
|
export_full_backup ordered its queries, editing one old note moved it to the
|
|
bottom of `notes`. Every byte after its old place shifted, and a restic run
|
|
over the next backup stored most of the table again. A mocked session returns
|
|
whatever order it was handed and could never catch it.
|
|
"""
|
|
import io
|
|
import tarfile
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import select
|
|
|
|
from scribe.models import async_session
|
|
from scribe.models.note import Note
|
|
from scribe.models.user import User
|
|
from scribe.services import backup
|
|
from scribe.services.backup_file import write_archive_to
|
|
from tests.helpers import ensure_user
|
|
|
|
pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
|
|
|
|
USERNAME = "backup_file_ordering_owner"
|
|
|
|
|
|
async def _purge() -> None:
|
|
async with async_session() as s:
|
|
for user in (await s.execute(
|
|
select(User).where(User.username == USERNAME)
|
|
)).scalars().all():
|
|
for note in (await s.execute(
|
|
select(Note).where(Note.user_id == user.id)
|
|
)).scalars().all():
|
|
await s.delete(note)
|
|
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 a teardown
|
|
here would break the next test's event loop."""
|
|
await _purge()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def notes():
|
|
async with async_session() as s:
|
|
user = await ensure_user(s, USERNAME)
|
|
rows = [Note(user_id=user.id, title=f"ordering {n}", body="b" * 100)
|
|
for n in range(5)]
|
|
s.add_all(rows)
|
|
await s.commit()
|
|
ids = [n.id for n in rows]
|
|
yield ids
|
|
await _purge()
|
|
|
|
|
|
def _sections(payload: dict) -> dict[str, bytes]:
|
|
buf = io.BytesIO()
|
|
write_archive_to(payload, buf, app_version="test")
|
|
buf.seek(0)
|
|
with tarfile.open(fileobj=buf, mode="r:") as tar:
|
|
return {m.name: tar.extractfile(m).read() for m in tar.getmembers()
|
|
if m.name != "manifest.json"}
|
|
|
|
|
|
async def _edit_first(ids: list[int]) -> None:
|
|
async with async_session() as s:
|
|
note = await s.get(Note, ids[0])
|
|
note.body = "edited after the first export"
|
|
await s.commit()
|
|
|
|
|
|
async def test_an_edited_row_keeps_its_place(notes):
|
|
await _edit_first(notes)
|
|
payload = await backup.export_full_backup()
|
|
exported = [row["id"] for row in payload["notes"]]
|
|
assert exported == sorted(exported)
|
|
mine = [i for i in exported if i in set(notes)]
|
|
assert mine == notes
|
|
|
|
|
|
async def test_two_exports_of_unchanged_data_are_the_same_bytes(notes):
|
|
await _edit_first(notes)
|
|
first = _sections(await backup.export_full_backup())
|
|
second = _sections(await backup.export_full_backup())
|
|
assert first.keys() == second.keys()
|
|
for name in first:
|
|
assert first[name] == second[name], name
|