tegwick
dd83c90d73
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 6m43s
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a119cf-56d0-7fa3-840b-eb8732e1db0b
285 lines
13 KiB
Python
285 lines
13 KiB
Python
"""Conflict/failure and attachment fidelity checks with synthetic local content."""
|
|
|
|
import subprocess
|
|
import sys
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
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, save_space_page
|
|
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 test_simultaneous_edits_have_one_winner_and_durable_old_and_new_versions(trees):
|
|
before = pageops.write_page("source", "notes", title="Notes", body="Original")
|
|
original = Path(before.path).read_bytes()
|
|
|
|
def save(body):
|
|
try:
|
|
pageops.write_page("source", "notes", title="Notes", body=body,
|
|
expected_revision=before.revision)
|
|
return "saved"
|
|
except pageops.PageOpsError:
|
|
return "conflict"
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
assert sorted(pool.map(save, ("First edit", "Second edit"))) == ["conflict", "saved"]
|
|
versions = [p.read_bytes() for p in (trees / ".revisions/source/notes").glob("*.md")]
|
|
assert original in versions
|
|
assert Path(before.path).read_bytes() in versions
|
|
assert len(versions) == 2
|
|
pageops.delete_page("source", "notes")
|
|
assert not Path(before.path).exists()
|
|
assert original in versions
|
|
|
|
|
|
def test_failed_atomic_publish_leaves_original_readable_and_history_retained(trees, monkeypatch):
|
|
doc = pageops.write_page("source", "notes", title="Notes", body="Original")
|
|
original = Path(doc.path).read_bytes()
|
|
replace = pageops.os.replace
|
|
|
|
def fail_page(source, destination):
|
|
if Path(destination) == Path(doc.path):
|
|
raise OSError("synthetic disk failure")
|
|
return replace(source, destination)
|
|
|
|
monkeypatch.setattr(pageops.os, "replace", fail_page)
|
|
with pytest.raises(OSError):
|
|
pageops.write_page("source", "notes", title="Notes", body="New")
|
|
assert Path(doc.path).read_bytes() == original
|
|
assert sorted(p.name for p in Path(doc.path).parent.iterdir()) == ["index.md", "notes.md"]
|
|
|
|
|
|
def test_empty_imported_page_has_a_real_revision_and_can_be_edited(trees):
|
|
pageops.page_file("source", "empty").write_bytes(b"")
|
|
source = pageops.read_page("source", "empty")
|
|
assert source.exists and source.revision
|
|
pageops.write_page("source", "empty", body="New notes", expected_revision=source.revision)
|
|
assert pageops.read_page("source", "empty").body == "New notes\n"
|
|
assert any(p.read_bytes() == b"" for p in (trees / ".revisions/source/empty").glob("*.md"))
|
|
|
|
|
|
@pytest.mark.parametrize("operation", ["copy", "move"])
|
|
def test_body_assets_reference_links_visual_and_nested_equal_names_survive(trees, operation):
|
|
assets = pageops.assets_dir("source")
|
|
(assets / "nested").mkdir()
|
|
(assets / "diagram.png").write_bytes(b"synthetic image one")
|
|
(assets / "nested/diagram.png").write_bytes(b"synthetic image two")
|
|
body = ('\n'
|
|
'![Nested][picture]\n\n[picture]: assets/nested/diagram.png\n'
|
|
'[File](assets/diagram.png#detail)\n')
|
|
source = pageops.write_page("source", "notes", title="Notes", body=body,
|
|
visual="assets/diagram.png")
|
|
original = Path(source.path).read_bytes()
|
|
if operation == "copy":
|
|
doc = pageops.copy_page("source", "notes", "destination", "new-notes")
|
|
assert Path(source.path).read_bytes() == original
|
|
else:
|
|
doc = pageops.transfer_page("source", "notes", "destination", "new-notes")
|
|
assert not Path(source.path).exists()
|
|
assert doc.visual == "assets/new-notes-diagram.png"
|
|
assert "" in doc.body
|
|
assert "[File](assets/new-notes-diagram.png#detail)" in doc.body
|
|
assert "assets/nested/" not in doc.body
|
|
assert sorted(p.read_bytes() for p in pageops.assets_dir("destination").iterdir()) == [
|
|
b"synthetic image one", b"synthetic image two",
|
|
]
|
|
|
|
|
|
def test_attachment_failure_rolls_back_partial_copy_and_does_not_move_source(trees):
|
|
(pageops.assets_dir("source") / "present.png").write_bytes(b"synthetic image")
|
|
source = pageops.write_page("source", "notes", title="Notes",
|
|
body="\n")
|
|
before = Path(source.path).read_bytes()
|
|
with pytest.raises(pageops.PageOpsError, match="attachment safely"):
|
|
pageops.transfer_page("source", "notes", "destination", "notes")
|
|
assert Path(source.path).read_bytes() == before
|
|
assert list(pageops.assets_dir("destination").iterdir()) == []
|
|
assert not pageops.read_page("destination", "notes").exists
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_stale_rename_and_missing_revision_preserve_user_input_and_current_page(trees, client):
|
|
owner = make_member("owner")
|
|
space = create_space(owner, title="Conflict", slug="conflict")
|
|
before = pageops.write_page(space.slug, "notes", title="Notes", body="Original")
|
|
pageops.write_page(space.slug, "notes", title="Notes", body="Other writer")
|
|
client.force_login(owner.user)
|
|
url = reverse("spaces:page_edit", args=[space.slug, "notes"])
|
|
for revision in (None, before.revision):
|
|
fields = {"title": "Mine", "slug": "renamed", "body": "My unsaved notes"}
|
|
if revision is not None:
|
|
fields["revision"] = revision
|
|
response = client.post(url, fields)
|
|
assert response.status_code == 200
|
|
assert b"Reopen it and merge" in response.content
|
|
assert b"My unsaved notes" in response.content
|
|
assert pageops.read_page(space.slug, "notes").body == "Other writer\n"
|
|
assert not pageops.read_page(space.slug, "renamed").exists
|
|
current = pageops.read_page(space.slug, "notes")
|
|
save_space_page(owner, space, "notes", "renamed", expected_revision=current.revision,
|
|
title="Merged", body="Other writer plus my notes")
|
|
assert not pageops.read_page(space.slug, "notes").exists
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_space_edit_conflict_does_not_overwrite_new_index_body_or_metadata(trees, client):
|
|
owner = make_member("owner")
|
|
space = create_space(owner, title="Conflict", slug="conflict")
|
|
before = pageops.read_page(space.slug, "index")
|
|
pageops.write_page(space.slug, "index", title="Conflict", body="Another writer's index",
|
|
extra_meta={"source_id": "synthetic-source"})
|
|
client.force_login(owner.user)
|
|
fields = {"title": "My title", "abstractor": "My abstract", "revision": before.revision}
|
|
url = reverse("spaces:edit", args=[space.slug])
|
|
response = client.post(url, fields)
|
|
assert b"space changed" in response.content and b"My title" in response.content
|
|
space.refresh_from_db()
|
|
assert space.title == "Conflict"
|
|
current = pageops.read_page(space.slug, "index")
|
|
assert current.body == "Another writer's index\n"
|
|
fields["revision"] = current.revision
|
|
assert client.post(url, fields).status_code == 302
|
|
assert pageops.read_page(space.slug, "index").meta["source_id"] == "synthetic-source"
|
|
|
|
|
|
def test_page_symlink_cannot_read_or_overwrite_another_space(trees):
|
|
target = pageops.page_file("destination", "index")
|
|
pageops.page_file("source", "linked").symlink_to(target)
|
|
before = target.read_bytes()
|
|
with pytest.raises(pageops.PageOpsError, match="symlink"):
|
|
pageops.read_page("source", "linked")
|
|
with pytest.raises(pageops.PageOpsError, match="symlink"):
|
|
pageops.write_page("source", "linked", body="Tampered")
|
|
assert target.read_bytes() == before
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_incoming_page_addresses_survive_rename_move_and_enforce_every_grant(trees, client):
|
|
from coulomb_social.apps.spaces.services import set_space_access, transfer_space_page
|
|
|
|
owner, reader = make_member("owner"), make_member("reader")
|
|
source = create_space(owner, title="Shared", slug="shared")
|
|
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")
|
|
save_space_page(owner, source, "notes", "renamed", expected_revision=doc.revision,
|
|
title="Notes", body=doc.body)
|
|
client.force_login(reader.user)
|
|
old_url = reverse("spaces:detail", args=[source.slug]) + "?page=notes"
|
|
response = client.get(old_url, follow=True)
|
|
assert response.status_code == 200
|
|
assert b"Retained content" in response.content
|
|
transfer_space_page(owner, source, "renamed", target, dest_slug="moved")
|
|
assert client.get(old_url).status_code == 404
|
|
assert b"Private" not in client.get(old_url).content
|
|
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
|
|
assert response.redirect_chain == [
|
|
(reverse("spaces:detail", args=[target.slug]) + "?page=moved", 302),
|
|
]
|
|
with pytest.raises(pageops.PageOpsError, match="reserved"):
|
|
pageops.write_page(source.slug, "notes", body="Hijack previous address")
|
|
set_space_access(owner, source, reader.pk, "remove")
|
|
assert client.get(old_url).status_code == 404
|
|
|
|
|
|
def test_failed_destination_publish_rolls_back_assets_and_keeps_source(trees, monkeypatch):
|
|
assets = pageops.assets_dir("source")
|
|
(assets / "diagram.png").write_bytes(b"synthetic image")
|
|
source = pageops.write_page("source", "notes", body="")
|
|
before = Path(source.path).read_bytes()
|
|
original = pageops.os.replace
|
|
|
|
def fail_target(old, new):
|
|
if Path(new) == pageops.page_file("destination", "notes"):
|
|
raise OSError("synthetic destination failure")
|
|
return original(old, new)
|
|
|
|
monkeypatch.setattr(pageops.os, "replace", fail_target)
|
|
with pytest.raises(OSError):
|
|
pageops.transfer_page("source", "notes", "destination", "notes")
|
|
assert Path(source.path).read_bytes() == before
|
|
assert not pageops.read_page("destination", "notes").exists
|
|
assert list(pageops.assets_dir("destination").iterdir()) == []
|
|
assert not pageops.page_aliases("source")
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_destination_disk_failure_is_visible_without_losing_source(trees, client, monkeypatch):
|
|
owner = make_member("owner")
|
|
space = create_space(owner, title="Work", slug="work")
|
|
source = pageops.write_page(space.slug, "notes", title="Notes", body="Original notes")
|
|
original = pageops.os.replace
|
|
|
|
def fail_target(old, new):
|
|
if Path(new) == pageops.page_file(space.slug, "copy"):
|
|
raise OSError("synthetic private storage path must not be disclosed")
|
|
return original(old, new)
|
|
|
|
monkeypatch.setattr(pageops.os, "replace", fail_target)
|
|
client.force_login(owner.user)
|
|
response = client.post(reverse("spaces:page_copy", args=[space.slug, source.slug]), {
|
|
"dest_slug": "copy", "title": "Copy",
|
|
})
|
|
assert response.status_code == 200
|
|
assert b"could not be confirmed" in response.content
|
|
assert b"private storage path" not in response.content
|
|
assert pageops.read_page(space.slug, "notes").body == "Original notes\n"
|
|
assert not pageops.read_page(space.slug, "copy").exists
|
|
|
|
|
|
def test_separate_processes_share_the_writer_lock_and_reject_stale_revision(trees):
|
|
doc = pageops.write_page("source", "notes", title="Notes", body="Original")
|
|
code = '''
|
|
import sys
|
|
from django.conf import settings
|
|
settings.configure(CONTENT_ROOT=sys.argv[1])
|
|
from coulomb_social.apps.spaces import pageops
|
|
if sys.argv[3] == "hold":
|
|
with pageops.content_write_lock():
|
|
print("locked", flush=True)
|
|
sys.stdin.readline()
|
|
pageops.write_page("source", "notes", body="First process", expected_revision=sys.argv[2])
|
|
else:
|
|
print("attempting", flush=True)
|
|
try:
|
|
pageops.write_page("source", "notes", body="Second process", expected_revision=sys.argv[2])
|
|
except pageops.PageOpsError:
|
|
print("conflict", flush=True)
|
|
'''
|
|
children = []
|
|
try:
|
|
for mode in ("hold", "write"):
|
|
child = subprocess.Popen(
|
|
[sys.executable, "-c", code, str(trees), doc.revision, mode],
|
|
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
|
|
)
|
|
children.append(child)
|
|
assert child.stdout.readline().strip() == ("locked" if mode == "hold" else "attempting")
|
|
with pytest.raises(subprocess.TimeoutExpired):
|
|
children[1].communicate(timeout=0.25)
|
|
_, errors = children[0].communicate(input="release\n", timeout=10)
|
|
assert children[0].returncode == 0, errors
|
|
output, errors = children[1].communicate(timeout=10)
|
|
assert children[1].returncode == 0, errors
|
|
assert output.strip() == "conflict"
|
|
assert pageops.read_page("source", "notes").body == "First process\n"
|
|
finally:
|
|
for child in children:
|
|
if child.poll() is None:
|
|
child.terminate()
|
|
child.communicate(timeout=10)
|