"""Tests for DMS Phase 18 — Realtime Integration (WebSocket / Django Channels). Covers: * dms/realtime.py — room name helpers + publish functions * simorgh/core/realtime — ChannelsRealtimeBackend importable * dms/routing.py — websocket_urlpatterns defined * dms/consumers/ — DocumentConsumer, CommentsConsumer, DmsNotificationConsumer * dms/signals.py — post_save → realtime publish integration """ from __future__ import annotations import pytest from channels.layers import InMemoryChannelLayer, get_channel_layer from channels.routing import URLRouter from channels.testing import WebsocketCommunicator from django.urls import re_path from simorgh.apps.dms.consumers.comments import CommentsConsumer from simorgh.apps.dms.consumers.document import DocumentConsumer from simorgh.apps.dms.consumers.notifications import DmsNotificationConsumer from simorgh.apps.dms.realtime import ( comments_room, document_room, publish_comment_event, publish_document_event, publish_user_notification, user_notifications_room, ) from simorgh.apps.dms.routing import websocket_urlpatterns from simorgh.core.realtime import MemoryRealtimeBackend, set_backend # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- class _InjectUser: """Minimal ASGI middleware that injects a user into the scope.""" def __init__(self, app, user): self.app = app self.user = user async def __call__(self, scope, receive, send): await self.app({**scope, "user": self.user}, receive, send) def _consumer_app(consumer_cls, user, extra_patterns=None): """Build a minimal URLRouter application wrapping one consumer class.""" patterns = extra_patterns or websocket_urlpatterns return _InjectUser(URLRouter(patterns), user) def _doc_app(user, public_id=None): pattern = re_path( r"^ws/dms/documents/(?P[0-9a-f-]+)/$", DocumentConsumer.as_asgi(), ) return _InjectUser(URLRouter([pattern]), user) def _comments_app(user): pattern = re_path( r"^ws/dms/documents/(?P[0-9a-f-]+)/comments/$", CommentsConsumer.as_asgi(), ) return _InjectUser(URLRouter([pattern]), user) def _notifications_app(user): pattern = re_path(r"^ws/dms/notifications/$", DmsNotificationConsumer.as_asgi()) return _InjectUser(URLRouter([pattern]), user) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def _mem_backend(): """Use MemoryRealtimeBackend for all tests in this module.""" backend = MemoryRealtimeBackend() set_backend(backend) yield backend set_backend(None) @pytest.fixture def repo(tenant_acme, acme_tree): from simorgh.apps.dms.repositories.models import Repository return Repository.objects.create( name="Phase 18 Repo", slug="ph18-repo", tenant=tenant_acme, organization_node=acme_tree["root"], ) @pytest.fixture def doc(repo, tenant_acme, acme_tree): from simorgh.apps.dms.documents.models import Document, DocumentStatus return Document.objects.create( title="Phase 18 Document", repository=repo, status=DocumentStatus.DRAFT, tenant=tenant_acme, organization_node=acme_tree["root"], ) @pytest.fixture def anonymous_user(): """A user-like object that is not authenticated.""" class _Anon: is_authenticated = False pk = None return _Anon() # --------------------------------------------------------------------------- # 1. Room name helpers # --------------------------------------------------------------------------- def test_document_room_name_format(): room = document_room("abc-123") assert room == "dms_doc_abc-123" assert ":" not in room # valid channel group name def test_comments_room_name_format(): room = comments_room("abc-123") assert room == "dms_comments_abc-123" assert ":" not in room def test_user_notifications_room_name_format(): room = user_notifications_room(42) assert room == "dms_user_42" assert ":" not in room # --------------------------------------------------------------------------- # 2. Publish helpers delegate to the backend # --------------------------------------------------------------------------- def test_publish_document_event_writes_to_backend(_mem_backend): publish_document_event("uuid-1", "document.updated", {"id": "uuid-1"}) room = document_room("uuid-1") assert room in _mem_backend.messages events = [e for e, _ in _mem_backend.messages[room]] assert "document.updated" in events def test_publish_comment_event_writes_to_backend(_mem_backend): publish_comment_event("uuid-2", "comment.created", {"id": "cmt-1"}) room = comments_room("uuid-2") assert room in _mem_backend.messages def test_publish_user_notification_writes_to_backend(_mem_backend): publish_user_notification(7, "notification.dms", {"title": "Hello"}) room = user_notifications_room(7) assert room in _mem_backend.messages # --------------------------------------------------------------------------- # 3. ChannelsRealtimeBackend importable # --------------------------------------------------------------------------- def test_channels_backend_importable(): from simorgh.core.realtime import ChannelsRealtimeBackend assert callable(getattr(ChannelsRealtimeBackend, "publish", None)) # --------------------------------------------------------------------------- # 4. websocket_urlpatterns defined # --------------------------------------------------------------------------- def test_routing_has_three_url_patterns(): assert len(websocket_urlpatterns) == 3 # --------------------------------------------------------------------------- # 5. Signal → realtime backend integration # --------------------------------------------------------------------------- @pytest.mark.django_db def test_document_save_broadcasts_created_event(doc, _mem_backend): """Saving a new document fires document.created via MemoryRealtimeBackend.""" room = document_room(str(doc.public_id)) events = [e for e, _ in _mem_backend.messages.get(room, [])] assert "document.created" in events @pytest.mark.django_db def test_document_save_broadcasts_updated_event(doc, _mem_backend): """Updating an existing document fires document.updated.""" _mem_backend.reset() doc.title = "Updated Title" doc.save(update_fields=["title", "updated_at"]) room = document_room(str(doc.public_id)) events = [e for e, _ in _mem_backend.messages.get(room, [])] assert "document.updated" in events @pytest.mark.django_db def test_version_save_broadcasts_version_added(doc, tenant_acme, acme_tree, _mem_backend): """Adding a version fires version.added to the document room.""" from simorgh.apps.dms.documents.services import add_version _mem_backend.reset() add_version( document=doc, bump="minor", tenant_id=tenant_acme.pk, organization_node_id=acme_tree["root"].pk, ) room = document_room(str(doc.public_id)) events = [e for e, _ in _mem_backend.messages.get(room, [])] assert "version.added" in events @pytest.mark.django_db def test_comment_save_broadcasts_comment_created(doc, alice, tenant_acme, acme_tree, _mem_backend): """Creating a comment fires comment.created to the comments room.""" from simorgh.apps.dms.collaboration.models import Comment _mem_backend.reset() Comment.objects.create( document=doc, author=alice, body="Hello!", tenant=tenant_acme, organization_node=acme_tree["root"], ) room = comments_room(str(doc.public_id)) events = [e for e, _ in _mem_backend.messages.get(room, [])] assert "comment.created" in events # --------------------------------------------------------------------------- # 6. DocumentConsumer — WebSocket consumer tests (async) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_document_consumer_rejects_unauthenticated(anonymous_user): app = _doc_app(anonymous_user) # Use a valid hex UUID so the URL regex matches. communicator = WebsocketCommunicator(app, "ws/dms/documents/aaaabbbb-cccc-dddd-eeee-000000000001/") connected, subprotocol = await communicator.connect() assert not connected await communicator.disconnect() @pytest.mark.asyncio @pytest.mark.django_db(transaction=True) async def test_document_consumer_accepts_authenticated(alice): app = _doc_app(alice) public_id = "aaaabbbb-cccc-dddd-eeee-ffff00001111" communicator = WebsocketCommunicator(app, f"ws/dms/documents/{public_id}/") connected, _ = await communicator.connect() assert connected await communicator.disconnect() @pytest.mark.asyncio @pytest.mark.django_db(transaction=True) async def test_document_consumer_receives_group_broadcast(alice): # Use the global channel layer — same instance the consumer uses. layer = get_channel_layer() app = _doc_app(alice) public_id = "aaaabbbb-cccc-dddd-eeee-ffff00001111" communicator = WebsocketCommunicator(app, f"ws/dms/documents/{public_id}/") connected, _ = await communicator.connect() assert connected # Drain the presence.join message first. _ = await communicator.receive_json_from() # Push a broadcast directly to the group via the same global layer. group = document_room(public_id) await layer.group_send( group, {"type": "broadcast.message", "event": "document.updated", "payload": {"title": "X"}}, ) msg = await communicator.receive_json_from() assert msg["event"] == "document.updated" assert msg["payload"]["title"] == "X" await communicator.disconnect() # --------------------------------------------------------------------------- # 7. CommentsConsumer tests # --------------------------------------------------------------------------- @pytest.mark.asyncio @pytest.mark.django_db(transaction=True) async def test_comments_consumer_accepts_authenticated(alice): app = _comments_app(alice) public_id = "aaaabbbb-cccc-dddd-eeee-ffff00001111" communicator = WebsocketCommunicator(app, f"ws/dms/documents/{public_id}/comments/") connected, _ = await communicator.connect() assert connected await communicator.disconnect() @pytest.mark.asyncio @pytest.mark.django_db(transaction=True) async def test_comments_consumer_receives_broadcast(alice): layer = get_channel_layer() app = _comments_app(alice) public_id = "aaaabbbb-cccc-dddd-eeee-ffff00001111" communicator = WebsocketCommunicator(app, f"ws/dms/documents/{public_id}/comments/") connected, _ = await communicator.connect() assert connected group = comments_room(public_id) await layer.group_send( group, {"type": "broadcast.message", "event": "comment.created", "payload": {"id": "c1"}}, ) msg = await communicator.receive_json_from() assert msg["event"] == "comment.created" await communicator.disconnect() # --------------------------------------------------------------------------- # 8. DmsNotificationConsumer tests # --------------------------------------------------------------------------- @pytest.mark.asyncio @pytest.mark.django_db(transaction=True) async def test_notifications_consumer_accepts_authenticated(alice): app = _notifications_app(alice) communicator = WebsocketCommunicator(app, "ws/dms/notifications/") connected, _ = await communicator.connect() assert connected await communicator.disconnect() @pytest.mark.asyncio @pytest.mark.django_db(transaction=True) async def test_notifications_consumer_receives_broadcast(alice): layer = get_channel_layer() app = _notifications_app(alice) communicator = WebsocketCommunicator(app, "ws/dms/notifications/") connected, _ = await communicator.connect() assert connected group = user_notifications_room(alice.pk) await layer.group_send( group, { "type": "broadcast.message", "event": "notification.dms", "payload": {"title": "Document shared"}, }, ) msg = await communicator.receive_json_from() assert msg["event"] == "notification.dms" assert msg["payload"]["title"] == "Document shared" await communicator.disconnect() @pytest.mark.asyncio async def test_notifications_consumer_rejects_unauthenticated(anonymous_user): app = _notifications_app(anonymous_user) communicator = WebsocketCommunicator(app, "ws/dms/notifications/") connected, _ = await communicator.connect() assert not connected await communicator.disconnect()