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
217 lines
8.3 KiB
Python
217 lines
8.3 KiB
Python
"""Uncertain imports retain identity and can be inspected without repair writes."""
|
|
|
|
import fcntl
|
|
import io
|
|
import json
|
|
import os
|
|
import uuid
|
|
from datetime import UTC
|
|
|
|
import pytest
|
|
from django.core.management import call_command
|
|
from django.core.management.base import CommandError
|
|
|
|
from coulomb_social.apps.spaces import import_inspection, import_receipts, pageops
|
|
from coulomb_social.apps.spaces.import_tree import import_space_tree
|
|
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 inspect(manifest, directory):
|
|
output = io.StringIO()
|
|
call_command("inspect_content_corpus", str(manifest), receipt_dir=str(directory), stdout=output)
|
|
return json.loads(output.getvalue())
|
|
|
|
|
|
def uncertain(manifest, directory, monkeypatch):
|
|
original = import_receipts.CorpusReceipts.save
|
|
|
|
def lose_receipt(self, data):
|
|
if data["spaces"][0]["state"] == "completed":
|
|
raise OSError("synthetic lost receipt")
|
|
original(self, data)
|
|
|
|
with monkeypatch.context() as patch:
|
|
patch.setattr(import_receipts.CorpusReceipts, "save", lose_receipt)
|
|
with pytest.raises(CommandError):
|
|
call_command(
|
|
"import_content_corpus",
|
|
str(manifest),
|
|
receipt_dir=str(directory),
|
|
stdout=io.StringIO(),
|
|
)
|
|
|
|
|
|
def test_committed_without_receipt_is_classified_without_writes(corpus, tmp_path, monkeypatch):
|
|
manifest, _, _ = corpus
|
|
directory = tmp_path / "receipts"
|
|
uncertain(manifest, directory, monkeypatch)
|
|
checkpoint = (directory / "checkpoint.json").read_bytes()
|
|
rows = list(Space.objects.values())
|
|
grants = list(SpaceMembership.objects.values())
|
|
files = {str(p): p.read_bytes() for p in pageops.content_root().rglob("*") if p.is_file()}
|
|
result = inspect(manifest, directory)
|
|
assert result["spaces"][0]["classification"] == "reserved_destination_matches_source"
|
|
assert result["spaces"][1]["classification"] == "pending"
|
|
assert (
|
|
result["read_only"]
|
|
and not result["retry_authorized"]
|
|
and not result["checkpoint_completed"]
|
|
)
|
|
assert not result["external_writer_fence_verified"]
|
|
assert (directory / "checkpoint.json").read_bytes() == checkpoint
|
|
assert list(Space.objects.values()) == rows and list(SpaceMembership.objects.values()) == grants
|
|
assert {
|
|
str(p): p.read_bytes() for p in pageops.content_root().rglob("*") if p.is_file()
|
|
} == files
|
|
assert result == inspect(manifest, directory)
|
|
assert str(Space.objects.get().pk) == json.loads(checkpoint)["spaces"][0]["space_id"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"change", ["body", "metadata", "grant", "identity", "legacy", "no_files", "no_sql"]
|
|
)
|
|
def test_inspection_distinguishes_later_changes_partial_and_unbound_states(
|
|
corpus, tmp_path, monkeypatch, change
|
|
):
|
|
manifest, _, member = corpus
|
|
directory = tmp_path / "receipts"
|
|
uncertain(manifest, directory, monkeypatch)
|
|
path = pageops.space_dir("one") / "pages/index.md"
|
|
if change == "body":
|
|
path.write_text("later local edit")
|
|
if change == "metadata":
|
|
Space.objects.filter(slug="one").update(title="later title")
|
|
if change == "grant":
|
|
SpaceMembership.objects.filter(member=member).update(role="viewer")
|
|
if change == "identity":
|
|
Space.objects.get(slug="one").delete()
|
|
Space.objects.create(slug="one", tenant_id="tenant:a", title="one")
|
|
if change == "legacy":
|
|
checkpoint = directory / "checkpoint.json"
|
|
data = json.loads(checkpoint.read_bytes())
|
|
data["spaces"][0].pop("space_id")
|
|
checkpoint.write_text(json.dumps(data))
|
|
if change == "no_files":
|
|
import shutil
|
|
|
|
shutil.rmtree(pageops.space_dir("one"))
|
|
if change == "no_sql":
|
|
Space.objects.get(slug="one").delete()
|
|
before = (directory / "checkpoint.json").read_bytes()
|
|
result = inspect(manifest, directory)
|
|
expected = {
|
|
"identity": "foreign_destination_identity",
|
|
"legacy": "started_identity_unbound",
|
|
"no_files": "sql_without_files",
|
|
"no_sql": "files_without_sql",
|
|
}.get(change, "reserved_destination_differs")
|
|
assert result["spaces"][0]["classification"] == expected
|
|
assert (directory / "checkpoint.json").read_bytes() == before
|
|
|
|
|
|
def test_reserved_id_creation_refuses_occupied_slug_without_updating_row(corpus):
|
|
manifest, entries, member = corpus
|
|
source = manifest.parent / entries[0]["tree"]
|
|
original = Space.objects.create
|
|
reserved = uuid.uuid4()
|
|
# Insert outside the import transaction, at the entry to the importer. It
|
|
# must not adopt/update the already occupied slug even with no content tree.
|
|
original(slug="one", tenant_id="tenant:a", title="concurrent owner row")
|
|
before = list(Space.objects.values())
|
|
with pytest.raises(pageops.PageOpsError):
|
|
import_space_tree(
|
|
source, tenant_id="tenant:a", memberships=[(member, "owner")], new_space_id=reserved
|
|
)
|
|
assert list(Space.objects.values()) == before and not SpaceMembership.objects.exists()
|
|
assert not pageops.space_dir("one").exists()
|
|
|
|
|
|
def test_row_created_after_corpus_preflight_is_not_adopted(corpus, tmp_path, monkeypatch):
|
|
manifest, _, _ = corpus
|
|
original = import_receipts.import_space_tree
|
|
|
|
def race(*args, **kwargs):
|
|
Space.objects.create(slug="one", tenant_id="tenant:a", title="concurrent row")
|
|
return original(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(import_receipts, "import_space_tree", race)
|
|
directory = tmp_path / "receipts"
|
|
with pytest.raises(CommandError):
|
|
call_command(
|
|
"import_content_corpus", str(manifest), receipt_dir=str(directory), stdout=io.StringIO()
|
|
)
|
|
assert Space.objects.get().title == "concurrent row" and not SpaceMembership.objects.exists()
|
|
assert not pageops.space_dir("one").exists()
|
|
assert (
|
|
inspect(manifest, directory)["spaces"][0]["classification"]
|
|
== "foreign_destination_identity"
|
|
)
|
|
|
|
|
|
def test_source_original_date_is_compared(corpus, tmp_path, monkeypatch):
|
|
from datetime import datetime
|
|
|
|
manifest, entries, _ = corpus
|
|
source = manifest.parent / entries[0]["tree"] / "pages/index.md"
|
|
source.write_text(
|
|
"---\ntitle: Original\nsource_created_at: '2020-01-02T03:04:05+00:00'\n---\nBody\n"
|
|
)
|
|
directory = tmp_path / "receipts"
|
|
uncertain(manifest, directory, monkeypatch)
|
|
result = inspect(manifest, directory)
|
|
assert result["spaces"][0]["original_space_date_checked"]
|
|
assert result["spaces"][0]["classification"] == "reserved_destination_matches_source"
|
|
Space.objects.filter(slug="one").update(created_at=datetime(2021, 1, 1, tzinfo=UTC))
|
|
assert (
|
|
inspect(manifest, directory)["spaces"][0]["classification"]
|
|
== "reserved_destination_differs"
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("lock_name", ["receipt", "content"])
|
|
def test_inspection_refuses_active_writers(corpus, tmp_path, monkeypatch, lock_name):
|
|
manifest, _, _ = corpus
|
|
directory = tmp_path / "receipts"
|
|
uncertain(manifest, directory, monkeypatch)
|
|
path = (
|
|
directory / ".lock" if lock_name == "receipt" else pageops.content_root() / ".writer.lock"
|
|
)
|
|
fd = os.open(path, os.O_RDONLY)
|
|
try:
|
|
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
with pytest.raises(CommandError):
|
|
inspect(manifest, directory)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def test_no_inspection_directory_or_locks_are_created(corpus, tmp_path):
|
|
manifest, _, _ = corpus
|
|
directory = tmp_path / "absent-receipts"
|
|
with pytest.raises(CommandError):
|
|
inspect(manifest, directory)
|
|
assert not directory.exists() and not pageops.content_root().exists()
|
|
|
|
|
|
def test_inspection_refuses_drift_during_comparison(corpus, tmp_path, monkeypatch):
|
|
manifest, _, _ = corpus
|
|
directory = tmp_path / "receipts"
|
|
uncertain(manifest, directory, monkeypatch)
|
|
original = import_inspection.observe_unit
|
|
calls = 0
|
|
|
|
def observe(*args):
|
|
nonlocal calls
|
|
result = original(*args)
|
|
calls += 1
|
|
if calls == 2:
|
|
(pageops.space_dir("one") / "pages/index.md").write_text("external writer")
|
|
return result
|
|
|
|
monkeypatch.setattr(import_inspection, "observe_unit", observe)
|
|
with pytest.raises(CommandError):
|
|
inspect(manifest, directory)
|