Files
guilan-ace-backend/apps/blog/tests/integration/test_public_blog.py

337 lines
20 KiB
Python

import pytest
from django.core.files.uploadedfile import SimpleUploadedFile
from apps.blog.models import Category, Comment, Like, Post, PostAsset, SavedPost, Tag
pytestmark = [pytest.mark.integration, pytest.mark.django_db]
def test_public_blog_lists_empty_content(client):
response = client.get("/api/blog/posts")
assert response.status_code == 200
assert response.json() == []
def test_admin_blog_requires_authentication(client):
assert client.get("/api/blog/admin/posts").status_code == 401
@pytest.fixture
def blog_content(user_factory):
author = user_factory()
category = Category.objects.create(name="Events")
child = Category.objects.create(name="News", parent=category)
tag = Tag.objects.create(name="Featured")
first = Post.objects.create(title="First story", content="**First** story", author=author,
category=child, status=Post.StatusChoices.PUBLISHED, is_featured=True)
first.tags.add(tag)
second = Post.objects.create(title="Second story", content="Second content", author=author,
category=category, status=Post.StatusChoices.PUBLISHED)
draft = Post.objects.create(title="Private draft", content="Draft", author=author)
return author, category, child, tag, first, second, draft
def test_public_filters_posts_and_recommendations(client, blog_content):
author, category, child, tag, first, second, draft = blog_content
assert {post["slug"] for post in client.get("/api/blog/posts").json()} == {first.slug, second.slug}
assert {post["slug"] for post in client.get("/api/blog/posts", {"category": category.slug}).json()} == {first.slug, second.slug}
assert [post["slug"] for post in client.get("/api/blog/posts", {"tag": tag.slug}).json()] == [first.slug]
assert [post["slug"] for post in client.get("/api/blog/posts", {"search": "Second"}).json()] == [second.slug]
assert [post["slug"] for post in client.get("/api/blog/posts", {"featured": "true"}).json()] == [first.slug]
assert {post["slug"] for post in client.get("/api/blog/posts", {"author": author.username}).json()} == {first.slug, second.slug}
assert client.get("/api/blog/posts", {"category": "unknown"}).json() == []
assert client.get(f"/api/blog/posts/{draft.slug}").status_code == 404
assert client.get(f"/api/blog/posts/{first.slug}").json()["title"] == first.title
assert [post["slug"] for post in client.get(f"/api/blog/posts/{first.slug}/recommended").json()] == [second.slug]
filters = client.get("/api/blog/filters").json()
assert filters["categories"][0]["children"][0]["slug"] == child.slug
assert filters["tags"][0]["post_count"] == 1
assert filters["authors"][0]["post_count"] == 2
def test_public_taxonomy_and_comment_visibility(client, blog_content, user_factory):
author, category, child, tag, post, _, _ = blog_content
member = user_factory()
approved = Comment.objects.create(post=post, author=member, content="Visible")
hidden = Comment.objects.create(post=post, author=member, content="Hidden", is_hidden=True)
assert [row["slug"] for row in client.get("/api/blog/categories").json()] == [category.slug, child.slug]
assert client.get(f"/api/blog/categories/{category.slug}").json()["name"] == category.name
assert client.get(f"/api/blog/tags/{tag.slug}").json()["name"] == tag.name
assert [row["id"] for row in client.get(f"/api/blog/posts/{post.slug}/comments").json()] == [approved.id]
assert hidden.id != approved.id
def test_blog_interactions_and_activity(client, blog_content, user_factory, auth_headers, mocker):
_, _, _, _, post, _, _ = blog_content
member = user_factory()
headers = auth_headers(member)
mocker.patch("apps.blog.api.views.notify_user")
assert client.post(f"/api/blog/posts/{post.slug}/like", **headers).json()["liked"] is True
assert Like.objects.filter(post=post, user=member).exists()
assert client.post(f"/api/blog/posts/{post.slug}/save", **headers).json()["saved"] is True
assert SavedPost.objects.filter(post=post, user=member).exists()
response = client.post(f"/api/blog/posts/{post.slug}/comments", data={"content": "A useful comment"},
content_type="application/json", **headers)
assert response.status_code == 201, response.content
comment = Comment.objects.get(pk=response.json()["id"])
assert comment.content == "A useful comment"
assert client.get(f"/api/blog/posts/{post.slug}/interaction", **headers).json()["comments_count"] == 1
activity = client.get("/api/blog/me/activity", **headers).json()
assert activity["liked_posts"][0]["id"] == post.id
assert activity["saved_posts"][0]["id"] == post.id
assert activity["comments"][0]["id"] == comment.id
assert client.put(f"/api/blog/comments/{comment.id}", data={"content": "Edited"},
content_type="application/json", **headers).status_code == 200
comment.refresh_from_db()
assert comment.content == "Edited"
assert client.post(f"/api/blog/posts/{post.slug}/like", **headers).json()["liked"] is False
assert not Like.objects.filter(post=post, user=member).exists()
def test_my_blog_activity_paginates_each_category_independently(client, blog_content, user_factory, auth_headers):
_, _, _, _, _, _, _ = blog_content
member = user_factory()
headers = auth_headers(member)
posts = [
Post.objects.create(title=f"Saved and liked {index}", content="Content", author=member,
status=Post.StatusChoices.PUBLISHED)
for index in range(12)
]
for post in posts:
Like.objects.create(post=post, user=member)
SavedPost.objects.create(post=post, user=member)
Comment.objects.create(post=post, author=member, content=f"Comment {post.id}")
liked_page = client.get("/api/blog/me/activity/liked", {"page": 1}, **headers)
saved_page = client.get("/api/blog/me/activity/saved", {"page": 2}, **headers)
comments_page = client.get("/api/blog/me/activity/comments", {"page": 2}, **headers)
assert liked_page.status_code == 200, liked_page.content
assert liked_page.json()["count"] == 12
assert liked_page.json()["page_size"] == 10
assert len(liked_page.json()["results"]) == 10
assert len(saved_page.json()["results"]) == 2
assert len(comments_page.json()["results"]) == 2
assert comments_page.json()["total_pages"] == 2
def test_blog_editor_review_and_moderation(client, blog_content, user_factory, auth_headers, mocker):
author, _, _, _, _, _, _ = blog_content
editor = user_factory(is_staff=True)
member = user_factory()
editor_headers = auth_headers(editor)
member_headers = auth_headers(member)
mocker.patch("apps.blog.api.views.notify_user")
assert client.get("/api/blog/admin/posts", **member_headers).status_code == 403
created = client.post("/api/blog/admin/posts", data={"title": "Reviewed story", "content": "Body"},
content_type="application/json", **editor_headers)
assert created.status_code == 201, created.content
post = Post.objects.get(pk=created.json()["id"])
assert post.author == editor and post.status == Post.StatusChoices.DRAFT
submitted = client.post(f"/api/blog/admin/posts/{post.id}/submit", **editor_headers)
assert submitted.status_code == 200
post.refresh_from_db()
assert post.status == Post.StatusChoices.SUBMITTED
invalid = client.post(f"/api/blog/admin/posts/{post.id}/review", data={"action": "nope"},
content_type="application/json", **editor_headers)
assert invalid.status_code == 400
reviewed = client.post(f"/api/blog/admin/posts/{post.id}/review", data={"action": "publish"},
content_type="application/json", **editor_headers)
assert reviewed.status_code == 200, reviewed.content
post.refresh_from_db()
assert post.status == Post.StatusChoices.PUBLISHED and post.published_by == editor
assert client.get(f"/api/blog/posts/{post.slug}").status_code == 200
response = client.post(f"/api/blog/posts/{post.slug}/comments", data={"content": "Review me"},
content_type="application/json", **member_headers)
comment = Comment.objects.get(pk=response.json()["id"])
assert client.post(f"/api/blog/comments/{comment.id}/hide", data={"note": "spam"},
content_type="application/json", **member_headers).status_code == 403
assert client.post(f"/api/blog/comments/{comment.id}/hide", data={"note": "spam"},
content_type="application/json", **editor_headers).status_code == 200
comment.refresh_from_db()
assert comment.is_hidden
assert client.get(f"/api/blog/posts/{post.slug}/comments").json() == []
assert client.post(f"/api/blog/comments/{comment.id}/unhide", **editor_headers).status_code == 200
comment.refresh_from_db()
assert not comment.is_hidden
assert client.post(f"/api/blog/comments/{comment.id}/delete", data={"note": "remove"},
content_type="application/json", **editor_headers).status_code == 200
assert Comment.deleted_objects.filter(pk=comment.pk).exists()
assert client.post(f"/api/blog/deleted/comments/{comment.id}/restore", **editor_headers).status_code == 200
assert Comment.objects.filter(pk=comment.pk).exists()
def test_blog_taxonomy_admin_crud_and_restore(client, user_factory, auth_headers):
admin = user_factory(is_superuser=True, is_staff=True)
member = user_factory()
admin_headers = auth_headers(admin)
assert client.get("/api/blog/admin/categories", **auth_headers(member)).status_code == 403
root = client.post("/api/blog/admin/categories", data={"name": "Research"},
content_type="application/json", **admin_headers)
assert root.status_code == 201, root.content
category_id = root.json()["id"]
assert Category.objects.filter(pk=category_id).exists()
child = client.post("/api/blog/admin/categories", data={"name": "Reports", "parent_id": category_id},
content_type="application/json", **admin_headers)
assert child.status_code == 201, child.content
invalid = client.put(f"/api/blog/admin/categories/{category_id}",
data={"name": "Research", "parent_id": child.json()["id"]},
content_type="application/json", **admin_headers)
assert invalid.status_code == 400
assert Category.objects.get(pk=category_id).parent_id is None
tag = client.post("/api/blog/admin/tags", data={"name": "Python"},
content_type="application/json", **admin_headers)
assert tag.status_code == 201, tag.content
tag_id = tag.json()["id"]
assert client.put(f"/api/blog/admin/tags/{tag_id}", data={"name": "Django"},
content_type="application/json", **admin_headers).status_code == 200
assert Tag.objects.get(pk=tag_id).name == "Django"
assert client.delete(f"/api/blog/admin/tags/{tag_id}", **admin_headers).status_code == 200
assert Tag.deleted_objects.filter(pk=tag_id).exists()
assert client.post(f"/api/blog/deleted/tags/{tag_id}/restore", **admin_headers).status_code == 200
assert Tag.objects.filter(pk=tag_id).exists()
def test_blog_post_admin_listing_update_assets_and_soft_delete(client, blog_content, user_factory,
auth_headers, tmp_path, settings):
settings.MEDIA_ROOT = tmp_path
author, _, _, _, post, _, draft = blog_content
admin = user_factory(is_superuser=True, is_staff=True)
member = user_factory()
headers = auth_headers(admin)
assert client.get("/api/blog/admin/posts", **headers).status_code == 200
assert client.get(f"/api/blog/admin/posts/{post.pk}", **auth_headers(member)).status_code == 403
assert client.get(f"/api/blog/admin/posts/{post.pk}", **headers).json()["id"] == post.pk
updated = client.put(f"/api/blog/admin/posts/{draft.pk}", data='{"title":"New title","content":"Changed"}',
content_type="application/json", **headers)
assert updated.status_code == 200, updated.content
draft.refresh_from_db()
assert draft.title == "New title" and draft.content == "Changed"
assert client.get(f"/api/blog/admin/posts/{post.pk}/assets", **headers).json() == []
invalid = client.post(f"/api/blog/admin/posts/{post.pk}/assets",
{"file": SimpleUploadedFile("unsafe.exe", b"data", content_type="application/octet-stream")},
**headers)
assert invalid.status_code == 400
asset = client.post(f"/api/blog/admin/posts/{post.pk}/assets",
{"file": SimpleUploadedFile("notes.txt", b"notes", content_type="text/plain"), "title": "Notes"},
**headers)
assert asset.status_code == 201, asset.content
asset_id = asset.json()["id"]
assert PostAsset.objects.filter(pk=asset_id, post=post).exists()
assert len(client.get(f"/api/blog/admin/posts/{post.pk}/assets", **headers).json()) == 1
assert client.delete(f"/api/blog/admin/posts/{post.pk}/assets/{asset_id}", **headers).status_code == 200
assert not PostAsset.objects.filter(pk=asset_id).exists()
assert client.delete(f"/api/blog/posts/{post.slug}", **auth_headers(author)).status_code == 403
assert client.delete(f"/api/blog/posts/{post.slug}", **headers).status_code == 200
assert Post.deleted_objects.filter(pk=post.pk).exists()
assert client.get("/api/blog/deleted/posts", **headers).json()[0]["id"] == post.pk
assert client.post(f"/api/blog/deleted/posts/{post.pk}/restore", **headers).status_code == 200
assert Post.objects.filter(pk=post.pk).exists()
def test_blog_comment_reply_and_edit_authorization(client, blog_content, user_factory, auth_headers, mocker):
_, _, _, _, post, _, _ = blog_content
first_user = user_factory()
second_user = user_factory()
notify = mocker.patch("apps.blog.api.views.notify_user")
root = client.post(f"/api/blog/posts/{post.slug}/comments", data='{"content":"Root"}',
content_type="application/json", **auth_headers(first_user))
assert root.status_code == 201, root.content
root_id = root.json()["id"]
reply = client.post(f"/api/blog/posts/{post.slug}/comments",
data=f'{{"content":"Reply","parent_id":{root_id}}}',
content_type="application/json", **auth_headers(second_user))
assert reply.status_code == 201, reply.content
assert Comment.objects.get(pk=reply.json()["id"]).parent_id == root_id
assert notify.call_count >= 2
assert client.put(f"/api/blog/comments/{root_id}", data='{"content":"Changed"}',
content_type="application/json", **auth_headers(second_user)).status_code == 403
comments = client.get(f"/api/blog/posts/{post.slug}/comments").json()
assert comments[0]["replies"][0]["id"] == reply.json()["id"]
def test_blog_taxonomy_deletion_and_validation(client, auth_headers, user_factory):
admin = user_factory(is_superuser=True, is_staff=True)
headers = auth_headers(admin)
category = Category.objects.create(name="Delete category")
tag = Tag.objects.create(name="Delete tag")
assert client.post("/api/blog/admin/categories", data='{"name":" "}',
content_type="application/json", **headers).status_code == 400
assert client.post("/api/blog/admin/tags", data='{"name":" "}',
content_type="application/json", **headers).status_code == 400
assert client.delete(f"/api/blog/admin/categories/{category.pk}", **headers).status_code == 200
assert client.get("/api/blog/deleted/categories", **headers).json()[0]["id"] == category.pk
assert client.post(f"/api/blog/deleted/categories/{category.pk}/restore", **headers).status_code == 200
assert Category.objects.filter(pk=category.pk).exists()
assert client.post("/api/blog/deleted/categories/999999/restore", **headers).status_code == 400
assert client.get(f"/api/blog/tags/{tag.slug}").json()["id"] == tag.pk
def test_blog_featured_image_and_compatibility_routes(client, blog_content, user_factory,
auth_headers, tmp_path, settings):
settings.MEDIA_ROOT = tmp_path
_, _, _, _, post, _, _ = blog_content
admin = user_factory(is_staff=True, is_superuser=True)
headers = auth_headers(admin)
assert client.get("/api/blog/admin/writers", **headers).status_code == 200
assert client.get("/api/blog/admin/writers", **auth_headers(user_factory())).status_code == 403
invalid = client.post(f"/api/blog/admin/posts/{post.pk}/featured-image",
{"file": SimpleUploadedFile("notes.txt", b"text", content_type="text/plain")}, **headers)
assert invalid.status_code == 400
uploaded = client.post(f"/api/blog/admin/posts/{post.pk}/featured-image",
{"file": SimpleUploadedFile("cover.png", b"image", content_type="image/png")}, **headers)
assert uploaded.status_code == 200, uploaded.content
post.refresh_from_db()
assert post.featured_image
assert client.delete(f"/api/blog/admin/posts/{post.pk}/featured-image", **headers).status_code == 200
post.refresh_from_db()
assert not post.featured_image
compatible = client.post("/api/blog/posts", data='{"title":"Compat","content":"Body"}',
content_type="application/json", **headers)
assert compatible.status_code == 201, compatible.content
compat_post = Post.objects.get(pk=compatible.json()["id"])
update = client.put(f"/api/blog/posts/{compat_post.slug}", data='{"title":"Compat edited","content":"New body"}',
content_type="application/json", **headers)
assert update.status_code == 200, update.content
compat_post.refresh_from_db()
assert compat_post.title == "Compat edited"
def test_blog_review_changes_and_taxonomy_parent_validation(client, blog_content, user_factory, auth_headers):
_, _, _, _, _, _, draft = blog_content
admin = user_factory(is_staff=True)
headers = auth_headers(admin)
changed = client.post(f"/api/blog/admin/posts/{draft.pk}/review",
data='{"action":"request_changes","note":"Revise details"}',
content_type="application/json", **headers)
assert changed.status_code == 200, changed.content
draft.refresh_from_db()
assert draft.status == Post.StatusChoices.CHANGES_REQUESTED and draft.review_note == "Revise details"
archived = client.post(f"/api/blog/admin/posts/{draft.pk}/review", data='{"action":"archive"}',
content_type="application/json", **headers)
assert archived.status_code == 200
draft.refresh_from_db()
assert draft.status == Post.StatusChoices.ARCHIVED
root = Category.objects.create(name="Root for validation")
child = Category.objects.create(name="Child for validation", parent=root)
invalid = client.put(f"/api/blog/admin/categories/{child.pk}",
data=f'{{"name":"Child for validation","parent_id":{child.pk}}}',
content_type="application/json", **headers)
assert invalid.status_code == 400
child.refresh_from_db()
assert child.parent_id == root.pk
def test_blog_filters_accept_multiple_tag_values_and_page_results(client, blog_content):
_, _, _, tag, first, second, _ = blog_content
tagged = client.get("/api/blog/posts", {"tag": f"{tag.slug},unknown,{tag.slug}"})
assert tagged.status_code == 200
assert [row["id"] for row in tagged.json()] == [first.pk]
first_page = client.get("/api/blog/posts", {"page": 1, "limit": 1}).json()
second_page = client.get("/api/blog/posts", {"page": 2, "limit": 1}).json()
assert len(first_page) == len(second_page) == 1
assert {first_page[0]["id"], second_page[0]["id"]} == {first.pk, second.pk}
assert client.get("/api/blog/posts", {"page": 3, "limit": 1}).json() == []
assert [row["id"] for row in client.get(f"/api/blog/posts/{second.slug}/recommended").json()] == [first.pk]