tegwick
57a329949f
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a10e2d-a5ef-7e92-b03d-b1694e915680
79 lines
3.2 KiB
Python
79 lines
3.2 KiB
Python
"""Corpus transfer acceptance: preflight, identity mapping and restart policy."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from django.core.management import call_command
|
|
from django.core.management.base import CommandError
|
|
|
|
from coulomb_social.apps.members.models import Member, User
|
|
from coulomb_social.apps.spaces import pageops
|
|
from coulomb_social.apps.spaces.models import Space, SpaceMembership
|
|
|
|
|
|
@pytest.fixture
|
|
def corpus(tmp_path, settings):
|
|
settings.CONTENT_ROOT = str(tmp_path / "content")
|
|
member = Member.objects.create(
|
|
tenant_id="tenant:a", user=User.objects.create_user(username="owner"),
|
|
user_engine_user_id="owner", issuer="https://issuer.test", subject="owner",
|
|
)
|
|
entries = []
|
|
for slug in ["one", "two"]:
|
|
pages = tmp_path / "export" / slug / "pages"
|
|
pages.mkdir(parents=True)
|
|
(pages / "index.md").write_text(f"# {slug}\n")
|
|
entries.append({"tree": f"export/{slug}", "tenant": "tenant:a",
|
|
"memberships": [{"issuer": member.issuer,
|
|
"subject": member.subject, "role": "owner"}]})
|
|
manifest = tmp_path / "manifest.json"
|
|
manifest.write_text(json.dumps({"version": 1, "spaces": entries}))
|
|
return manifest, entries, member
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_corpus_dry_run_and_import(corpus):
|
|
manifest, _, member = corpus
|
|
call_command("import_content_corpus", str(manifest), dry_run=True)
|
|
assert not Space.objects.exists()
|
|
assert not pageops.content_root().exists()
|
|
call_command("import_content_corpus", str(manifest))
|
|
assert Space.objects.count() == 2
|
|
assert SpaceMembership.objects.filter(member=member, role="owner").count() == 2
|
|
assert pageops.read_page("two", "index").body == "# two\n"
|
|
with pytest.raises(CommandError, match="already exists"):
|
|
call_command("import_content_corpus", str(manifest))
|
|
call_command("import_content_corpus", str(manifest), replace=True)
|
|
assert Space.objects.count() == 2
|
|
assert SpaceMembership.objects.count() == 2
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.parametrize("failure", ["identity", "tenant", "duplicate", "owner", "missing"])
|
|
def test_entire_corpus_preflight_before_writes(corpus, failure):
|
|
manifest, entries, _ = corpus
|
|
if failure == "identity":
|
|
entries[1]["memberships"][0]["subject"] = "missing"
|
|
elif failure == "tenant":
|
|
entries[1]["tenant"] = "tenant:b"
|
|
elif failure == "duplicate":
|
|
entries[1]["slug"] = "one"
|
|
elif failure == "owner":
|
|
entries[1]["memberships"][0]["role"] = "viewer"
|
|
else:
|
|
entries[1]["tree"] = "missing"
|
|
manifest.write_text(json.dumps({"version": 1, "spaces": entries}))
|
|
with pytest.raises(CommandError):
|
|
call_command("import_content_corpus", str(manifest))
|
|
assert not Space.objects.exists()
|
|
assert not pageops.content_root().exists()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.parametrize("data", [[], {"version": 1, "spaces": []},
|
|
{"version": 1, "spaces": [42]}])
|
|
def test_invalid_manifest_shape_is_command_error(tmp_path, data):
|
|
manifest = tmp_path / "manifest.json"
|
|
manifest.write_text(json.dumps(data))
|
|
with pytest.raises(CommandError, match="preflight"):
|
|
call_command("import_content_corpus", str(manifest), dry_run=True)
|