tegwick
e75d3b0f6a
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 7m14s
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
224 lines
9 KiB
Python
224 lines
9 KiB
Python
"""Synthetic corpus checkpoints, real process exits and non-overwriting retries."""
|
|
|
|
import fcntl
|
|
import io
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
from django.conf import settings
|
|
from django.core.management import call_command
|
|
from django.core.management.base import CommandError
|
|
from django.db import transaction
|
|
|
|
from coulomb_social.apps.spaces import pageops
|
|
from coulomb_social.apps.spaces.import_receipts import CorpusReceipts
|
|
from coulomb_social.apps.spaces.models import Space, SpaceMembership
|
|
from tests import test_import_corpus
|
|
|
|
pytestmark = pytest.mark.django_db(transaction=True)
|
|
corpus = test_import_corpus.corpus
|
|
|
|
|
|
def run(manifest, private, **kwargs):
|
|
output = io.StringIO()
|
|
call_command("import_content_corpus", str(manifest), receipt_dir=str(private),
|
|
stdout=output, **kwargs)
|
|
return output.getvalue()
|
|
|
|
|
|
def test_dry_run_is_write_free_and_completed_retry_preserves_edits(corpus, tmp_path):
|
|
manifest, _, member = corpus
|
|
private = tmp_path / "receipts"
|
|
run(manifest, private, dry_run=True)
|
|
assert not private.exists() and not pageops.content_root().exists()
|
|
assert "Imported 2" in run(manifest, private)
|
|
before = (private / "checkpoint.json").read_bytes()
|
|
page = pageops.space_dir("one") / "pages/index.md"
|
|
page.write_text("later local edit")
|
|
Space.objects.filter(slug="one").update(title="Local title")
|
|
SpaceMembership.objects.filter(space__slug="one", member=member).update(role="viewer")
|
|
assert "skipped 2" in run(manifest, private)
|
|
assert (private / "checkpoint.json").read_bytes() == before
|
|
assert page.read_text() == "later local edit"
|
|
assert Space.objects.get(slug="one").title == "Local title"
|
|
assert SpaceMembership.objects.get(space__slug="one", member=member).role == "viewer"
|
|
assert private.stat().st_mode & 0o777 == 0o700
|
|
assert (private / "checkpoint.json").stat().st_mode & 0o777 == 0o600
|
|
|
|
|
|
@pytest.mark.parametrize("phase", ["between", "intent", "committed"])
|
|
def test_process_death_resumes_only_at_completed_boundaries(corpus, tmp_path, phase):
|
|
manifest, _, _ = corpus
|
|
private = tmp_path / "receipts"
|
|
code = '''
|
|
import json, os, sys
|
|
from django.conf import settings
|
|
settings.DATABASES["default"] = json.loads(sys.argv[1])
|
|
settings.CONTENT_ROOT = sys.argv[2]
|
|
import django
|
|
django.setup()
|
|
from django.core.management import call_command
|
|
from coulomb_social.apps.spaces.import_receipts import CorpusReceipts
|
|
original = CorpusReceipts.save
|
|
def save(self, data):
|
|
states = [row["state"] for row in data["spaces"]]
|
|
phase = sys.argv[5]
|
|
if phase == "committed" and states == ["completed", "pending"]:
|
|
os._exit(73)
|
|
original(self, data)
|
|
if ((phase == "intent" and states == ["started", "pending"])
|
|
or (phase == "between" and states == ["completed", "pending"])):
|
|
os._exit(73)
|
|
CorpusReceipts.save = save
|
|
call_command("import_content_corpus", sys.argv[3], receipt_dir=sys.argv[4])
|
|
'''
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", code, json.dumps(settings.DATABASES["default"], default=str),
|
|
str(pageops.content_root()), str(manifest), str(private), phase],
|
|
env={**os.environ, "DJANGO_SETTINGS_MODULE": "coulomb_social.settings.test"},
|
|
capture_output=True, timeout=30)
|
|
assert result.returncode == 73, result.stderr.decode()
|
|
checkpoint = private / "checkpoint.json"
|
|
before = checkpoint.read_bytes()
|
|
assert not Space.objects.filter(slug="two").exists()
|
|
if phase == "between":
|
|
page = pageops.space_dir("one") / "pages/index.md"
|
|
page.write_text("later edit after boundary")
|
|
first_id = Space.objects.get(slug="one").pk
|
|
assert "Imported 1 space(s); skipped 1" in run(manifest, private)
|
|
assert Space.objects.get(slug="one").pk == first_id
|
|
assert page.read_text() == "later edit after boundary"
|
|
assert Space.objects.filter(slug="two").exists()
|
|
else:
|
|
assert json.loads(before)["spaces"][0]["state"] == "started"
|
|
count = Space.objects.count()
|
|
with pytest.raises(CommandError, match="Uncertain"):
|
|
run(manifest, private)
|
|
assert checkpoint.read_bytes() == before
|
|
assert Space.objects.count() == count
|
|
assert count == (1 if phase == "committed" else 0)
|
|
output = io.StringIO()
|
|
call_command("inspect_content_corpus", str(manifest), receipt_dir=str(private),
|
|
stdout=output)
|
|
report = json.loads(output.getvalue())
|
|
assert report["spaces"][0]["classification"] == (
|
|
"reserved_destination_matches_source" if phase == "committed"
|
|
else "no_sql_or_published_tree"
|
|
)
|
|
assert checkpoint.read_bytes() == before and not report["retry_authorized"]
|
|
|
|
|
|
@pytest.mark.parametrize("drift", ["source", "manifest", "identity", "content_root"])
|
|
def test_changed_inputs_refuse_replay(corpus, tmp_path, settings, drift):
|
|
manifest, entries, member = corpus
|
|
private = tmp_path / "receipts"
|
|
run(manifest, private)
|
|
checkpoint = (private / "checkpoint.json").read_bytes()
|
|
if drift == "source":
|
|
(manifest.parent / entries[0]["tree"] / "pages/index.md").write_text("changed source")
|
|
elif drift == "manifest":
|
|
manifest.write_text(manifest.read_text() + "\n")
|
|
elif drift == "identity":
|
|
entries[0]["memberships"][0]["role"] = "viewer"
|
|
manifest.write_text(json.dumps({"version": 1, "spaces": entries}))
|
|
else:
|
|
settings.CONTENT_ROOT = str(tmp_path / "other-root")
|
|
with pytest.raises(CommandError):
|
|
run(manifest, private)
|
|
assert (private / "checkpoint.json").read_bytes() == checkpoint
|
|
assert SpaceMembership.objects.filter(member=member, role="owner").count() == 2
|
|
|
|
|
|
@pytest.mark.parametrize("kind", ["database", "files", "replace", "public", "symlink"])
|
|
def test_unsafe_destination_or_receipt_refused_before_import(corpus, tmp_path, kind):
|
|
manifest, _, _ = corpus
|
|
private = tmp_path / "receipts"
|
|
if kind == "database":
|
|
Space.objects.create(tenant_id="tenant:a", slug="two", title="Existing")
|
|
elif kind == "files":
|
|
pageops.space_dir("two").mkdir(parents=True)
|
|
elif kind == "public":
|
|
private.mkdir(mode=0o755)
|
|
elif kind == "symlink":
|
|
target = tmp_path / "target"
|
|
target.mkdir(mode=0o700)
|
|
private.symlink_to(target, target_is_directory=True)
|
|
with pytest.raises(CommandError):
|
|
run(manifest, private, replace=kind == "replace")
|
|
assert not Space.objects.filter(slug="one").exists()
|
|
assert not (private / "checkpoint.json").exists()
|
|
|
|
|
|
def test_deleted_completed_identity_is_not_recreated(corpus, tmp_path):
|
|
manifest, _, _ = corpus
|
|
private = tmp_path / "receipts"
|
|
run(manifest, private)
|
|
before = (private / "checkpoint.json").read_bytes()
|
|
Space.objects.filter(slug="one").delete()
|
|
with pytest.raises(CommandError, match="identity changed"):
|
|
run(manifest, private)
|
|
assert not Space.objects.filter(slug="one").exists()
|
|
assert (private / "checkpoint.json").read_bytes() == before
|
|
|
|
|
|
def test_receipt_write_failure_leaves_uncertain_commit(corpus, tmp_path, monkeypatch):
|
|
manifest, _, _ = corpus
|
|
private = tmp_path / "receipts"
|
|
original = CorpusReceipts.save
|
|
|
|
def fail(self, data):
|
|
if data["spaces"][0]["state"] == "completed":
|
|
raise OSError("synthetic disk failure")
|
|
original(self, data)
|
|
|
|
monkeypatch.setattr(CorpusReceipts, "save", fail)
|
|
with pytest.raises(CommandError, match="uncertain"):
|
|
run(manifest, private)
|
|
assert Space.objects.filter(slug="one").exists()
|
|
assert not Space.objects.filter(slug="two").exists()
|
|
assert json.loads((private / "checkpoint.json").read_bytes())["spaces"][0]["state"] == "started"
|
|
with pytest.raises(CommandError, match="Uncertain"):
|
|
run(manifest, private)
|
|
|
|
|
|
def test_receipt_lock_refuses_another_importer(corpus, tmp_path):
|
|
manifest, _, _ = corpus
|
|
private = tmp_path / "receipts"
|
|
private.mkdir(mode=0o700)
|
|
lock = private / ".lock"
|
|
fd = os.open(lock, os.O_CREAT | os.O_RDWR, 0o600)
|
|
try:
|
|
fcntl.flock(fd, fcntl.LOCK_EX)
|
|
with pytest.raises(CommandError):
|
|
run(manifest, private)
|
|
finally:
|
|
os.close(fd)
|
|
assert not Space.objects.exists()
|
|
assert not (private / "checkpoint.json").exists()
|
|
|
|
|
|
def test_outer_database_transaction_cannot_receive_completed_receipts(corpus, tmp_path):
|
|
manifest, _, _ = corpus
|
|
private = tmp_path / "receipts"
|
|
with transaction.atomic(), pytest.raises(CommandError):
|
|
run(manifest, private)
|
|
assert not private.exists()
|
|
assert not Space.objects.exists()
|
|
|
|
|
|
def test_malformed_completed_checkpoint_is_refused(corpus, tmp_path):
|
|
manifest, _, _ = corpus
|
|
private = tmp_path / "receipts"
|
|
run(manifest, private)
|
|
path = private / "checkpoint.json"
|
|
data = json.loads(path.read_bytes())
|
|
data["spaces"][0]["space_id"] = "invalid"
|
|
path.write_text(json.dumps(data))
|
|
before = path.read_bytes()
|
|
with pytest.raises(CommandError, match="Invalid completed"):
|
|
run(manifest, private)
|
|
assert path.read_bytes() == before
|
|
assert Space.objects.count() == 2
|