Replace content publishing workflow with direct Unfold editing

This commit is contained in:
2026-09-27 11:09:25 +03:30
parent 148b7038a9
commit f6b4ef7dd8
14 changed files with 680 additions and 1232 deletions

View File

@@ -1,212 +1,96 @@
import io
import pytest
from django.contrib.auth.models import Group
from django.core.files.uploadedfile import SimpleUploadedFile
from django.contrib.auth.models import Permission
from django.db import models
from django.urls import reverse
from PIL import Image
from apps.content.models import EditorialPage, MediaAsset, PageRevision, SiteIdentity
from apps.content.current import AboutContent, HomeContent, ListingContent, SiteContent
from apps.users.models import User
pytestmark = [pytest.mark.django_db, pytest.mark.integration]
def staff(name, group):
user = User.objects.create_user(username=name, mobile=f"0912{User.objects.count():07d}", password="TestPass123!")
user.is_staff = True
user.save(update_fields=["is_staff"])
user.groups.add(Group.objects.get(name=group))
def staff(username, *permissions):
user = User.objects.create_user(
username=username, mobile=f"0912{User.objects.count():07d}",
password="TestPass123!", is_staff=True,
)
user.user_permissions.add(*Permission.objects.filter(content_type__app_label="content", codename__in=permissions))
return user
def home_post():
data = {"title": "Home", "intro": "Welcome", "seo_title": "Home", "seo_description": "Association"}
for position, key in enumerate(("hero", "featured_events", "featured_posts", "participation")):
data[f"section_{key}-heading"] = key.replace("_", " ").title()
if key in {"featured_events", "featured_posts"}:
data[f"section_{key}-all_link_label"] = "See all"
data[f"section_{key}-body"] = "Introduction" if key in {"hero", "participation"} else ""
return data
def test_home_uses_standard_unfold_change_form_and_saves_directly(client):
client.force_login(staff("home-editor", "change_homecontent"))
url = reverse("admin:content_homecontent_change", args=[1])
def test_editor_can_save_and_preview_draft_without_publishing(client):
page = EditorialPage.objects.create(slug="home", page_kind="sections")
editor = staff("editor", "content_editor")
client.force_login(editor)
draft_url = reverse("admin:content_editorialpage_draft", args=[page.pk])
response = client.get(draft_url)
assert response.status_code == 200
assert b"section_hero-heading" in response.content
assert b"primary_action" not in response.content
response = client.post(draft_url, home_post(), follow=True)
assert response.status_code == 200
revision = PageRevision.objects.get(page=page)
assert revision.status == "draft"
assert page.published_revision_id is None
preview_url = reverse("admin:content_editorialpage_preview", args=[page.pk, revision.pk])
response = client.get(preview_url)
assert response.status_code == 200
assert b"Home" in response.content
assert "private" in response["Cache-Control"] and "no-store" in response["Cache-Control"]
publish_url = reverse("admin:content_editorialpage_publish", args=[page.pk, revision.pk])
assert client.post(publish_url, {"expected_published_id": ""}).status_code == 403
def test_publisher_can_publish_after_preview(client):
page = EditorialPage.objects.create(slug="home", page_kind="sections")
editor = staff("editor", "content_editor")
publisher = staff("publisher", "content_publisher")
client.force_login(editor)
client.post(reverse("admin:content_editorialpage_draft", args=[page.pk]), home_post())
revision = PageRevision.objects.get(page=page)
client.force_login(publisher)
response = client.post(reverse("admin:content_editorialpage_publish", args=[page.pk, revision.pk]), {"expected_published_id": ""}, follow=True)
assert response.status_code == 200
page.refresh_from_db()
assert page.published_revision_id == revision.pk
assert b"Published revision" in response.content
def test_site_admin_uses_named_contact_fields_and_rejects_bad_url(client):
site = SiteIdentity.objects.create()
editor = staff("editor", "content_editor")
client.force_login(editor)
url = reverse("admin:content_siteidentity_draft", args=[site.pk])
response = client.get(url)
assert response.status_code == 200
assert b"contact_email_url" in response.content
assert b'name="navigation"' not in response.content
data = {"official_name": "Association", "short_name": "ACE", "contact_email_url": "javascript:alert(1)", "contact_email_label": "Email"}
for position, route in enumerate(("home", "about", "events", "blog")):
data[f"nav_{route}_label"] = route
data[f"nav_{route}_position"] = position
data[f"nav_{route}_visible"] = "on"
assert b'name="hero_heading"' in response.content
assert b'name="seo_title"' in response.content
assert b"published_revision" not in response.content
assert client.get(reverse("admin:content_homecontent_add")).status_code == 403
current = HomeContent.objects.get(pk=1)
data = {
field.name: getattr(current, field.name)
for field in current._meta.fields
if field.editable and not field.is_relation and not isinstance(field, models.ImageField)
}
data["hero_heading"] = "A new heading"
data["seo_title"] = "New SEO title"
data["_save"] = "Save"
response = client.post(url, data)
assert response.status_code == 302
current.refresh_from_db()
assert current.hero_heading == "A new heading"
assert current.seo_title == "New SEO title"
def test_staff_without_model_change_permission_cannot_edit(client):
client.force_login(staff("other-staff"))
assert client.get(reverse("admin:content_aboutcontent_change", args=[1])).status_code == 403
assert client.post(reverse("admin:content_aboutcontent_change", args=[1]), {"title": "Changed"}).status_code == 403
def test_site_links_use_native_fields_and_reject_unsafe_url(client):
client.force_login(staff("site-editor", "change_sitecontent"))
url = reverse("admin:content_sitecontent_change", args=[1])
response = client.get(url)
assert response.status_code == 200
assert b'name="telegram_url"' in response.content
assert b'name="logo"' in response.content
current = SiteContent.objects.get(pk=1)
data = {
field.name: getattr(current, field.name)
for field in current._meta.fields
if field.editable and not field.is_relation and not isinstance(field, models.ImageField)
}
data["telegram_url"] = "javascript:alert(1)"
response = client.post(url, data)
assert response.status_code == 200
assert b"contact_email_url" in response.content
assert site.revisions.count() == 0
current.refresh_from_db()
assert current.telegram_url != "javascript:alert(1)"
def test_about_optional_section_can_be_hidden_and_reordered(client):
page = EditorialPage.objects.create(slug="about", page_kind="sections")
client.force_login(staff("editor", "content_editor"))
data = {"title": "About", "intro": "Introduction", "seo_title": "About", "seo_description": "Association"}
for key, heading in (("identity", "Identity"), ("mission", "Mission"), ("contact", "Contact")):
data[f"section_{key}-heading"] = heading
data["section_identity-body"] = "Official story"
data["section_mission-position"] = "2"
data["section_mission-body"] = "Our mission"
for key in ("mission", "activities", "values", "committees", "affiliations", "faq"):
data[f"entries_{key}-TOTAL_FORMS"] = "1"
data[f"entries_{key}-INITIAL_FORMS"] = "0"
data[f"entries_{key}-MIN_NUM_FORMS"] = "0"
data[f"entries_{key}-MAX_NUM_FORMS"] = "20"
response = client.post(reverse("admin:content_editorialpage_draft", args=[page.pk]), data, follow=True)
assert response.status_code == 200
revision = PageRevision.objects.get(page=page)
assert list(revision.sections.values_list("key", "position", "visible")) == [
("identity", 0, True), ("mission", 2, False), ("contact", 3, True),
]
def test_fixed_page_records_exist_and_cannot_be_added_or_deleted(client):
assert HomeContent.objects.filter(pk=1).exists()
assert AboutContent.objects.filter(pk=1).exists()
assert SiteContent.objects.filter(pk=1).exists()
assert set(ListingContent.objects.values_list("slug", flat=True)) == {"events", "blog"}
client.force_login(staff("listing-editor", "change_listingcontent"))
assert client.get(reverse("admin:content_listingcontent_add")).status_code == 403
assert client.get(reverse("admin:content_listingcontent_delete", args=["events"])).status_code == 403
assert client.get(reverse("admin:content_listingcontent_change", args=["events"])).status_code == 200
def test_site_editor_can_save_named_links_and_preview(client):
site = SiteIdentity.objects.create()
client.force_login(staff("editor", "content_editor"))
data = {"official_name": "Association", "short_name": "ACE", "contact_email_label": "Email", "contact_email_url": "mailto:test@example.com"}
for position, route in enumerate(("home", "about", "events", "blog")):
data[f"nav_{route}_label"] = route
data[f"nav_{route}_position"] = str(position)
data[f"nav_{route}_visible"] = "on"
response = client.post(reverse("admin:content_siteidentity_draft", args=[site.pk]), data, follow=True)
assert response.status_code == 200
revision = site.revisions.get()
assert revision.contacts == [{"kind": "email", "label": "Email", "url": "mailto:test@example.com", "position": 0, "visible": False}]
assert site.published_revision_id is None
preview = client.get(reverse("admin:content_siteidentity_preview", args=[site.pk, revision.pk]))
assert preview.status_code == 200 and b"test@example.com" in preview.content
def test_anonymous_cannot_preview_draft(client):
page = EditorialPage.objects.create(slug="home", page_kind="sections")
client.force_login(staff("editor", "content_editor"))
client.post(reverse("admin:content_editorialpage_draft", args=[page.pk]), home_post())
revision = page.revisions.get()
client.logout()
response = client.get(reverse("admin:content_editorialpage_preview", args=[page.pk, revision.pk]))
assert response.status_code == 302 and "/admin/login/" in response["Location"]
def test_staff_without_content_permission_cannot_preview_or_publish(client):
page = EditorialPage.objects.create(slug="home", page_kind="sections")
client.force_login(staff("editor", "content_editor"))
client.post(reverse("admin:content_editorialpage_draft", args=[page.pk]), home_post())
revision = page.revisions.get()
plain_staff = User.objects.create_user(username="plain", mobile="09121111111", password="TestPass123!", is_staff=True)
client.force_login(plain_staff)
assert client.get(reverse("admin:content_editorialpage_preview", args=[page.pk, revision.pk])).status_code == 403
assert client.post(reverse("admin:content_editorialpage_publish", args=[page.pk, revision.pk]), {"expected_published_id": ""}).status_code == 403
def test_stale_admin_draft_post_reports_conflict(client):
page = EditorialPage.objects.create(slug="home", page_kind="sections")
client.force_login(staff("editor", "content_editor"))
url = reverse("admin:content_editorialpage_draft", args=[page.pk])
assert client.get(url).status_code == 200
client.post(url, home_post())
response = client.post(url, home_post())
assert response.status_code == 200
assert b"newer revision" in response.content
assert page.revisions.count() == 1
def test_media_approval_is_publisher_only_and_locks_asset(client, settings, tmp_path):
settings.MEDIA_ROOT = tmp_path
buffer = io.BytesIO()
Image.new("RGB", (20, 20)).save(buffer, format="PNG")
client.force_login(staff("editor", "content_editor"))
upload = SimpleUploadedFile("logo.png", buffer.getvalue(), content_type="image/png")
response = client.post(reverse("admin:content_mediaasset_add"), {
"file": upload, "alt": "Association logo", "source": "Owner", "license": "Owned", "approved": "on", "_save": "Save",
})
assert response.status_code == 302
asset = MediaAsset.objects.get()
assert not asset.approved
client.force_login(staff("publisher", "content_publisher"))
change_url = reverse("admin:content_mediaasset_change", args=[asset.pk])
response = client.post(change_url, {
"alt": asset.alt, "source": asset.source, "license": asset.license, "approved": "on", "_save": "Save",
})
assert response.status_code == 302
asset.refresh_from_db()
assert asset.approved
assert client.get(change_url).status_code == 200
assert client.post(change_url, {"approved": ""}).status_code == 403
def test_listing_intro_page_saves_without_sections(client):
page = EditorialPage.objects.create(slug="events", page_kind="listing_intro")
client.force_login(staff("editor", "content_editor"))
response = client.post(reverse("admin:content_editorialpage_draft", args=[page.pk]), {
"title": "Events", "intro": "Find upcoming events", "seo_title": "Events", "seo_description": "Association events",
}, follow=True)
assert response.status_code == 200
revision = page.revisions.get()
assert revision.intro == "Find upcoming events" and revision.sections.count() == 0
def test_admin_restore_creates_new_published_revision(client):
page = EditorialPage.objects.create(slug="home", page_kind="sections")
client.force_login(staff("publisher", "content_publisher"))
draft_url = reverse("admin:content_editorialpage_draft", args=[page.pk])
client.post(draft_url, home_post() | {"title": "First"})
first = page.revisions.get()
client.post(reverse("admin:content_editorialpage_publish", args=[page.pk, first.pk]), {"expected_published_id": ""})
client.post(draft_url, home_post() | {"title": "Second", "base_revision_id": str(first.pk)})
second = page.revisions.order_by("-number").first()
client.post(reverse("admin:content_editorialpage_publish", args=[page.pk, second.pk]), {"expected_published_id": str(first.pk)})
response = client.post(reverse("admin:content_editorialpage_restore", args=[page.pk, first.pk]), {"expected_published_id": str(second.pk)}, follow=True)
assert response.status_code == 200
page.refresh_from_db()
assert page.published_revision.number == 3 and page.published_revision.title == "First"
def test_current_website_copy_is_seeded_once():
home = HomeContent.objects.get(pk=1)
about = AboutContent.objects.get(pk=1)
events = ListingContent.objects.get(pk="events")
blog = ListingContent.objects.get(pk="blog")
assert home.hero_heading == "انجمن علمی کامپیوتر دانشگاه گیلان"
assert about.items.filter(kind="faq").count() == 3
assert events.seo_title == "رویدادها"
assert blog.title == "بلاگ" and blog.seo_title == "وبلاگ"

View File

@@ -1,218 +0,0 @@
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()