Files
blog.pyaqa.ru/tests/api/test_posts.py
Sergey Vanyushkin c9b380c601 test(api): add full API test suite with get_keycloak_client async fix
Add 45 API tests covering all 12 post endpoints (CRUD, publish/unpublish) with RBAC policy coverage across guest, user, admin roles.

Fix get_keycloak_client() in deps.py to be async - Dishka's async container requires await on get(), without it a coroutine object was returned instead of the actual client.
2026-05-10 11:21:58 +00:00

789 lines
26 KiB
Python

"""API tests for blog post CRUD and publish operations.
This module tests all 12 blog post API endpoints covering create, read,
update, delete, publish, and unpublish operations with full authorization
policy coverage across guest, user, and admin roles.
"""
from typing import Any
from uuid import UUID
from fastapi.testclient import TestClient
from tests.api.conftest import API_PREFIX, USER_ID
class TestCreatePost:
"""Tests for POST /api/v1/posts — create a new blog post."""
def test_create_post_success(
self,
client: TestClient,
user_headers: dict[str, str],
post_payload: dict[str, Any],
) -> None:
"""Test creating a post returns 201 with correct fields.
TC-API-001: Positive — create post as authenticated user.
"""
response = client.post(API_PREFIX, json=post_payload, headers=user_headers)
assert response.status_code == 201
data = response.json()
assert data["title"] == post_payload["title"]
assert data["content"] == post_payload["content"]
assert data["author_id"] == USER_ID
assert data["published"] is False
assert data["tags"] == post_payload["tags"]
assert UUID(data["id"])
assert data["slug"]
assert data["created_at"]
assert data["updated_at"]
def test_create_post_no_auth(
self,
client: TestClient,
post_payload: dict[str, Any],
) -> None:
"""Test creating a post without auth returns 401.
TC-API-101: Negative — no authorization header.
"""
response = client.post(API_PREFIX, json=post_payload)
assert response.status_code == 401
error = response.json()
assert error["error"] == "UnauthorizedException"
def test_create_post_guest_token(
self,
client: TestClient,
guest_headers: dict[str, str],
post_payload: dict[str, Any],
) -> None:
"""Test creating a post with inactive token returns 401.
TC-API-102: Negative — guest/inactive token.
"""
response = client.post(API_PREFIX, json=post_payload, headers=guest_headers)
assert response.status_code == 401
error = response.json()
assert error["error"] == "UnauthorizedException"
def test_create_post_invalid_payload(
self,
client: TestClient,
user_headers: dict[str, str],
) -> None:
"""Test creating a post with too-short title returns 422."""
response = client.post(
API_PREFIX,
json={"title": "ab", "content": "valid content here"},
headers=user_headers,
)
assert response.status_code == 422
class TestListPosts:
"""Tests for GET /api/v1/posts — list posts with filters."""
def test_list_posts_default(
self,
client: TestClient,
) -> None:
"""Test default listing returns published posts.
TC-API-004: Positive — default listing without auth.
"""
response = client.get(API_PREFIX)
assert response.status_code == 200
data = response.json()
assert "items" in data
assert "total" in data
assert isinstance(data["items"], list)
assert isinstance(data["total"], int)
def test_list_posts_pagination(
self,
client: TestClient,
user_headers: dict[str, str],
) -> None:
"""Test listing posts with limit and offset query params."""
for i in range(3):
payload = {
"title": f"Pagination Post {i}",
"content": f"Content for pagination test post {i}. Enough characters here.",
"tags": [],
}
client.post(API_PREFIX, json=payload, headers=user_headers)
response = client.get(f"{API_PREFIX}?limit=2&offset=0")
assert response.status_code == 200
data = response.json()
assert len(data["items"]) <= 2
def test_list_posts_include_unpublished_as_admin(
self,
client: TestClient,
admin_headers: dict[str, str],
) -> None:
"""Test admin can list unpublished posts.
TC-API-005: Positive — admin can include unpublished.
"""
response = client.get(f"{API_PREFIX}?include_unpublished=true", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert "items" in data
def test_list_posts_include_unpublished_as_user_returns_403(
self,
client: TestClient,
user_headers: dict[str, str],
) -> None:
"""Test user gets 403 when requesting unpublished posts.
TC-API-006: Policy — user cannot list unpublished.
"""
response = client.get(f"{API_PREFIX}?include_unpublished=true", headers=user_headers)
assert response.status_code == 403
error = response.json()
assert error["error"] == "ForbiddenException"
def test_list_posts_include_unpublished_as_guest_returns_403(
self,
client: TestClient,
guest_headers: dict[str, str],
) -> None:
"""Test guest gets 403 when requesting unpublished posts.
TC-API-007: Policy — guest cannot list unpublished.
"""
response = client.get(f"{API_PREFIX}?include_unpublished=true", headers=guest_headers)
assert response.status_code == 403
error = response.json()
assert error["error"] == "ForbiddenException"
def test_list_posts_without_auth_include_unpublished_returns_403(
self,
client: TestClient,
) -> None:
"""Test anonymous gets 403 when requesting unpublished posts."""
response = client.get(f"{API_PREFIX}?include_unpublished=true")
assert response.status_code == 403
error = response.json()
assert error["error"] == "ForbiddenException"
class TestListPublishedPosts:
"""Tests for GET /api/v1/posts/published — list published posts."""
def test_list_published_posts_success(
self,
client: TestClient,
) -> None:
"""Test listing published posts returns 200 with items.
TC-API-008: Positive — public endpoint, no auth needed.
"""
response = client.get(f"{API_PREFIX}/published")
assert response.status_code == 200
data = response.json()
assert "items" in data
assert "total" in data
class TestSearchPosts:
"""Tests for GET /api/v1/posts/search — search posts."""
def test_search_posts_success(
self,
client: TestClient,
user_headers: dict[str, str],
) -> None:
"""Test searching posts returns matching results.
TC-API-009: Positive — public endpoint, no auth needed.
"""
payload = {
"title": "Searchable Unique Title",
"content": "This content contains a very special search keyword. Enough length.",
"tags": [],
}
client.post(API_PREFIX, json=payload, headers=user_headers)
response = client.get(f"{API_PREFIX}/search?query=Searchable")
assert response.status_code == 200
data = response.json()
assert len(data["items"]) >= 1
assert any("Searchable" in item["title"] for item in data["items"])
def test_search_posts_no_results(
self,
client: TestClient,
) -> None:
"""Test searching with no matches returns empty list."""
response = client.get(f"{API_PREFIX}/search?query=xyznonexistent12345")
assert response.status_code == 200
data = response.json()
assert data["items"] == []
assert data["total"] == 0
def test_search_posts_public(
self,
client: TestClient,
) -> None:
"""Test search is accessible without authentication."""
response = client.get(f"{API_PREFIX}/search?query=test")
assert response.status_code == 200
class TestGetPostsByTag:
"""Tests for GET /api/v1/posts/by-tag/{tag} — posts by tag."""
def test_get_posts_by_tag_success(
self,
client: TestClient,
user_headers: dict[str, str],
) -> None:
"""Test filtering posts by tag returns matching items.
TC-API-010: Positive — public endpoint.
"""
payload = {
"title": "Tagged Test Post",
"content": "Post with a specific tag for testing. Enough characters here.",
"tags": ["unique-test-tag-xyz"],
}
client.post(API_PREFIX, json=payload, headers=user_headers)
response = client.get(f"{API_PREFIX}/by-tag/unique-test-tag-xyz")
assert response.status_code == 200
data = response.json()
assert len(data["items"]) >= 1
def test_get_posts_by_tag_no_results(
self,
client: TestClient,
) -> None:
"""Test filtering by nonexistent tag returns empty list."""
response = client.get(f"{API_PREFIX}/by-tag/nonexistent-tag-xyz-123")
assert response.status_code == 200
data = response.json()
assert data["items"] == []
class TestGetPostsByAuthor:
"""Tests for GET /api/v1/posts/by-author/{author_id} — posts by author."""
def test_get_posts_by_author_success(
self,
client: TestClient,
user_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test filtering posts by author returns their posts.
TC-API-011: Positive — public endpoint.
"""
response = client.get(f"{API_PREFIX}/by-author/{USER_ID}")
assert response.status_code == 200
data = response.json()
assert len(data["items"]) >= 1
assert all(item["author_id"] == USER_ID for item in data["items"])
def test_get_posts_by_author_no_results(
self,
client: TestClient,
) -> None:
"""Test filtering by nonexistent author returns empty list."""
response = client.get(f"{API_PREFIX}/by-author/nonexistent-author-123")
assert response.status_code == 200
data = response.json()
assert data["items"] == []
class TestGetPost:
"""Tests for GET /api/v1/posts/{post_id} — get post by ID."""
def test_get_post_by_id_success(
self,
client: TestClient,
created_post: dict[str, Any],
) -> None:
"""Test getting a post by ID returns 200 with correct data.
TC-API-012: Positive — public endpoint.
"""
post_id = created_post["id"]
response = client.get(f"{API_PREFIX}/{post_id}")
assert response.status_code == 200
data = response.json()
assert data["id"] == post_id
assert data["title"] == created_post["title"]
assert data["author_id"] == USER_ID
def test_get_post_by_id_not_found(
self,
client: TestClient,
) -> None:
"""Test getting a nonexistent post returns 404.
TC-API-013: Negative — nonexistent post ID.
"""
fake_id = "00000000-0000-0000-0000-000000000000"
response = client.get(f"{API_PREFIX}/{fake_id}")
assert response.status_code == 404
error = response.json()
assert error["error"] == "NotFoundException"
def test_get_unpublished_post_by_id(
self,
client: TestClient,
created_post: dict[str, Any],
) -> None:
"""Test that unpublished posts are accessible by ID.
The current implementation does not filter by published status
for individual post retrieval.
"""
post_id = created_post["id"]
assert created_post["published"] is False
response = client.get(f"{API_PREFIX}/{post_id}")
assert response.status_code == 200
data = response.json()
assert data["published"] is False
class TestGetPostBySlug:
"""Tests for GET /api/v1/posts/slug/{slug} — get post by slug."""
def test_get_post_by_slug_success(
self,
client: TestClient,
created_post: dict[str, Any],
) -> None:
"""Test getting a post by slug returns 200 with correct data.
TC-API-014: Positive — public endpoint.
"""
slug = created_post["slug"]
response = client.get(f"{API_PREFIX}/slug/{slug}")
assert response.status_code == 200
data = response.json()
assert data["slug"] == slug
assert data["id"] == created_post["id"]
def test_get_post_by_slug_not_found(
self,
client: TestClient,
) -> None:
"""Test getting a post by nonexistent slug returns 404.
TC-API-015: Negative — nonexistent slug.
"""
response = client.get(f"{API_PREFIX}/slug/nonexistent-slug-xyz-123")
assert response.status_code == 404
error = response.json()
assert error["error"] == "NotFoundException"
class TestUpdatePost:
"""Tests for PATCH /api/v1/posts/{post_id} — update a post."""
def test_update_own_post_success(
self,
client: TestClient,
user_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test updating own post returns 200 with updated fields.
TC-API-016: Positive — owner updates own post.
"""
post_id = created_post["id"]
update_data = {"title": "Updated Title For Testing"}
response = client.patch(
f"{API_PREFIX}/{post_id}",
json=update_data,
headers=user_headers,
)
assert response.status_code == 200
data = response.json()
assert data["title"] == "Updated Title For Testing"
assert data["id"] == post_id
assert data["updated_at"] != created_post["updated_at"]
def test_update_own_post_all_fields(
self,
client: TestClient,
user_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test updating all fields of own post."""
post_id = created_post["id"]
update_data = {
"title": "Completely New Title Here",
"content": "Updated content with sufficient length for validation check.",
"tags": ["new-tag"],
}
response = client.patch(
f"{API_PREFIX}/{post_id}",
json=update_data,
headers=user_headers,
)
assert response.status_code == 200
data = response.json()
assert data["title"] == update_data["title"]
assert data["content"] == update_data["content"]
assert data["tags"] == update_data["tags"]
def test_update_other_user_post_returns_403(
self,
client: TestClient,
user2_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test updating another user's post returns 403.
TC-API-017: Policy — user2 cannot update user's post.
"""
post_id = created_post["id"]
update_data = {"title": "Unauthorized Update Attempt"}
response = client.patch(
f"{API_PREFIX}/{post_id}",
json=update_data,
headers=user2_headers,
)
assert response.status_code == 403
error = response.json()
assert error["error"] == "ForbiddenException"
def test_update_post_admin_can_update_any(
self,
client: TestClient,
admin_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test admin can update any user's post."""
post_id = created_post["id"]
update_data = {"title": "Admin Updated This Post Title"}
response = client.patch(
f"{API_PREFIX}/{post_id}",
json=update_data,
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["title"] == "Admin Updated This Post Title"
def test_update_post_no_auth(
self,
client: TestClient,
created_post: dict[str, Any],
) -> None:
"""Test updating a post without auth returns 401.
TC-API-018: Negative — no authorization header.
"""
post_id = created_post["id"]
response = client.patch(
f"{API_PREFIX}/{post_id}",
json={"title": "No Auth Update Attempt"},
)
assert response.status_code == 401
error = response.json()
assert error["error"] == "UnauthorizedException"
def test_update_post_guest_token(
self,
client: TestClient,
guest_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test updating a post with guest token returns 401."""
post_id = created_post["id"]
response = client.patch(
f"{API_PREFIX}/{post_id}",
json={"title": "Guest Update Attempt"},
headers=guest_headers,
)
assert response.status_code == 401
error = response.json()
assert error["error"] == "UnauthorizedException"
class TestDeletePost:
"""Tests for DELETE /api/v1/posts/{post_id} — delete a post."""
def test_delete_own_post_success(
self,
client: TestClient,
user_headers: dict[str, str],
post_payload: dict[str, Any],
) -> None:
"""Test deleting own post returns 204.
TC-API-019: Positive — owner deletes own post.
"""
create_resp = client.post(API_PREFIX, json=post_payload, headers=user_headers)
post_id = create_resp.json()["id"]
response = client.delete(f"{API_PREFIX}/{post_id}", headers=user_headers)
assert response.status_code == 204
get_response = client.get(f"{API_PREFIX}/{post_id}")
assert get_response.status_code == 404
def test_delete_other_user_post_returns_403(
self,
client: TestClient,
user2_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test deleting another user's post returns 403.
TC-API-020: Policy — user2 cannot delete user's post.
"""
post_id = created_post["id"]
response = client.delete(f"{API_PREFIX}/{post_id}", headers=user2_headers)
assert response.status_code == 403
error = response.json()
assert error["error"] == "ForbiddenException"
def test_delete_post_admin_can_delete_any(
self,
client: TestClient,
admin_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test admin can delete any user's post."""
post_id = created_post["id"]
response = client.delete(f"{API_PREFIX}/{post_id}", headers=admin_headers)
assert response.status_code == 204
get_response = client.get(f"{API_PREFIX}/{post_id}")
assert get_response.status_code == 404
def test_delete_post_no_auth(
self,
client: TestClient,
created_post: dict[str, Any],
) -> None:
"""Test deleting a post without auth returns 401.
TC-API-021: Negative — no authorization header.
"""
post_id = created_post["id"]
response = client.delete(f"{API_PREFIX}/{post_id}")
assert response.status_code == 401
error = response.json()
assert error["error"] == "UnauthorizedException"
def test_delete_post_guest_token(
self,
client: TestClient,
guest_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test deleting a post with guest token returns 401."""
post_id = created_post["id"]
response = client.delete(f"{API_PREFIX}/{post_id}", headers=guest_headers)
assert response.status_code == 401
error = response.json()
assert error["error"] == "UnauthorizedException"
class TestPublishPost:
"""Tests for POST /api/v1/posts/{post_id}/publish — publish a post."""
def test_publish_own_post_success(
self,
client: TestClient,
user_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test publishing own post returns 200 with published=True.
TC-API-022: Positive — owner publishes own post.
"""
post_id = created_post["id"]
assert created_post["published"] is False
response = client.post(
f"{API_PREFIX}/{post_id}/publish",
headers=user_headers,
)
assert response.status_code == 200
data = response.json()
assert data["published"] is True
assert data["id"] == post_id
def test_publish_other_user_post_returns_403(
self,
client: TestClient,
user2_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test publishing another user's post returns 403.
TC-API-023: Policy — user2 cannot publish user's post.
"""
post_id = created_post["id"]
response = client.post(
f"{API_PREFIX}/{post_id}/publish",
headers=user2_headers,
)
assert response.status_code == 403
error = response.json()
assert error["error"] == "ForbiddenException"
def test_publish_admin_can_publish_any(
self,
client: TestClient,
admin_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test admin can publish any user's post."""
post_id = created_post["id"]
response = client.post(
f"{API_PREFIX}/{post_id}/publish",
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["published"] is True
def test_publish_post_no_auth(
self,
client: TestClient,
created_post: dict[str, Any],
) -> None:
"""Test publishing a post without auth returns 401."""
post_id = created_post["id"]
response = client.post(f"{API_PREFIX}/{post_id}/publish")
assert response.status_code == 401
error = response.json()
assert error["error"] == "UnauthorizedException"
def test_publish_post_guest_token(
self,
client: TestClient,
guest_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test publishing a post with guest token returns 401."""
post_id = created_post["id"]
response = client.post(
f"{API_PREFIX}/{post_id}/publish",
headers=guest_headers,
)
assert response.status_code == 401
error = response.json()
assert error["error"] == "UnauthorizedException"
def test_publish_post_not_found(
self,
client: TestClient,
user_headers: dict[str, str],
) -> None:
"""Test publishing a nonexistent post returns 404."""
fake_id = "00000000-0000-0000-0000-000000000000"
response = client.post(
f"{API_PREFIX}/{fake_id}/publish",
headers=user_headers,
)
assert response.status_code == 404
error = response.json()
assert error["error"] == "NotFoundException"
class TestUnpublishPost:
"""Tests for POST /api/v1/posts/{post_id}/unpublish — unpublish a post."""
def test_unpublish_own_post_success(
self,
client: TestClient,
user_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test unpublishing own post returns 200 with published=False.
TC-API-024: Positive — owner unpublishes own post.
"""
post_id = created_post["id"]
client.post(f"{API_PREFIX}/{post_id}/publish", headers=user_headers)
response = client.post(
f"{API_PREFIX}/{post_id}/unpublish",
headers=user_headers,
)
assert response.status_code == 200
data = response.json()
assert data["published"] is False
assert data["id"] == post_id
def test_unpublish_other_user_post_returns_403(
self,
client: TestClient,
user2_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test unpublishing another user's post returns 403.
TC-API-025: Policy — user2 cannot unpublish user's post.
"""
post_id = created_post["id"]
response = client.post(
f"{API_PREFIX}/{post_id}/unpublish",
headers=user2_headers,
)
assert response.status_code == 403
error = response.json()
assert error["error"] == "ForbiddenException"
def test_unpublish_admin_can_unpublish_any(
self,
client: TestClient,
admin_headers: dict[str, str],
created_post: dict[str, Any],
) -> None:
"""Test admin can unpublish any user's post."""
post_id = created_post["id"]
client.post(f"{API_PREFIX}/{post_id}/publish", headers=admin_headers)
response = client.post(
f"{API_PREFIX}/{post_id}/unpublish",
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert data["published"] is False
def test_unpublish_post_not_found(
self,
client: TestClient,
user_headers: dict[str, str],
) -> None:
"""Test unpublishing a nonexistent post returns 404."""
fake_id = "00000000-0000-0000-0000-000000000000"
response = client.post(
f"{API_PREFIX}/{fake_id}/unpublish",
headers=user_headers,
)
assert response.status_code == 404
error = response.json()
assert error["error"] == "NotFoundException"
def test_unpublish_post_no_auth(
self,
client: TestClient,
created_post: dict[str, Any],
) -> None:
"""Test unpublishing a post without auth returns 401."""
post_id = created_post["id"]
response = client.post(f"{API_PREFIX}/{post_id}/unpublish")
assert response.status_code == 401
error = response.json()
assert error["error"] == "UnauthorizedException"