Amirhossein Khalili
ff99130938
Split backend tests into unit and integration suites, add isolated fixtures and coverage enforcement, and update CI and test documentation. Cover current authentication, content, notification, payment, media, and admin flows; fix behaviors revealed by the new tests. Validate both suites and the per-module gate against local PostgreSQL.
52 lines
2.7 KiB
Python
52 lines
2.7 KiB
Python
import json
|
|
|
|
import pytest
|
|
|
|
from apps.notifications import services
|
|
|
|
|
|
pytestmark = [pytest.mark.integration, pytest.mark.django_db]
|
|
|
|
|
|
def test_notification_api_lifecycle_uses_authenticated_redis_namespace(client, user_factory, auth_headers,
|
|
fake_notification_redis, monkeypatch):
|
|
monkeypatch.setattr(services, "redis_client", fake_notification_redis)
|
|
owner = user_factory()
|
|
other = user_factory()
|
|
item = services.RedisNotificationStore.add(str(owner.pk), {"type": "event", "title": "Workshop"})
|
|
path = "/api/notifications/"
|
|
assert client.get(path).status_code == 401
|
|
assert client.get(path, **auth_headers(other)).json()["notifications"] == []
|
|
listing = client.get(path, {"status": "unread", "limit": 1}, **auth_headers(owner))
|
|
assert listing.status_code == 200 and listing.json()["unread_count"] == 1
|
|
assert listing.json()["notifications"][0]["id"] == item["id"]
|
|
assert client.get(path, {"status": "invalid"}, **auth_headers(owner)).status_code == 400
|
|
assert client.post(path + "mark-seen", data=json.dumps({"id": "unknown"}),
|
|
content_type="application/json", **auth_headers(owner)).status_code == 404
|
|
seen = client.post(path + "mark-seen", data=json.dumps({"id": item["id"]}),
|
|
content_type="application/json", **auth_headers(owner))
|
|
assert seen.status_code == 200 and seen.json()["unread_count"] == 0
|
|
assert client.get(path, {"status": "read"}, **auth_headers(owner)).json()["count"] == 1
|
|
assert client.post(path + "mark-all-read", **auth_headers(owner)).json()["marked_read"] == 1
|
|
assert services.RedisNotificationStore.unread_count(str(owner.pk)) == 0
|
|
assert client.delete(path + item["id"], **auth_headers(owner)).json()["deleted"]
|
|
assert services.RedisNotificationStore.get(str(owner.pk), item["id"]) is None
|
|
|
|
|
|
def test_stream_token_endpoint_and_sse_auth(client, user_factory, auth_headers, settings):
|
|
user = user_factory()
|
|
headers = auth_headers(user)
|
|
settings.NOTIFICATIONS_ENABLED = False
|
|
assert client.post("/api/notifications/stream-token", **headers).status_code == 503
|
|
assert client.get("/api/notifications/stream/").status_code == 503
|
|
settings.NOTIFICATIONS_ENABLED = True
|
|
issued = client.post("/api/notifications/stream-token", **headers)
|
|
assert issued.status_code == 200, issued.content
|
|
token = issued.json()["token"]
|
|
assert client.get("/api/notifications/stream/").status_code == 401
|
|
assert client.get("/api/notifications/stream/", {"token": "bad"}).status_code == 401
|
|
response = client.get("/api/notifications/stream/", {"token": token})
|
|
assert response.status_code == 200
|
|
assert response["Content-Type"] == "text/event-stream"
|
|
response.close()
|