Add guarded content draft and publication services
This commit is contained in:
17
apps/content/migrations/0002_alter_siteidentity_options.py
Normal file
17
apps/content/migrations/0002_alter_siteidentity_options.py
Normal file
@@ -0,0 +1,17 @@
|
|||||||
|
# Generated by Django 5.2.5 on 2026-09-27 06:15
|
||||||
|
|
||||||
|
from django.db import migrations
|
||||||
|
|
||||||
|
|
||||||
|
class Migration(migrations.Migration):
|
||||||
|
|
||||||
|
dependencies = [
|
||||||
|
('content', '0001_initial'),
|
||||||
|
]
|
||||||
|
|
||||||
|
operations = [
|
||||||
|
migrations.AlterModelOptions(
|
||||||
|
name='siteidentity',
|
||||||
|
options={'permissions': [('change_content_draft', 'Can change content drafts'), ('preview_content_draft', 'Can preview content drafts'), ('publish_content', 'Can publish content'), ('restore_content', 'Can restore published content')]},
|
||||||
|
),
|
||||||
|
]
|
||||||
34
apps/content/migrations/0003_content_groups.py
Normal file
34
apps/content/migrations/0003_content_groups.py
Normal file
@@ -0,0 +1,34 @@
|
|||||||
|
from django.db import migrations
|
||||||
|
|
||||||
|
|
||||||
|
def create_groups(apps, schema_editor):
|
||||||
|
ContentType = apps.get_model("contenttypes", "ContentType")
|
||||||
|
Permission = apps.get_model("auth", "Permission")
|
||||||
|
Group = apps.get_model("auth", "Group")
|
||||||
|
content_type, _ = ContentType.objects.using(schema_editor.connection.alias).get_or_create(app_label="content", model="siteidentity")
|
||||||
|
labels = {
|
||||||
|
"change_content_draft": "Can change content drafts",
|
||||||
|
"preview_content_draft": "Can preview content drafts",
|
||||||
|
"publish_content": "Can publish content",
|
||||||
|
"restore_content": "Can restore published content",
|
||||||
|
}
|
||||||
|
permissions = {}
|
||||||
|
for codename, name in labels.items():
|
||||||
|
permission, _ = Permission.objects.using(schema_editor.connection.alias).get_or_create(content_type=content_type, codename=codename, defaults={"name": name})
|
||||||
|
permissions[codename] = permission
|
||||||
|
for group_name, codenames in {
|
||||||
|
"content_editor": ("change_content_draft", "preview_content_draft"),
|
||||||
|
"content_publisher": tuple(labels),
|
||||||
|
}.items():
|
||||||
|
group, _ = Group.objects.using(schema_editor.connection.alias).get_or_create(name=group_name)
|
||||||
|
group.permissions.add(*(permissions[code] for code in codenames))
|
||||||
|
|
||||||
|
|
||||||
|
class Migration(migrations.Migration):
|
||||||
|
dependencies = [
|
||||||
|
("content", "0002_alter_siteidentity_options"),
|
||||||
|
("auth", "0012_alter_user_first_name_max_length"),
|
||||||
|
("contenttypes", "0002_remove_content_type_name"),
|
||||||
|
]
|
||||||
|
|
||||||
|
operations = [migrations.RunPython(create_groups, migrations.RunPython.noop)]
|
||||||
@@ -70,6 +70,12 @@ class SiteIdentity(models.Model):
|
|||||||
|
|
||||||
class Meta:
|
class Meta:
|
||||||
constraints = [models.CheckConstraint(condition=Q(id=1), name="content_single_site")]
|
constraints = [models.CheckConstraint(condition=Q(id=1), name="content_single_site")]
|
||||||
|
permissions = [
|
||||||
|
("change_content_draft", "Can change content drafts"),
|
||||||
|
("preview_content_draft", "Can preview content drafts"),
|
||||||
|
("publish_content", "Can publish content"),
|
||||||
|
("restore_content", "Can restore published content"),
|
||||||
|
]
|
||||||
|
|
||||||
def clean(self):
|
def clean(self):
|
||||||
if self.published_revision_id and self.published_revision.site_id != self.pk:
|
if self.published_revision_id and self.published_revision.site_id != self.pk:
|
||||||
@@ -114,6 +120,14 @@ class SiteRevision(models.Model):
|
|||||||
if not self.logo_id and self.logo_alt:
|
if not self.logo_id and self.logo_alt:
|
||||||
raise ValidationError({"logo_alt": "Alt text requires a logo."})
|
raise ValidationError({"logo_alt": "Alt text requires a logo."})
|
||||||
|
|
||||||
|
def save(self, *args, **kwargs):
|
||||||
|
if self.pk:
|
||||||
|
previous = type(self).objects.get(pk=self.pk)
|
||||||
|
protected = ("site_id", "number", "official_name", "short_name", "faculty_subline", "description", "logo_id", "logo_alt", "navigation", "contacts", "footer_copy", "seo_title", "seo_description", "seo_image_id", "created_by_id", "source_seed_key", "source_seed_version", "content_checksum")
|
||||||
|
if any(getattr(self, field) != getattr(previous, field) for field in protected):
|
||||||
|
raise ValidationError("Site revision content is immutable.")
|
||||||
|
return super().save(*args, **kwargs)
|
||||||
|
|
||||||
|
|
||||||
class EditorialPage(models.Model):
|
class EditorialPage(models.Model):
|
||||||
slug = models.CharField(max_length=20, unique=True)
|
slug = models.CharField(max_length=20, unique=True)
|
||||||
@@ -155,6 +169,16 @@ class PageRevision(models.Model):
|
|||||||
for field in ("title", "intro", "seo_title", "seo_description"):
|
for field in ("title", "intro", "seo_title", "seo_description"):
|
||||||
validators.text(getattr(self, field), self._meta.get_field(field).max_length, required=field != "intro")
|
validators.text(getattr(self, field), self._meta.get_field(field).max_length, required=field != "intro")
|
||||||
|
|
||||||
|
def save(self, *args, **kwargs):
|
||||||
|
if self.pk:
|
||||||
|
previous = type(self).objects.get(pk=self.pk)
|
||||||
|
protected = ("page_id", "number", "title", "intro", "seo_title", "seo_description", "seo_image_id", "created_by_id", "source_seed_key", "source_seed_version")
|
||||||
|
if any(getattr(self, field) != getattr(previous, field) for field in protected):
|
||||||
|
raise ValidationError("Page revision content is immutable.")
|
||||||
|
if self.content_checksum != previous.content_checksum and previous.content_checksum:
|
||||||
|
raise ValidationError("Page revision checksum is immutable.")
|
||||||
|
return super().save(*args, **kwargs)
|
||||||
|
|
||||||
|
|
||||||
class PageSection(models.Model):
|
class PageSection(models.Model):
|
||||||
revision = models.ForeignKey(PageRevision, on_delete=models.CASCADE, related_name="sections")
|
revision = models.ForeignKey(PageRevision, on_delete=models.CASCADE, related_name="sections")
|
||||||
@@ -226,6 +250,16 @@ class PageSection(models.Model):
|
|||||||
if self.body and self.entries:
|
if self.body and self.entries:
|
||||||
raise ValidationError("Use body or entries, not both.")
|
raise ValidationError("Use body or entries, not both.")
|
||||||
|
|
||||||
|
def save(self, *args, **kwargs):
|
||||||
|
if self.pk:
|
||||||
|
raise ValidationError("Page sections are immutable; create a new revision.")
|
||||||
|
if self.revision.status != "draft":
|
||||||
|
raise ValidationError("Sections can only be added to a draft revision.")
|
||||||
|
return super().save(*args, **kwargs)
|
||||||
|
|
||||||
|
def delete(self, *args, **kwargs):
|
||||||
|
raise ValidationError("Page sections are immutable; create a new revision.")
|
||||||
|
|
||||||
|
|
||||||
class PublicationAudit(models.Model):
|
class PublicationAudit(models.Model):
|
||||||
actor = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT)
|
actor = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT)
|
||||||
@@ -238,3 +272,11 @@ class PublicationAudit(models.Model):
|
|||||||
|
|
||||||
class Meta:
|
class Meta:
|
||||||
constraints = [models.CheckConstraint(condition=(Q(site__isnull=False, page__isnull=True) | Q(site__isnull=True, page__isnull=False)), name="content_audit_one_target")]
|
constraints = [models.CheckConstraint(condition=(Q(site__isnull=False, page__isnull=True) | Q(site__isnull=True, page__isnull=False)), name="content_audit_one_target")]
|
||||||
|
|
||||||
|
def save(self, *args, **kwargs):
|
||||||
|
if self.pk:
|
||||||
|
raise ValidationError("Publication audit records are append only.")
|
||||||
|
return super().save(*args, **kwargs)
|
||||||
|
|
||||||
|
def delete(self, *args, **kwargs):
|
||||||
|
raise ValidationError("Publication audit records are append only.")
|
||||||
|
|||||||
241
apps/content/services.py
Normal file
241
apps/content/services.py
Normal file
@@ -0,0 +1,241 @@
|
|||||||
|
"""The only application path for changing editorial revisions and publication pointers."""
|
||||||
|
|
||||||
|
import hashlib
|
||||||
|
import json
|
||||||
|
|
||||||
|
from django.core.exceptions import PermissionDenied, ValidationError
|
||||||
|
from django.db import transaction
|
||||||
|
from django.db.models import Max
|
||||||
|
from django.utils import timezone
|
||||||
|
|
||||||
|
from .models import ABOUT_KEYS, HOME_KEYS, EditorialPage, PageRevision, PageSection, PublicationAudit, SiteIdentity, SiteRevision
|
||||||
|
|
||||||
|
|
||||||
|
PAGE_FIELDS = ("title", "intro", "seo_title", "seo_description", "seo_image")
|
||||||
|
SITE_FIELDS = (
|
||||||
|
"official_name", "short_name", "faculty_subline", "description", "logo", "logo_alt",
|
||||||
|
"navigation", "contacts", "footer_copy", "seo_title", "seo_description", "seo_image",
|
||||||
|
)
|
||||||
|
SECTION_FIELDS = (
|
||||||
|
"key", "section_type", "position", "visible", "heading", "body", "intro",
|
||||||
|
"all_link_label", "primary_action", "secondary_action", "entries", "image", "image_alt",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class PublishConflict(Exception):
|
||||||
|
"""The published pointer changed since the editor loaded it."""
|
||||||
|
|
||||||
|
|
||||||
|
def _allowed(actor, codename):
|
||||||
|
return bool(actor and actor.is_authenticated and actor.is_active and (
|
||||||
|
actor.is_superuser or (actor.is_staff and actor.has_perm(f"content.{codename}"))
|
||||||
|
))
|
||||||
|
|
||||||
|
|
||||||
|
def require_permission(actor, codename):
|
||||||
|
if not _allowed(actor, codename):
|
||||||
|
raise PermissionDenied("You do not have permission for this content action.")
|
||||||
|
|
||||||
|
|
||||||
|
def _fields(data, allowed):
|
||||||
|
if not isinstance(data, dict) or not set(data) <= set(allowed):
|
||||||
|
raise ValidationError("Unexpected editorial fields.")
|
||||||
|
return dict(data)
|
||||||
|
|
||||||
|
|
||||||
|
def _next_number(target):
|
||||||
|
return (target.revisions.aggregate(Max("number"))["number__max"] or 0) + 1
|
||||||
|
|
||||||
|
|
||||||
|
def _checksum(data):
|
||||||
|
encoded = json.dumps(data, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
||||||
|
return hashlib.sha256(encoded).hexdigest()
|
||||||
|
|
||||||
|
|
||||||
|
def _field_values(instance, names):
|
||||||
|
return {field: getattr(instance, f"{field}_id") if field in {"logo", "seo_image", "image"} else getattr(instance, field) for field in names}
|
||||||
|
|
||||||
|
|
||||||
|
def _copy_fields(instance, names):
|
||||||
|
return {field: getattr(instance, field) for field in names}
|
||||||
|
|
||||||
|
|
||||||
|
def _page_checksum(revision):
|
||||||
|
sections = [_field_values(section, SECTION_FIELDS) for section in revision.sections.order_by("position")]
|
||||||
|
return _checksum({"page": _field_values(revision, PAGE_FIELDS), "sections": sections})
|
||||||
|
|
||||||
|
|
||||||
|
def _site_checksum(revision):
|
||||||
|
return _checksum(_field_values(revision, SITE_FIELDS))
|
||||||
|
|
||||||
|
|
||||||
|
def _check_media_approved(revision):
|
||||||
|
media = []
|
||||||
|
for field in ("logo", "seo_image"):
|
||||||
|
if hasattr(revision, field):
|
||||||
|
media.append(getattr(revision, field))
|
||||||
|
if isinstance(revision, PageRevision):
|
||||||
|
media.extend(section.image for section in revision.sections.all())
|
||||||
|
if any(asset is not None and not asset.approved for asset in media):
|
||||||
|
raise ValidationError("All published media must be approved.")
|
||||||
|
|
||||||
|
|
||||||
|
def _validate_page(revision):
|
||||||
|
revision.full_clean()
|
||||||
|
sections = list(revision.sections.order_by("position"))
|
||||||
|
for section in sections:
|
||||||
|
section.full_clean()
|
||||||
|
slug = revision.page.slug
|
||||||
|
keys = [section.key for section in sections]
|
||||||
|
if slug == "home" and keys != list(HOME_KEYS):
|
||||||
|
raise ValidationError("Home requires all four sections in the fixed order.")
|
||||||
|
if slug == "about" and (not keys or keys[0] != "identity" or keys[-1] != "contact" or set(keys) - set(ABOUT_KEYS)):
|
||||||
|
raise ValidationError("About requires identity first and contact last.")
|
||||||
|
if slug in {"events", "blog"} and sections:
|
||||||
|
raise ValidationError("Listing pages cannot have sections.")
|
||||||
|
_check_media_approved(revision)
|
||||||
|
|
||||||
|
|
||||||
|
def _validate_site(revision):
|
||||||
|
revision.full_clean()
|
||||||
|
_check_media_approved(revision)
|
||||||
|
|
||||||
|
|
||||||
|
@transaction.atomic
|
||||||
|
def save_page_draft(actor, page: EditorialPage, fields: dict, sections: list[dict], *, base_revision=None):
|
||||||
|
require_permission(actor, "change_content_draft")
|
||||||
|
page = EditorialPage.objects.select_for_update().get(pk=page.pk)
|
||||||
|
if base_revision is not None:
|
||||||
|
base_revision = PageRevision.objects.get(pk=base_revision.pk)
|
||||||
|
if base_revision.page_id != page.pk:
|
||||||
|
raise ValidationError("Base revision belongs to another page.")
|
||||||
|
if not isinstance(sections, list):
|
||||||
|
raise ValidationError("Sections must be a list.")
|
||||||
|
revision = PageRevision(page=page, number=_next_number(page), created_by=actor, **_fields(fields, PAGE_FIELDS))
|
||||||
|
revision.full_clean()
|
||||||
|
revision.save()
|
||||||
|
for data in sections:
|
||||||
|
section = PageSection(revision=revision, **_fields(data, SECTION_FIELDS))
|
||||||
|
section.full_clean()
|
||||||
|
section.save()
|
||||||
|
revision.content_checksum = _page_checksum(revision)
|
||||||
|
revision.save(update_fields=["content_checksum"])
|
||||||
|
PublicationAudit.objects.create(actor=actor, page=page, from_revision=base_revision.number if base_revision else None, to_revision=revision.number, operation="update_draft" if base_revision else "create_draft")
|
||||||
|
return revision
|
||||||
|
|
||||||
|
|
||||||
|
@transaction.atomic
|
||||||
|
def save_site_draft(actor, site: SiteIdentity, fields: dict, *, base_revision=None):
|
||||||
|
require_permission(actor, "change_content_draft")
|
||||||
|
site = SiteIdentity.objects.select_for_update().get(pk=site.pk)
|
||||||
|
if base_revision is not None:
|
||||||
|
base_revision = SiteRevision.objects.get(pk=base_revision.pk)
|
||||||
|
if base_revision.site_id != site.pk:
|
||||||
|
raise ValidationError("Base revision belongs to another site.")
|
||||||
|
revision = SiteRevision(site=site, number=_next_number(site), created_by=actor, **_fields(fields, SITE_FIELDS))
|
||||||
|
revision.full_clean()
|
||||||
|
revision.content_checksum = _site_checksum(revision)
|
||||||
|
revision.save()
|
||||||
|
PublicationAudit.objects.create(actor=actor, site=site, from_revision=base_revision.number if base_revision else None, to_revision=revision.number, operation="update_draft" if base_revision else "create_draft")
|
||||||
|
return revision
|
||||||
|
|
||||||
|
|
||||||
|
def _check_expected(target, expected_published_id):
|
||||||
|
if target.published_revision_id != expected_published_id:
|
||||||
|
raise PublishConflict("Published content changed. Reload before publishing.")
|
||||||
|
|
||||||
|
|
||||||
|
@transaction.atomic
|
||||||
|
def publish_page(actor, draft: PageRevision, *, expected_published_id):
|
||||||
|
require_permission(actor, "publish_content")
|
||||||
|
page = EditorialPage.objects.select_for_update().get(pk=draft.page_id)
|
||||||
|
_check_expected(page, expected_published_id)
|
||||||
|
revision = PageRevision.objects.get(pk=draft.pk, page=page)
|
||||||
|
if revision.status != "draft":
|
||||||
|
raise ValidationError("Only a draft can be published.")
|
||||||
|
_validate_page(revision)
|
||||||
|
if revision.content_checksum != _page_checksum(revision):
|
||||||
|
raise ValidationError("Draft content changed outside the content service.")
|
||||||
|
revision.status = "published"
|
||||||
|
revision.published_by = actor
|
||||||
|
revision.published_at = timezone.now()
|
||||||
|
revision.save(update_fields=["status", "published_by", "published_at"])
|
||||||
|
if not EditorialPage.objects.filter(pk=page.pk, published_revision_id=expected_published_id).update(published_revision=revision):
|
||||||
|
raise PublishConflict("Published content changed. Reload before publishing.")
|
||||||
|
PublicationAudit.objects.create(actor=actor, page=page, from_revision=page.revisions.filter(pk=expected_published_id).values_list("number", flat=True).first(), to_revision=revision.number, operation="publish")
|
||||||
|
return revision
|
||||||
|
|
||||||
|
|
||||||
|
@transaction.atomic
|
||||||
|
def publish_site(actor, draft: SiteRevision, *, expected_published_id):
|
||||||
|
require_permission(actor, "publish_content")
|
||||||
|
site = SiteIdentity.objects.select_for_update().get(pk=draft.site_id)
|
||||||
|
_check_expected(site, expected_published_id)
|
||||||
|
revision = SiteRevision.objects.get(pk=draft.pk, site=site)
|
||||||
|
if revision.status != "draft":
|
||||||
|
raise ValidationError("Only a draft can be published.")
|
||||||
|
_validate_site(revision)
|
||||||
|
if revision.content_checksum != _site_checksum(revision):
|
||||||
|
raise ValidationError("Draft content changed outside the content service.")
|
||||||
|
revision.status = "published"
|
||||||
|
revision.published_by = actor
|
||||||
|
revision.published_at = timezone.now()
|
||||||
|
revision.save(update_fields=["status", "published_by", "published_at"])
|
||||||
|
if not SiteIdentity.objects.filter(pk=site.pk, published_revision_id=expected_published_id).update(published_revision=revision):
|
||||||
|
raise PublishConflict("Published content changed. Reload before publishing.")
|
||||||
|
PublicationAudit.objects.create(actor=actor, site=site, from_revision=site.revisions.filter(pk=expected_published_id).values_list("number", flat=True).first(), to_revision=revision.number, operation="publish")
|
||||||
|
return revision
|
||||||
|
|
||||||
|
|
||||||
|
@transaction.atomic
|
||||||
|
def restore_page(actor, page: EditorialPage, source: PageRevision, *, expected_published_id):
|
||||||
|
require_permission(actor, "restore_content")
|
||||||
|
page = EditorialPage.objects.select_for_update().get(pk=page.pk)
|
||||||
|
_check_expected(page, expected_published_id)
|
||||||
|
source = PageRevision.objects.get(pk=source.pk, page=page)
|
||||||
|
if source.status != "published":
|
||||||
|
raise ValidationError("Only a published revision can be restored.")
|
||||||
|
if source.content_checksum != _page_checksum(source):
|
||||||
|
raise ValidationError("Published content checksum does not match.")
|
||||||
|
revision = PageRevision(page=page, number=_next_number(page), created_by=actor, **_copy_fields(source, PAGE_FIELDS))
|
||||||
|
revision.full_clean()
|
||||||
|
revision.save()
|
||||||
|
for original in source.sections.all():
|
||||||
|
section = PageSection(revision=revision, **_copy_fields(original, SECTION_FIELDS))
|
||||||
|
section.full_clean()
|
||||||
|
section.save()
|
||||||
|
_validate_page(revision)
|
||||||
|
revision.content_checksum = _page_checksum(revision)
|
||||||
|
revision.status = "published"
|
||||||
|
revision.published_by = actor
|
||||||
|
revision.published_at = timezone.now()
|
||||||
|
revision.save(update_fields=["content_checksum", "status", "published_by", "published_at"])
|
||||||
|
if not EditorialPage.objects.filter(pk=page.pk, published_revision_id=expected_published_id).update(published_revision=revision):
|
||||||
|
raise PublishConflict("Published content changed. Reload before restoring.")
|
||||||
|
previous = page.revisions.filter(pk=expected_published_id).values_list("number", flat=True).first()
|
||||||
|
PublicationAudit.objects.create(actor=actor, page=page, from_revision=previous, to_revision=revision.number, operation="restore")
|
||||||
|
return revision
|
||||||
|
|
||||||
|
|
||||||
|
@transaction.atomic
|
||||||
|
def restore_site(actor, site: SiteIdentity, source: SiteRevision, *, expected_published_id):
|
||||||
|
require_permission(actor, "restore_content")
|
||||||
|
site = SiteIdentity.objects.select_for_update().get(pk=site.pk)
|
||||||
|
_check_expected(site, expected_published_id)
|
||||||
|
source = SiteRevision.objects.get(pk=source.pk, site=site)
|
||||||
|
if source.status != "published":
|
||||||
|
raise ValidationError("Only a published revision can be restored.")
|
||||||
|
if source.content_checksum != _site_checksum(source):
|
||||||
|
raise ValidationError("Published content checksum does not match.")
|
||||||
|
revision = SiteRevision(site=site, number=_next_number(site), created_by=actor, **_copy_fields(source, SITE_FIELDS))
|
||||||
|
_validate_site(revision)
|
||||||
|
revision.content_checksum = _site_checksum(revision)
|
||||||
|
revision.status = "published"
|
||||||
|
revision.published_by = actor
|
||||||
|
revision.published_at = timezone.now()
|
||||||
|
revision.save()
|
||||||
|
if not SiteIdentity.objects.filter(pk=site.pk, published_revision_id=expected_published_id).update(published_revision=revision):
|
||||||
|
raise PublishConflict("Published content changed. Reload before restoring.")
|
||||||
|
previous = site.revisions.filter(pk=expected_published_id).values_list("number", flat=True).first()
|
||||||
|
PublicationAudit.objects.create(actor=actor, site=site, from_revision=previous, to_revision=revision.number, operation="restore")
|
||||||
|
return revision
|
||||||
183
apps/content/tests/integration/test_services.py
Normal file
183
apps/content/tests/integration/test_services.py
Normal file
@@ -0,0 +1,183 @@
|
|||||||
|
import io
|
||||||
|
|
||||||
|
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 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"
|
||||||
|
|
||||||
|
|
||||||
|
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()
|
||||||
Reference in New Issue
Block a user