"""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()