coulomb-social/tests/test_space_slug_constraint.py

"""Global content-slug storage boundary (CSOC-WP-0013-T01)."""

import importlib
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import pytest
from django.contrib.admin.sites import AdminSite
from django.core.exceptions import ValidationError
from django.db import IntegrityError, close_old_connections, connection
from django.test import RequestFactory

from coulomb_social.apps.spaces import pageops
from coulomb_social.apps.spaces.admin import SpaceAdmin
from coulomb_social.apps.spaces.import_tree import import_space_tree
from coulomb_social.apps.spaces.models import Space
from coulomb_social.apps.spaces.services import create_space
from tests.legacy_slug import (
    GLOBAL_SLUG,
    _constraint_present,
    without_global_slug_constraint,
)
from tests.test_import_tree import _write_sample_tree
from tests.test_space_permissions import make_member

refuse_legacy_slug_collisions = importlib.import_module(
    "coulomb_social.apps.spaces.migrations.0003_space_slug_global_uniq"
).refuse_legacy_slug_collisions


class _HistoricalApps:
    def get_model(self, app_label, model_name):
        assert (app_label, model_name) == ("spaces", "Space")
        return Space


@pytest.fixture
def content_root(tmp_path, settings):
    root = tmp_path / "content"
    settings.CONTENT_ROOT = str(root)
    return root


@pytest.mark.django_db
def test_sql_rejects_cross_tenant_slug_when_model_clean_is_skipped():
    Space.objects.create(tenant_id="tenant:a", slug="shared", title="A")
    with pytest.raises(IntegrityError):
        Space.objects.bulk_create([
            Space(tenant_id="tenant:b", slug="shared", title="B"),
        ])
    assert list(Space.objects.filter(slug="shared").values_list("tenant_id", flat=True)) == [
        "tenant:a",
    ]


@pytest.mark.django_db
def test_inactive_slug_stays_reserved():
    space = Space.objects.create(
        tenant_id="tenant:a", slug="archived", title="Old", is_active=False,
    )
    member = make_member("next")
    with pytest.raises(pageops.PageOpsError, match="already exists"):
        create_space(member, title="Next", slug=space.slug)
    assert Space.objects.filter(slug="archived").count() == 1


@pytest.mark.django_db(transaction=True)
def test_parallel_inserts_keep_one_slug():
    def insert(tenant):
        close_old_connections()
        try:
            Space.objects.create(tenant_id=tenant, slug="raced", title=tenant)
            return "ok"
        except IntegrityError:
            return "refused"
        finally:
            close_old_connections()

    with ThreadPoolExecutor(max_workers=2) as pool:
        results = list(pool.map(insert, ["tenant:a", "tenant:b"]))
    assert sorted(results) == ["ok", "refused"]
    assert Space.objects.filter(slug="raced").count() == 1
    Space.objects.filter(slug="raced").delete()


@pytest.mark.django_db(transaction=True)
def test_migration_refuses_legacy_collisions_without_rewriting_rows():
    with without_global_slug_constraint():
        Space.objects.create(tenant_id="tenant:a", slug="legacy", title="A")
        Space.objects.create(tenant_id="tenant:b", slug="legacy", title="B")
        before = set(Space.objects.filter(slug="legacy").values_list("pk", "title"))
        with pytest.raises(RuntimeError, match="does not rename"):
            refuse_legacy_slug_collisions(_HistoricalApps(), None)
        assert set(Space.objects.filter(slug="legacy").values_list("pk", "title")) == before


@pytest.mark.django_db(transaction=True)
def test_constraint_returns_only_after_duplicates_are_gone():
    with without_global_slug_constraint():
        Space.objects.create(tenant_id="tenant:a", slug="pending", title="A")
        Space.objects.create(tenant_id="tenant:b", slug="pending", title="B")
        with pytest.raises(Exception):
            with connection.schema_editor() as editor:
                editor.add_constraint(Space, GLOBAL_SLUG)
        assert not _constraint_present()
        assert Space.objects.filter(slug="pending").count() == 2
    assert _constraint_present()
    assert Space.objects.filter(slug="pending").count() == 1
    with pytest.raises(IntegrityError):
        Space.objects.create(tenant_id="tenant:b", slug="pending", title="Again")
    Space.objects.filter(slug="pending").delete()


@pytest.mark.django_db
def test_import_constraint_refuses_when_precheck_misses(content_root, tmp_path, monkeypatch):
    tree = _write_sample_tree(tmp_path / "export")
    import_space_tree(tree, tenant_id="tenant:a")
    before = pageops.page_file("sample-space", "index").read_bytes()
    original = Space.objects.filter

    def hide_cross_tenant_lookup(*args, **kwargs):
        if set(kwargs) == {"slug"}:
            return original(*args, **kwargs).none()
        return original(*args, **kwargs)

    monkeypatch.setattr(Space.objects, "filter", hide_cross_tenant_lookup)
    with pytest.raises(pageops.PageOpsError, match="another tenant"):
        import_space_tree(tree, tenant_id="tenant:b", replace=True)
    monkeypatch.undo()
    assert pageops.page_file("sample-space", "index").read_bytes() == before
    assert not Space.objects.filter(tenant_id="tenant:b").exists()
    assert not list(Path(content_root).rglob(".import-*"))


@pytest.mark.django_db(transaction=True)
def test_admin_constraint_race_is_a_validation_error(content_root):
    member = make_member("owner")
    space = create_space(member, title="Owned", slug="owned")
    admin = SpaceAdmin(Space, AdminSite())
    incoming = Space(tenant_id="tenant:b", slug=space.slug, title="Raced")
    incoming.full_clean = lambda: None
    with pytest.raises(ValidationError, match="already uses that slug"):
        admin.save_model(RequestFactory().post("/"), incoming, form=None, change=False)
    assert Space.objects.filter(slug=space.slug).count() == 1
    assert pageops.page_file(space.slug, "index").is_file()