tegwick
54eef1c4b5
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 6m41s
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
174 lines
8.5 KiB
Python
174 lines
8.5 KiB
Python
"""Client offboarding export uses synthetic tenants and never contacts providers."""
|
|
|
|
import copy
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import zipfile
|
|
|
|
import pytest
|
|
from django.contrib.admin.models import ADDITION, LogEntry
|
|
from django.contrib.auth.models import Group, Permission
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.contrib.sessions.models import Session
|
|
from django.core.management import call_command
|
|
from django.core.management.base import CommandError
|
|
|
|
from coulomb_social.apps.members import client_export
|
|
from coulomb_social.apps.members.client_export import ExportRefused, export_client_dataset
|
|
from coulomb_social.apps.members.models import Member, User
|
|
from coulomb_social.apps.spaces.models import Space, SpaceMembership
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def actor(tenant, name):
|
|
user = User.objects.create_user(username=name, email=f"{name}@example.test")
|
|
return Member.objects.create(user=user, tenant_id=tenant, issuer="https://issuer.example.test",
|
|
subject=name, user_engine_user_id=name)
|
|
|
|
|
|
@pytest.fixture
|
|
def dataset(tmp_path):
|
|
own, other = actor("tenant:a", "own"), actor("tenant:b", "other")
|
|
selected = Space.objects.create(tenant_id=own.tenant_id, slug="own", created_by=own)
|
|
foreign = Space.objects.create(tenant_id=other.tenant_id, slug="other", created_by=other)
|
|
SpaceMembership.objects.create(space=selected, member=own, role="owner")
|
|
SpaceMembership.objects.create(space=foreign, member=other, role="owner")
|
|
root = tmp_path / "content"
|
|
root.mkdir()
|
|
files = {
|
|
"spaces/own/pages/index.md": b"Private client A body",
|
|
"spaces/own/assets/image.bin": b"client A image",
|
|
"spaces/own/.page-aliases.json": b'{"prior":["own","index"]}',
|
|
".revisions/own/deleted/old.md": b"client A deleted page recovery bytes",
|
|
"spaces/other/pages/index.md": b"Other tenant private body",
|
|
".revisions/other/deleted/old.md": b"Other tenant deleted bytes",
|
|
".writer.lock": b"",
|
|
}
|
|
for name, value in files.items():
|
|
path = root / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(value)
|
|
private = tmp_path / "private"
|
|
private.mkdir(mode=0o700)
|
|
return root, private / "export.zip", selected, own, other
|
|
|
|
|
|
def test_export_includes_complete_own_local_data_and_revisions_without_other_tenant_or_credentials(
|
|
dataset):
|
|
root, output, space, own, other = dataset
|
|
permission = Permission.objects.get(content_type__app_label="spaces", codename="change_space")
|
|
group = Group.objects.create(name="Synthetic client operators")
|
|
group.permissions.add(permission)
|
|
own.user.groups.add(group)
|
|
own.user.user_permissions.add(permission)
|
|
LogEntry.objects.create(user=own.user, content_type=ContentType.objects.get_for_model(Space),
|
|
object_id=str(space.pk), object_repr="own client title",
|
|
action_flag=ADDITION, change_message="own change")
|
|
Session.objects.create(session_key="must-not-export", session_data="synthetic-session-secret",
|
|
expire_date="2030-01-01T00:00:00Z")
|
|
before = {p.relative_to(root).as_posix(): p.read_bytes()
|
|
for p in root.rglob("*") if p.is_file()}
|
|
result = export_client_dataset("tenant:a", root, output)
|
|
assert result["content_files"] == 4 and result["admin_events"] == 1
|
|
assert result["records"] == {"members.user": 1, "members.member": 1,
|
|
"spaces.space": 1, "spaces.spacemembership": 1}
|
|
assert output.stat().st_mode & 0o777 == 0o600
|
|
with zipfile.ZipFile(output) as archive:
|
|
manifest = json.loads(archive.read("manifest.json"))
|
|
assert len(archive.namelist()) == 5
|
|
for name, digest in manifest["files_sha256"].items():
|
|
assert hashlib.sha256(archive.read("content/" + name)).hexdigest() == digest
|
|
assert archive.read("content/.revisions/own/deleted/old.md") == before[
|
|
".revisions/own/deleted/old.md"]
|
|
serialized = json.dumps(manifest)
|
|
assert "other@example.test" not in serialized
|
|
assert str(other.pk) not in serialized and str(other.user_id) not in serialized
|
|
assert "password" not in manifest["data"]["records"]["members.user"][0]
|
|
assert "synthetic-session-secret" not in serialized and "must-not-export" not in serialized
|
|
access = manifest["data"]["user_authorization"][0]
|
|
assert access["groups"][0]["permissions"][0]["codename"] == "change_space"
|
|
assert access["direct_permissions"][0]["codename"] == "change_space"
|
|
assert {p.relative_to(root).as_posix(): p.read_bytes()
|
|
for p in root.rglob("*") if p.is_file()} == before
|
|
assert Member.objects.count() == 2 and Space.objects.count() == 2
|
|
assert not result["offboarding_complete"]
|
|
with pytest.raises(ExportRefused):
|
|
export_client_dataset("tenant:a", root, output)
|
|
|
|
|
|
@pytest.mark.parametrize("problem", ["foreign_grant", "foreign_creator", "pending_journal",
|
|
"orphan_file", "symlink", "usable_password", "remote_only"])
|
|
def test_ambiguous_or_incomplete_source_refuses_without_publishing(dataset, problem):
|
|
root, output, space, own, other = dataset
|
|
if problem == "foreign_grant":
|
|
SpaceMembership.objects.create(space=space, member=other)
|
|
elif problem == "foreign_creator":
|
|
Space.objects.filter(pk=space.pk).update(created_by=other)
|
|
elif problem == "pending_journal":
|
|
(root / ".recovery").mkdir()
|
|
(root / ".recovery/interrupted.json").write_text("{}")
|
|
elif problem == "orphan_file":
|
|
(root / "unattributed.bin").write_bytes(b"unknown owner")
|
|
elif problem == "symlink":
|
|
(root / "spaces/own/assets/link").symlink_to(root / "spaces/other/pages/index.md")
|
|
elif problem == "usable_password":
|
|
own.user.set_password("synthetic-not-an-app-credential")
|
|
own.user.save()
|
|
else:
|
|
Space.objects.create(tenant_id="tenant:a", slug="external", forgejo_owner="synthetic",
|
|
forgejo_repo="remote-only")
|
|
with pytest.raises(ExportRefused):
|
|
export_client_dataset("tenant:a", root, output)
|
|
assert not output.exists() and not list(output.parent.glob(".client-export-*"))
|
|
|
|
|
|
def test_content_drift_refuses_and_removes_private_partial(dataset, monkeypatch):
|
|
root, output, *_ = dataset
|
|
write = zipfile.ZipFile.write
|
|
|
|
def changed(archive, *args, **kwargs):
|
|
result = write(archive, *args, **kwargs)
|
|
(root / "spaces/own/pages/index.md").write_bytes(b"later edit")
|
|
return result
|
|
|
|
monkeypatch.setattr(zipfile.ZipFile, "write", changed)
|
|
with pytest.raises(ExportRefused, match="changed during export"):
|
|
export_client_dataset("tenant:a", root, output)
|
|
assert not output.exists() and not list(output.parent.glob(".client-export-*"))
|
|
|
|
|
|
def test_deleted_admin_target_cannot_be_silently_omitted(dataset):
|
|
root, output, space, own, _ = dataset
|
|
LogEntry.objects.create(user=own.user, content_type=ContentType.objects.get_for_model(Space),
|
|
object_id="00000000-0000-0000-0000-000000000001",
|
|
object_repr="deleted client data", action_flag=ADDITION)
|
|
with pytest.raises(ExportRefused, match="admin history"):
|
|
export_client_dataset("tenant:a", root, output)
|
|
assert not output.exists()
|
|
|
|
|
|
def test_inventory_exposes_schema_classification_and_refuses_unclassified_field(monkeypatch):
|
|
inventory = client_export.schema_inventory()
|
|
assert len(inventory["models"]) == 4
|
|
fields = copy.deepcopy(client_export.FIELDS)
|
|
fields["members.user"]["identity"] = "id username full_name" # new/unclassified email
|
|
monkeypatch.setattr(client_export, "FIELDS", fields)
|
|
with pytest.raises(ExportRefused, match="Unclassified application field"):
|
|
client_export.schema_inventory()
|
|
|
|
|
|
def test_command_inventory_is_value_free_and_export_requires_operator_source_assertion(
|
|
dataset, settings):
|
|
root, output, *_ = dataset
|
|
settings.CONTENT_ROOT = str(root)
|
|
stream = io.StringIO()
|
|
call_command("export_client_dataset", inventory=True, stdout=stream)
|
|
assert "own@example.test" not in stream.getvalue()
|
|
assert json.loads(stream.getvalue())["schema"] == "coulomb.client-data-inventory.v1"
|
|
with pytest.raises(CommandError):
|
|
call_command("export_client_dataset", tenant="tenant:a", output=output)
|
|
call_command("export_client_dataset", tenant="tenant:a", output=output,
|
|
quiesced_source=True, stdout=io.StringIO())
|
|
assert output.exists()
|