import io from concurrent.futures import ThreadPoolExecutor from threading import Barrier import pytest from django.contrib.auth.models import Group, Permission from django.core.exceptions import ValidationError from django.core.exceptions import PermissionDenied from django.core.files.uploadedfile import SimpleUploadedFile from django.db import connection, connections, close_old_connections from PIL import Image from apps.content.models import EditorialPage, MediaAsset, PageRevision, PublicationAudit, SiteIdentity from apps.content.services import ( PublishConflict, publish_page, publish_site, restore_page, restore_site, save_page_draft, save_site_draft, ) from apps.users.models import User pytestmark = [pytest.mark.django_db, pytest.mark.integration] def actor(name, *permissions, staff=True, superuser=False): user = User.objects.create_user(username=name, mobile=f"0912{User.objects.count():07d}", password="TestPass123!") user.is_staff = staff user.is_superuser = superuser user.save(update_fields=["is_staff", "is_superuser"]) user.user_permissions.add(*Permission.objects.filter(content_type__app_label="content", codename__in=permissions)) return user def page_fields(title="Home"): return {"title": title, "intro": "Introduction", "seo_title": title, "seo_description": "Description"} def home_sections(): return [ {"key": "hero", "section_type": "hero", "position": 0, "heading": "Welcome", "body": "Hello"}, {"key": "featured_events", "section_type": "featured_events", "position": 1, "heading": "Events", "all_link_label": "All events"}, {"key": "featured_posts", "section_type": "featured_posts", "position": 2, "heading": "Posts", "all_link_label": "All posts"}, {"key": "participation", "section_type": "participation", "position": 3, "heading": "Join", "body": "Join us"}, ] def site_fields(): return { "official_name": "Association", "short_name": "ACE", "navigation": [{"route_id": route, "label": route, "position": index, "visible": True} for index, route in enumerate(("home", "about", "events", "blog"))], "contacts": [], } def test_role_matrix_and_audit_for_page(): page = EditorialPage.objects.create(slug="home", page_kind="sections") outsider = actor("outsider", staff=False) staff = actor("staff") editor = actor("editor", "change_content_draft", "preview_content_draft") nonstaff_editor = actor("nonstaff", "change_content_draft", staff=False) publisher = actor("publisher", "change_content_draft", "preview_content_draft", "publish_content", "restore_content") for denied in (outsider, staff, nonstaff_editor): with pytest.raises(PermissionDenied): save_page_draft(denied, page, page_fields(), home_sections()) draft = save_page_draft(editor, page, page_fields(), home_sections()) assert draft.number == 1 and draft.status == "draft" with pytest.raises(PermissionDenied): publish_page(editor, draft, expected_published_id=None) published = publish_page(publisher, draft, expected_published_id=None) page.refresh_from_db() assert page.published_revision_id == published.id assert published.status == "published" and published.published_by == publisher assert list(PublicationAudit.objects.values_list("operation", flat=True)) == ["create_draft", "publish"] def test_edit_creates_revision_and_restore_preserves_history(): page = EditorialPage.objects.create(slug="home", page_kind="sections") user = actor("chief", superuser=True) first = save_page_draft(user, page, page_fields("First"), home_sections()) publish_page(user, first, expected_published_id=None) second = save_page_draft(user, page, page_fields("Second"), home_sections(), base_revision=first) publish_page(user, second, expected_published_id=first.id) restored = restore_page(user, page, first, expected_published_id=second.id) assert restored.number == 3 and restored.title == "First" assert restored.content_checksum == first.content_checksum assert PageRevision.objects.get(pk=second.pk).title == "Second" assert page.revisions.count() == 3 assert list(PublicationAudit.objects.values_list("operation", flat=True)) == ["create_draft", "publish", "update_draft", "publish", "restore"] def test_stale_publish_is_rejected_without_swapping_pointer(): page = EditorialPage.objects.create(slug="home", page_kind="sections") user = actor("chief", superuser=True) first = save_page_draft(user, page, page_fields("First"), home_sections()) second = save_page_draft(user, page, page_fields("Second"), home_sections()) publish_page(user, first, expected_published_id=None) with pytest.raises(PublishConflict): publish_page(user, second, expected_published_id=None) page.refresh_from_db() assert page.published_revision_id == first.id assert second.status == "draft" @pytest.mark.django_db(transaction=True) @pytest.mark.skipif(connection.vendor != "postgresql", reason="SQLite has no row-level select_for_update locking") def test_concurrent_publish_only_one_revision_wins(): page = EditorialPage.objects.create(slug="home", page_kind="sections") user = actor("chief", superuser=True) drafts = [save_page_draft(user, page, page_fields(title), home_sections()) for title in ("First", "Second")] barrier = Barrier(3) def attempt(revision_id): close_old_connections() barrier.wait() try: publish_page(user, PageRevision.objects.get(pk=revision_id), expected_published_id=None) return "published" except PublishConflict: return "conflict" finally: connections.close_all() with ThreadPoolExecutor(max_workers=2) as pool: futures = [pool.submit(attempt, draft.pk) for draft in drafts] barrier.wait() outcomes = [future.result(timeout=15) for future in futures] assert sorted(outcomes) == ["conflict", "published"] page.refresh_from_db() assert page.published_revision_id in {draft.pk for draft in drafts} def test_publish_rejects_incomplete_page_and_leaves_no_public_content(): page = EditorialPage.objects.create(slug="home", page_kind="sections") user = actor("chief", superuser=True) draft = save_page_draft(user, page, page_fields(), home_sections()[:-1]) with pytest.raises(ValidationError): publish_page(user, draft, expected_published_id=None) page.refresh_from_db() assert page.published_revision_id is None def test_site_draft_and_publish_require_distinct_permissions(): site = SiteIdentity.objects.create() editor = actor("editor", "change_content_draft") publisher = actor("publisher", "publish_content") draft = save_site_draft(editor, site, site_fields()) with pytest.raises(PermissionDenied): publish_site(editor, draft, expected_published_id=None) publish_site(publisher, draft, expected_published_id=None) site.refresh_from_db() assert site.published_revision_id == draft.id def test_restore_site_copies_media_reference_and_creates_new_revision(settings, tmp_path): settings.MEDIA_ROOT = tmp_path buffer = io.BytesIO() Image.new("RGB", (20, 20)).save(buffer, format="PNG") logo = MediaAsset.objects.create(file=SimpleUploadedFile("logo.png", buffer.getvalue(), content_type="image/png"), alt="Logo", source="owner", license="owned", approved=True) site = SiteIdentity.objects.create() user = actor("chief", superuser=True) fields = site_fields() | {"logo": logo, "logo_alt": "Association logo"} first = save_site_draft(user, site, fields) publish_site(user, first, expected_published_id=None) second = save_site_draft(user, site, site_fields() | {"official_name": "New name"}) publish_site(user, second, expected_published_id=first.id) restored = restore_site(user, site, first, expected_published_id=second.id) site.refresh_from_db() assert site.published_revision_id == restored.id assert restored.logo_id == logo.id and restored.number == 3 assert restored.content_checksum == first.content_checksum def test_unapproved_media_cannot_be_published(settings, tmp_path): settings.MEDIA_ROOT = tmp_path buffer = io.BytesIO() Image.new("RGB", (20, 20)).save(buffer, format="PNG") logo = MediaAsset.objects.create(file=SimpleUploadedFile("logo.png", buffer.getvalue(), content_type="image/png"), alt="Logo", source="owner", license="owned") site = SiteIdentity.objects.create() user = actor("chief", superuser=True) draft = save_site_draft(user, site, site_fields() | {"logo": logo, "logo_alt": "Association logo"}) with pytest.raises(ValidationError): publish_site(user, draft, expected_published_id=None) site.refresh_from_db() assert site.published_revision_id is None def test_groups_are_provisioned_without_assigning_accounts(): editor = Group.objects.get(name="content_editor") publisher = Group.objects.get(name="content_publisher") assert set(editor.permissions.values_list("codename", flat=True)) >= {"change_content_draft", "preview_content_draft"} assert set(publisher.permissions.values_list("codename", flat=True)) >= {"change_content_draft", "preview_content_draft", "publish_content", "restore_content"} assert editor.user_set.count() == 0 and publisher.user_set.count() == 0 def test_revision_sections_and_audit_are_immutable(): page = EditorialPage.objects.create(slug="home", page_kind="sections") user = actor("chief", superuser=True) draft = save_page_draft(user, page, page_fields(), home_sections()) draft.title = "Changed in place" with pytest.raises(ValidationError): draft.save() section = draft.sections.first() section.heading = "Changed in place" with pytest.raises(ValidationError): section.save() audit = PublicationAudit.objects.first() audit.operation = "publish" with pytest.raises(ValidationError): audit.save() published = publish_page(user, PageRevision.objects.get(pk=draft.pk), expected_published_id=None) published.status = "draft" with pytest.raises(ValidationError): published.save()