tegwick
d9da608e65
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a192e9-c28b-7533-bf25-e2341364c22a
283 lines
12 KiB
Python
283 lines
12 KiB
Python
"""Revision retention and crash recovery for copy/move. Synthetic content only."""
|
|
|
|
import os
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from django.urls import reverse
|
|
|
|
from coulomb_social.apps.spaces import pageops
|
|
from coulomb_social.apps.spaces.services import create_space, set_space_access
|
|
from tests.test_space_permissions import make_member
|
|
|
|
|
|
@pytest.fixture
|
|
def trees(tmp_path, settings):
|
|
settings.CONTENT_ROOT = str(tmp_path / "content")
|
|
for slug in ("source", "destination"):
|
|
pageops.init_space_content(slug, title=slug)
|
|
return tmp_path / "content"
|
|
|
|
|
|
def _notes(body="Retained content"):
|
|
asset = pageops.assets_dir("source") / "diagram.png"
|
|
asset.write_bytes(b"synthetic image")
|
|
return pageops.write_page(
|
|
"source", "notes", title="Notes", body=f"\n\n{body}\n",
|
|
)
|
|
|
|
|
|
def test_restart_drops_unconfirmed_destination_and_keeps_source(trees):
|
|
source = _notes()
|
|
before = Path(source.path).read_bytes()
|
|
partial = pageops.assets_dir("destination") / "notes-diagram.png"
|
|
partial.write_bytes(b"synthetic image")
|
|
record = pageops._start_record(
|
|
"transfer", "source", "notes", "destination", "notes", source.revision,
|
|
)
|
|
pageops._note_asset(record, "notes-diagram.png")
|
|
pageops.write_page("source", "other", title="Other", body="Unrelated")
|
|
assert not partial.exists()
|
|
assert Path(source.path).read_bytes() == before
|
|
assert pageops.page_aliases("source") == {}
|
|
assert not pageops.read_page("destination", "notes").exists
|
|
assert not list((trees / ".recovery").glob("*.json"))
|
|
|
|
|
|
def test_restart_finishes_a_published_move_without_changing_bytes(trees):
|
|
source = _notes()
|
|
copied = pageops.copy_page(
|
|
"source", "notes", "destination", "notes", title="Notes", journal=True,
|
|
)
|
|
record = pageops._start_record(
|
|
"transfer", "source", "notes", "destination", "notes", source.revision,
|
|
)
|
|
record["dest_revision"] = copied.revision
|
|
record["phase"] = "published"
|
|
pageops._write_record(record)
|
|
result = pageops.recover_interrupted_content()
|
|
assert result == {"cleared": 1, "held": 0}
|
|
assert not Path(source.path).exists()
|
|
assert pageops.page_aliases("source")["notes"] == ["destination", "notes"]
|
|
assert pageops.read_page("destination", "notes").revision == copied.revision
|
|
assert not list((trees / ".recovery").glob("*.json"))
|
|
|
|
|
|
def test_restart_keeps_a_source_that_changed_and_drops_its_alias(trees):
|
|
source = _notes()
|
|
copied = pageops.copy_page(
|
|
"source", "notes", "destination", "notes", title="Notes", journal=True,
|
|
)
|
|
pageops.remember_page_address("source", "notes", "destination", "notes")
|
|
record = pageops._start_record(
|
|
"transfer", "source", "notes", "destination", "notes", source.revision,
|
|
)
|
|
record["dest_revision"] = copied.revision
|
|
record["phase"] = "aliased"
|
|
pageops._write_record(record)
|
|
Path(source.path).write_bytes(b"changed after the crash window\n")
|
|
pageops.recover_interrupted_content()
|
|
assert Path(source.path).read_bytes() == b"changed after the crash window\n"
|
|
assert pageops.page_aliases("source") == {}
|
|
assert pageops.read_page("destination", "notes").revision == copied.revision
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_recovered_move_still_requires_the_destination_grant(trees, client):
|
|
owner, reader = make_member("owner"), make_member("reader")
|
|
source = create_space(owner, title="Shared", slug="source")
|
|
target = create_space(owner, title="Private", slug="private")
|
|
set_space_access(owner, source, reader.pk, "viewer")
|
|
doc = pageops.write_page(source.slug, "notes", title="Notes", body="Retained content")
|
|
copied = pageops.copy_page(
|
|
source.slug, "notes", target.slug, "moved", title="Notes", journal=True,
|
|
)
|
|
record = pageops._start_record(
|
|
"transfer", source.slug, "notes", target.slug, "moved", doc.revision,
|
|
)
|
|
record["dest_revision"] = copied.revision
|
|
record["phase"] = "published"
|
|
pageops._write_record(record)
|
|
pageops.recover_interrupted_content()
|
|
client.force_login(reader.user)
|
|
old_url = reverse("spaces:detail", args=[source.slug]) + "?page=notes"
|
|
assert client.get(old_url).status_code == 404
|
|
set_space_access(owner, target, reader.pk, "viewer")
|
|
response = client.get(old_url, follow=True)
|
|
assert response.status_code == 200
|
|
assert b"Retained content" in response.content
|
|
|
|
|
|
def test_existing_destination_is_refused_without_a_journal(trees):
|
|
source = _notes()
|
|
pageops.write_page("destination", "notes", title="Keep", body="Already here")
|
|
before = pageops.page_file("destination", "notes").read_bytes()
|
|
with pytest.raises(pageops.PageOpsError, match="already exists"):
|
|
pageops.transfer_page("source", "notes", "destination", "notes")
|
|
assert pageops.page_file("destination", "notes").read_bytes() == before
|
|
assert Path(source.path).read_bytes()
|
|
assert not list((trees / ".recovery").glob("*.json"))
|
|
|
|
|
|
def test_purge_keeps_live_bytes_and_deleted_page_history(trees):
|
|
first = pageops.write_page("source", "notes", title="Notes", body="First")
|
|
pageops.write_page(
|
|
"source", "notes", title="Notes", body="Second", expected_revision=first.revision,
|
|
)
|
|
removed = pageops.write_page("source", "gone", title="Gone", body="Recover me")
|
|
pageops.delete_page("source", "gone")
|
|
revision_dir = trees / ".revisions" / "source" / "notes"
|
|
old = next(path for path in revision_dir.glob("*.md") if path.stem == first.revision)
|
|
live = next(path for path in revision_dir.glob("*.md") if path.stem != first.revision)
|
|
deleted = next((trees / ".revisions" / "source" / "gone").glob("*.md"))
|
|
past = (datetime.now(UTC) - timedelta(days=30)).timestamp()
|
|
for path in (old, live, deleted):
|
|
os.utime(path, (past, past))
|
|
removed_count = pageops.purge_revisions(
|
|
older_than=datetime.now(UTC) - timedelta(days=7),
|
|
)
|
|
assert removed_count == 1
|
|
assert not old.exists()
|
|
assert live.exists()
|
|
assert deleted.exists()
|
|
assert pageops.read_page("source", "notes").body == "Second\n"
|
|
assert removed.body == "Recover me\n"
|
|
|
|
|
|
def test_purge_stops_before_deleting_when_a_revision_hash_does_not_match(trees):
|
|
doc = pageops.write_page("source", "notes", title="Notes", body="First")
|
|
pageops.write_page(
|
|
"source", "notes", title="Notes", body="Second", expected_revision=doc.revision,
|
|
)
|
|
stored = list((trees / ".revisions" / "source" / "notes").glob("*.md"))
|
|
stored[0].write_bytes(b"tampered")
|
|
past = datetime.now(UTC) - timedelta(days=1)
|
|
with pytest.raises(pageops.PageOpsError, match="integrity"):
|
|
pageops.purge_revisions(older_than=past)
|
|
assert len(list((trees / ".revisions" / "source" / "notes").glob("*.md"))) == 2
|
|
|
|
|
|
@pytest.mark.parametrize('operation,phase', [
|
|
(operation, phase)
|
|
for operation in ('copy_page', 'transfer_page')
|
|
for phase in ('before_asset_create', 'during_first_asset', 'during_second_asset',
|
|
'after_assets', 'after_page', 'published')
|
|
] + [('transfer_page', phase) for phase in ('aliased', 'source_deleted')])
|
|
def test_process_death_during_copy_or_move_is_recoverable(trees, operation, phase):
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
|
|
source = _notes('Retained content [attachment](assets/second.txt)')
|
|
(pageops.assets_dir('source') / 'second.txt').write_bytes(b'synthetic attachment')
|
|
(pageops.assets_dir('source') / 'managed.txt').write_bytes(b'managed attachment')
|
|
source = pageops.write_page('source', 'notes', title=source.title, body=source.body,
|
|
extra_meta={'coulomb_attachments': json.dumps([
|
|
{'name': 'Managed file', 'path': 'assets/managed.txt'}])})
|
|
original = Path(source.path).read_bytes()
|
|
child = r"""
|
|
import os, sys
|
|
from django.conf import settings
|
|
settings.configure(CONTENT_ROOT=sys.argv[1])
|
|
from coulomb_social.apps.spaces import pageops, assets
|
|
phase = sys.argv[3]
|
|
|
|
def stop():
|
|
os._exit(86)
|
|
|
|
original_open = assets.open_local_asset
|
|
def opening(*args, **kwargs):
|
|
if kwargs.get('create') and phase == 'before_asset_create':
|
|
stop()
|
|
return original_open(*args, **kwargs)
|
|
assets.open_local_asset = opening
|
|
|
|
original_copy, count = pageops.shutil.copyfileobj, 0
|
|
def copying(source, destination):
|
|
global count
|
|
count += 1
|
|
if phase == 'during_first_asset' or (phase == 'during_second_asset' and count == 2):
|
|
destination.write(source.read(4))
|
|
destination.flush()
|
|
os.fsync(destination.fileno())
|
|
stop()
|
|
original_copy(source, destination)
|
|
pageops.shutil.copyfileobj = copying
|
|
|
|
original_write = pageops.write_page
|
|
def writing(*args, **kwargs):
|
|
if phase == 'after_assets':
|
|
stop()
|
|
result = original_write(*args, **kwargs)
|
|
if phase == 'after_page':
|
|
stop()
|
|
return result
|
|
pageops.write_page = writing
|
|
|
|
original_record = pageops._write_record
|
|
def recording(record):
|
|
result = original_record(record)
|
|
if record['phase'] == phase:
|
|
stop()
|
|
return result
|
|
pageops._write_record = recording
|
|
|
|
original_delete = pageops.delete_page
|
|
def deleting(*args, **kwargs):
|
|
original_delete(*args, **kwargs)
|
|
if phase == 'source_deleted':
|
|
stop()
|
|
pageops.delete_page = deleting
|
|
getattr(pageops, sys.argv[2])('source', 'notes', 'destination', 'notes')
|
|
"""
|
|
result = subprocess.run([sys.executable, '-c', child, str(trees), operation, phase],
|
|
capture_output=True, timeout=15)
|
|
assert result.returncode == 86, result.stderr.decode()
|
|
assert pageops.recover_interrupted_content() == {'cleared': 1, 'held': 0}
|
|
published = phase in {'published', 'aliased', 'source_deleted'}
|
|
if published and operation == 'transfer_page':
|
|
assert not Path(source.path).exists()
|
|
assert pageops.page_aliases('source')['notes'] == ['destination', 'notes']
|
|
else:
|
|
assert Path(source.path).read_bytes() == original
|
|
assert pageops.page_aliases('source') == {}
|
|
assert pageops.read_page('destination', 'notes').exists == published
|
|
if not published:
|
|
assert not list(pageops.assets_dir('destination').iterdir())
|
|
copied = getattr(pageops, operation)('source', 'notes', 'destination', 'notes')
|
|
assert copied.exists
|
|
destination = pageops.assets_dir('destination')
|
|
assert (destination / 'notes-diagram.png').read_bytes() == b'synthetic image'
|
|
assert (destination / 'notes-second.txt').read_bytes() == b'synthetic attachment'
|
|
assert (destination / 'notes-managed.txt').read_bytes() == b'managed attachment'
|
|
from coulomb_social.apps.spaces.attachments import entries
|
|
assert entries(pageops.read_page('destination', 'notes').meta) == [
|
|
{'name': 'Managed file', 'path': 'assets/notes-managed.txt'}]
|
|
assert not list((trees / '.recovery').glob('*.json'))
|
|
assert pageops.recover_interrupted_content() == {'cleared': 0, 'held': 0}
|
|
|
|
|
|
@pytest.mark.parametrize('operation', ['copy_page', 'transfer_page'])
|
|
@pytest.mark.parametrize('kind', ['file', 'directory', 'symlink', 'dangling_symlink'])
|
|
def test_existing_attachment_never_becomes_recovery_cleanup_authority(trees, operation, kind):
|
|
source = _notes()
|
|
original = Path(source.path).read_bytes()
|
|
destination = pageops.assets_dir('destination') / 'notes-diagram.png'
|
|
if kind == 'file':
|
|
destination.write_bytes(b'existing attachment')
|
|
elif kind == 'directory':
|
|
destination.mkdir()
|
|
else:
|
|
target = (pageops.assets_dir('source') / 'diagram.png' if kind == 'symlink'
|
|
else trees / 'missing')
|
|
destination.symlink_to(target)
|
|
before = destination.lstat()
|
|
with pytest.raises(pageops.PageOpsError, match='Cannot copy'):
|
|
getattr(pageops, operation)('source', 'notes', 'destination', 'notes')
|
|
assert destination.lstat() == before
|
|
if kind == 'file':
|
|
assert destination.read_bytes() == b'existing attachment'
|
|
assert Path(source.path).read_bytes() == original
|
|
assert not pageops.read_page('destination', 'notes').exists
|
|
assert pageops.recover_interrupted_content() == {'cleared': 0, 'held': 0}
|