219 lines
10 KiB
Python
219 lines
10 KiB
Python
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()
|