# tests/realtime/test_websocket_events.py """ Тесты для WebSocket событий из API. Проверяем что Consumer правильно обрабатывает события от channel_layer: - Создание сообщения - Редактирование сообщения - Удаление сообщения - Добавление/удаление реакций - Голосование в опросе - Отметка прочтения Тесты не создают реальные Django объекты, а проверяют только правильность обработки событий Consumer'ом. """ import asyncio import pytest from channels.db import database_sync_to_async from channels.layers import get_channel_layer from notifications.models import Notification pytestmark = pytest.mark.django_db(transaction=True) class TestMessageWebSocketEvents: """Тесты WebSocket событий для сообщений.""" @pytest.mark.asyncio async def test_new_message_broadcast( self, ws_communicator, user, user2, test_chat_with_users ): """При создании нового сообщения оба участника получают событие.""" # Подключаем обоих пользователей comm1 = await ws_communicator(user=user) comm2 = await ws_communicator(user=user2) connected1, _ = await comm1.connect() connected2, _ = await comm2.connect() assert connected1 and connected2 # User1 открывает чат await comm1.send_json_to( { "action": "open_chat", "chat_id": test_chat_with_users.id, "load_history": False, } ) # User2 тоже открывает чат await comm2.send_json_to( { "action": "open_chat", "chat_id": test_chat_with_users.id, "load_history": False, } ) # Подтверждения открытия await comm1.receive_json_from(timeout=5) await comm2.receive_json_from(timeout=5) # Отправляем событие через channel layer (имитируем API) channel_layer = get_channel_layer() mock_payload = { "id": 999, "content": "Test message", "author_id": user.id, "author_name": user.get_full_name(), "chat_id": test_chat_with_users.id, "created_at": "2026-01-21T10:00:00Z", } await channel_layer.group_send( f"chat_{test_chat_with_users.id}", { "type": "chat.message", "chat_id": test_chat_with_users.id, "payload": mock_payload, }, ) # User1 должен получить new_message (активный чат) response1 = await comm1.receive_json_from(timeout=5) assert response1["type"] == "new_message" assert response1["chat_id"] == test_chat_with_users.id assert response1["message"]["content"] == "Test message" # И list_update для списка чатов response1_list = await comm1.receive_json_from(timeout=5) assert response1_list["type"] == "list_update" # User2 тоже должен получить оба события response2 = await comm2.receive_json_from(timeout=5) assert response2["type"] == "new_message" response2_list = await comm2.receive_json_from(timeout=5) assert response2_list["type"] == "list_update" await comm1.disconnect() await comm2.disconnect() @pytest.mark.asyncio async def test_new_message_list_update_only( self, ws_communicator, user, user2, test_chat_with_users ): """Если чат неактивен, пользователь получает только list_update.""" comm = await ws_communicator(user=user) connected, _ = await comm.connect() assert connected # НЕ открываем чат (чтобы он был неактивным) # Отправляем событие (имитируем API) channel_layer = get_channel_layer() mock_payload = { "id": 998, "content": "Message when chat is inactive", "author_id": user2.id, "author_name": user2.get_full_name(), "chat_id": test_chat_with_users.id, } await channel_layer.group_send( f"chat_{test_chat_with_users.id}", { "type": "chat.message", "chat_id": test_chat_with_users.id, "payload": mock_payload, }, ) # Когда чат неактивен, Consumer отправляет только list_update response = await comm.receive_json_from(timeout=5) assert response["type"] == "list_update" assert response["chat_id"] == test_chat_with_users.id @pytest.mark.asyncio async def test_message_edited_broadcast(self, ws_communicator, user, test_chat): """При редактировании сообщения участники получают событие.""" comm = await ws_communicator(user=user) connected, _ = await comm.connect() assert connected # Открываем чат await comm.send_json_to( {"action": "open_chat", "chat_id": test_chat.id, "load_history": False} ) await comm.receive_json_from(timeout=5) # chat_opened # Отправляем событие редактирования (имитируем API) channel_layer = get_channel_layer() mock_payload = { "id": 997, "content": "Edited content", "is_edited": True, "author_id": user.id, "chat_id": test_chat.id, } await channel_layer.group_send( f"chat_{test_chat.id}", { "type": "chat.message_edited", "chat_id": test_chat.id, "payload": mock_payload, }, ) # Должны получить message_updated для активного чата response1 = await comm.receive_json_from(timeout=5) assert response1["type"] == "message_updated" assert response1["message"]["content"] == "Edited content" assert response1["message"]["is_edited"] is True # И message_edited для списка response2 = await comm.receive_json_from(timeout=5) assert response2["type"] == "message_edited" await comm.disconnect() @pytest.mark.asyncio async def test_message_deleted_broadcast(self, ws_communicator, user, test_chat): """При удалении сообщения участники получают событие.""" comm = await ws_communicator(user=user) connected, _ = await comm.connect() assert connected # Открываем чат await comm.send_json_to( {"action": "open_chat", "chat_id": test_chat.id, "load_history": False} ) await comm.receive_json_from(timeout=5) # chat_opened # Отправляем событие удаления (имитируем API) channel_layer = get_channel_layer() mock_message_id = 996 await channel_layer.group_send( f"chat_{test_chat.id}", { "type": "chat.message_deleted", "chat_id": test_chat.id, "message_id": mock_message_id, "author_id": user.id, }, ) # Должны получить message_deleted response = await comm.receive_json_from(timeout=5) assert response["type"] == "message_deleted" assert response["chat_id"] == test_chat.id assert response["message_id"] == mock_message_id list_response = await comm.receive_json_from(timeout=5) assert list_response["type"] == "list_update" assert list_response["event"] == "message_deleted" assert list_response["chat_id"] == test_chat.id assert list_response["author_id"] == user.id await comm.disconnect() @pytest.mark.asyncio async def test_message_deleted_inactive_chat( self, ws_communicator, user, test_chat ): """Удаление в неактивном чате обновляет его строку в списке.""" comm = await ws_communicator(user=user) connected, _ = await comm.connect() assert connected # НЕ открываем чат # Отправляем событие удаления (имитируем API) channel_layer = get_channel_layer() mock_message_id = 995 await channel_layer.group_send( f"chat_{test_chat.id}", { "type": "chat.message_deleted", "chat_id": test_chat.id, "message_id": mock_message_id, "author_id": user.id, }, ) response = await comm.receive_json_from(timeout=5) assert response["type"] == "list_update" assert response["event"] == "message_deleted" assert response["chat_id"] == test_chat.id assert response["message_id"] == mock_message_id assert response["author_id"] == user.id class TestReactionWebSocketEvents: """Тесты WebSocket событий для реакций.""" @pytest.mark.asyncio async def test_reaction_added_broadcast( self, ws_communicator, user, user2, test_chat_with_users ): """При добавлении реакции участники получают событие.""" # Подключаем обоих пользователей comm1 = await ws_communicator(user=user) comm2 = await ws_communicator(user=user2) connected1, _ = await comm1.connect() connected2, _ = await comm2.connect() assert connected1 and connected2 # Оба открывают чат await comm1.send_json_to( { "action": "open_chat", "chat_id": test_chat_with_users.id, "load_history": False, } ) await comm2.send_json_to( { "action": "open_chat", "chat_id": test_chat_with_users.id, "load_history": False, } ) await comm1.receive_json_from(timeout=5) # chat_opened await comm2.receive_json_from(timeout=5) # chat_opened # Отправляем событие реакции (имитируем API) channel_layer = get_channel_layer() mock_message_id = 994 reactions_summary = { "👍": { "count": 1, "users": [user2.id], "user_names": [user2.get_full_name()], "user_details": [ { "id": user2.id, "name": user2.get_full_name(), "avatar": None, } ], } } await channel_layer.group_send( f"chat_{test_chat_with_users.id}", { "type": "chat.reaction_added", "chat_id": test_chat_with_users.id, "message_id": mock_message_id, "user_id": user2.id, "emoji": "👍", "reactions_summary": reactions_summary, }, ) # User1 должен получить reaction_added response1 = await comm1.receive_json_from(timeout=5) assert response1["type"] == "reaction_added" assert response1["message_id"] == mock_message_id assert response1["emoji"] == "👍" assert response1["user_id"] == user2.id assert "👍" in response1["reactions_summary"] assert response1["reactions_summary"]["👍"]["user_details"][0]["id"] == user2.id # User2 тоже должен получить response2 = await comm2.receive_json_from(timeout=5) assert response2["type"] == "reaction_added" assert response2["emoji"] == "👍" await comm1.disconnect() await comm2.disconnect() @pytest.mark.asyncio async def test_reaction_removed_broadcast(self, ws_communicator, user, test_chat): """При удалении реакции участники получают событие.""" comm = await ws_communicator(user=user) connected, _ = await comm.connect() assert connected # Открываем чат await comm.send_json_to( {"action": "open_chat", "chat_id": test_chat.id, "load_history": False} ) await comm.receive_json_from(timeout=5) # chat_opened # Отправляем событие удаления реакции (имитируем API) channel_layer = get_channel_layer() mock_message_id = 993 reactions_summary = {} # Пустой - все реакции удалены await channel_layer.group_send( f"chat_{test_chat.id}", { "type": "chat.reaction_removed", "chat_id": test_chat.id, "message_id": mock_message_id, "user_id": user.id, "emoji": "👍", "reactions_summary": reactions_summary, }, ) # Должны получить reaction_removed response = await comm.receive_json_from(timeout=5) assert response["type"] == "reaction_removed" assert response["message_id"] == mock_message_id assert response["emoji"] == "👍" assert response["reactions_summary"] == {} await comm.disconnect() @pytest.mark.asyncio async def test_reaction_inactive_chat_no_event( self, ws_communicator, user, test_chat ): """Если чат неактивен, событие реакции НЕ отправляется.""" comm = await ws_communicator(user=user) connected, _ = await comm.connect() assert connected # НЕ открываем чат # Отправляем событие реакции (имитируем API) channel_layer = get_channel_layer() mock_message_id = 992 await channel_layer.group_send( f"chat_{test_chat.id}", { "type": "chat.reaction_added", "chat_id": test_chat.id, "message_id": mock_message_id, "user_id": user.id, "emoji": "👍", "reactions_summary": {}, }, ) # НЕ должны получить событие import asyncio with pytest.raises(asyncio.TimeoutError): await comm.receive_json_from(timeout=2) class TestPollWebSocketEvents: """Тесты WebSocket событий для голосований.""" @pytest.mark.asyncio async def test_poll_vote_broadcast( self, ws_communicator, user, user2, test_chat_with_users ): """При голосовании участники получают обновление.""" comm1 = await ws_communicator(user=user) comm2 = await ws_communicator(user=user2) connected1, _ = await comm1.connect() connected2, _ = await comm2.connect() assert connected1 and connected2 # Оба открывают чат await comm1.send_json_to( { "action": "open_chat", "chat_id": test_chat_with_users.id, "load_history": False, } ) await comm2.send_json_to( { "action": "open_chat", "chat_id": test_chat_with_users.id, "load_history": False, } ) await comm1.receive_json_from(timeout=5) await comm2.receive_json_from(timeout=5) # Отправляем событие обновления опроса (имитируем API) channel_layer = get_channel_layer() mock_poll_id = 991 mock_message_id = 990 await channel_layer.group_send( f"chat_{test_chat_with_users.id}", { "type": "chat.poll_update", "chat_id": test_chat_with_users.id, "payload": { "poll_id": mock_poll_id, "message_id": mock_message_id, "total_voters": 1, "results": {"options": [{"id": 1, "vote_count": 1}]}, }, }, ) # User1 должен получить poll_update response1 = await comm1.receive_json_from(timeout=5) assert response1["type"] == "poll_update" assert response1["poll_id"] == mock_poll_id assert response1["message_id"] == mock_message_id # User2 тоже response2 = await comm2.receive_json_from(timeout=5) assert response2["type"] == "poll_update" await comm1.disconnect() await comm2.disconnect() @pytest.mark.asyncio async def test_poll_update_inactive_chat(self, ws_communicator, user, test_chat): """Если чат неактивен, событие опроса НЕ отправляется.""" comm = await ws_communicator(user=user) connected, _ = await comm.connect() assert connected # НЕ открываем чат # Отправляем событие (имитируем API) channel_layer = get_channel_layer() mock_poll_id = 989 await channel_layer.group_send( f"chat_{test_chat.id}", { "type": "chat.poll_update", "chat_id": test_chat.id, "payload": {"poll_id": mock_poll_id}, }, ) # НЕ должны получить import asyncio with pytest.raises(asyncio.TimeoutError): await comm.receive_json_from(timeout=2) @pytest.mark.asyncio async def test_vote_poll_action_updates_all_active_participants( self, ws_communicator, user, user2, test_chat_with_users, ): def create_poll(): from communications.models import Message, Poll, PollOption message = Message.objects.create( chat=test_chat_with_users, author=user2, content="Выберите вариант", ) poll = Poll.objects.create( message=message, author=user2, question="Выберите вариант", ) option = PollOption.objects.create( poll=poll, text="Первый", position=0, ) PollOption.objects.create( poll=poll, text="Второй", position=1, ) return poll.id, message.id, option.id poll_id, message_id, option_id = await database_sync_to_async(create_poll)() comm_voter = await ws_communicator(user=user) comm_other = await ws_communicator(user=user2) assert (await comm_voter.connect())[0] assert (await comm_other.connect())[0] for communicator in (comm_voter, comm_other): await communicator.send_json_to( { "action": "open_chat", "chat_id": test_chat_with_users.id, "load_history": False, } ) opened = await communicator.receive_json_from(timeout=5) assert opened["type"] == "chat_opened" await comm_voter.send_json_to( { "action": "vote_poll", "poll_id": poll_id, "option_ids": [option_id], } ) voter_public_update = await comm_voter.receive_json_from(timeout=5) voter_update = await comm_voter.receive_json_from(timeout=5) other_update = await comm_other.receive_json_from(timeout=5) for update in (voter_public_update, other_update): assert update["type"] == "poll_update" assert update["message_id"] == message_id assert update["voter_user_id"] is None assert update["selected_option_ids"] == [] assert update["results"]["total_voters"] == 1 assert update["results"]["options"][0]["vote_count"] == 1 assert voter_update["type"] == "poll_update" assert voter_update["message_id"] == message_id assert voter_update["voter_user_id"] == user.id assert voter_update["selected_option_ids"] == [option_id] await comm_voter.disconnect() await comm_other.disconnect() class TestChatMarkedReadEvent: """Тесты WebSocket события отметки прочтения.""" @pytest.mark.asyncio async def test_marked_read_sync_between_tabs( self, ws_communicator, user, test_chat ): """Отметка прочтения синхронизируется между вкладками читателя.""" # Имитируем две вкладки одного пользователя comm1 = await ws_communicator(user=user) comm2 = await ws_communicator(user=user) connected1, _ = await comm1.connect() connected2, _ = await comm2.connect() assert connected1 and connected2 await comm1.send_json_to( { "action": "open_chat", "chat_id": test_chat.id, "load_history": False, } ) await comm2.send_json_to( { "action": "open_chat", "chat_id": test_chat.id, "load_history": False, } ) opened1 = await comm1.receive_json_from(timeout=5) opened2 = await comm2.receive_json_from(timeout=5) assert opened1["type"] == "chat_opened" assert opened2["type"] == "chat_opened" # Отправляем sync-событие в пользовательский канал (как делает API) channel_layer = get_channel_layer() mock_message_id = 988 reader = { "id": user.id, "name": user.get_full_name() or user.username, "avatar": "", } await channel_layer.group_send( f"user_{user.id}", { "type": "chat_marked_read_sync", "chat_id": test_chat.id, "last_read_message_id": mock_message_id, "reader_user_id": user.id, "reader": reader, }, ) # Обе вкладки должны получить событие response1 = await comm1.receive_json_from(timeout=5) assert response1["type"] == "marked_read" assert response1["chat_id"] == test_chat.id assert response1["last_read_message_id"] == mock_message_id assert response1["reader_user_id"] == user.id assert response1["reader"] == reader response2 = await comm2.receive_json_from(timeout=5) assert response2["type"] == "marked_read" assert response2["chat_id"] == test_chat.id assert response2["last_read_message_id"] == mock_message_id assert response2["reader_user_id"] == user.id assert response2["reader"] == reader await comm1.disconnect() await comm2.disconnect() @pytest.mark.asyncio async def test_marked_read_receipt_is_sent_only_to_other_active_users( self, ws_communicator, user, user2, test_chat_with_users, ): """Read receipt из chat-группы видят другие участники, но не читатель.""" import asyncio comm_reader = await ws_communicator(user=user) comm_other = await ws_communicator(user=user2) connected1, _ = await comm_reader.connect() connected2, _ = await comm_other.connect() assert connected1 and connected2 await comm_reader.send_json_to( { "action": "open_chat", "chat_id": test_chat_with_users.id, "load_history": False, } ) await comm_other.send_json_to( { "action": "open_chat", "chat_id": test_chat_with_users.id, "load_history": False, } ) opened1 = await comm_reader.receive_json_from(timeout=5) opened2 = await comm_other.receive_json_from(timeout=5) assert opened1["type"] == "chat_opened" assert opened2["type"] == "chat_opened" # Отправляем read receipt в канал чата (как делает API для других участников) channel_layer = get_channel_layer() mock_message_id = 987 reader = { "id": user.id, "name": user.get_full_name() or user.username, "avatar": "", } await channel_layer.group_send( f"chat_{test_chat_with_users.id}", { "type": "chat_marked_read", "chat_id": test_chat_with_users.id, "last_read_message_id": mock_message_id, "reader_user_id": user.id, "reader": reader, }, ) # Другой участник должен получить receipt response_other = await comm_other.receive_json_from(timeout=5) assert response_other["type"] == "marked_read" assert response_other["chat_id"] == test_chat_with_users.id assert response_other["last_read_message_id"] == mock_message_id assert response_other["reader_user_id"] == user.id assert response_other["reader"] == reader # Сам читатель не должен получать receipt из chat-группы with pytest.raises(asyncio.TimeoutError): await comm_reader.receive_json_from(timeout=2) # Cleanup: after TimeoutError, communicator may be in cancelled state try: await comm_other.disconnect() except asyncio.CancelledError: pass try: await comm_reader.disconnect() except asyncio.CancelledError: pass class TestNotificationStateEvents: """Тесты синхронизации состояния уведомлений через user channel.""" @pytest.mark.asyncio async def test_active_chat_sessions_receive_notification_state_event( self, ws_communicator, user, test_chat, ): """Every session receives unread state; reading remains an HTTP command.""" comm_active = await ws_communicator(user=user) comm_background = await ws_communicator(user=user) connected_active, _ = await comm_active.connect() connected_background, _ = await comm_background.connect() assert connected_active and connected_background await comm_active.send_json_to( { "action": "open_chat", "chat_id": test_chat.id, "load_history": False, } ) await comm_active.receive_json_from(timeout=5) channel_layer = get_channel_layer() payload = { "type": "notification", "id": 777, "verb": "chat_new_message", "description": "New message in active chat", "action_url": f"/messages/{test_chat.id}", "timestamp": "2026-04-02T10:00:00Z", "unread": True, "data": { "chat_id": test_chat.id, "message_id": 123, }, } await channel_layer.group_send( f"user_{user.id}", { "type": "notification_message", "message": payload, }, ) active_response = await comm_active.receive_json_from(timeout=5) assert active_response["type"] == "notification" assert active_response["notification"]["id"] == 777 response = await comm_background.receive_json_from(timeout=5) assert response["type"] == "notification" assert response["notification"]["id"] == 777 assert response["notification"]["data"]["chat_id"] == test_chat.id try: await comm_background.disconnect() except asyncio.CancelledError: pass try: await comm_active.disconnect() except asyncio.CancelledError: pass @pytest.mark.asyncio async def test_active_chat_notification_stays_unread_until_http_ack( self, ws_communicator, user, test_chat, ): """Receiving an event in an active chat must not persist a read state.""" comm = await ws_communicator(user=user) connected, _ = await comm.connect() assert connected await comm.send_json_to( { "action": "open_chat", "chat_id": test_chat.id, "load_history": False, } ) response = await comm.receive_json_from(timeout=5) assert response["type"] == "chat_opened" notification = await database_sync_to_async(Notification.objects.create)( recipient=user, verb="chat_new_message", description="Suppressed notification", action_url=f"/messages/{test_chat.id}", data={"chat_id": test_chat.id, "message_id": 777}, ) channel_layer = get_channel_layer() payload = { "type": "notification", "id": notification.id, "verb": "chat_new_message", "description": "Suppressed notification", "action_url": f"/messages/{test_chat.id}", "timestamp": "2026-04-02T12:00:00Z", "unread": True, "data": { "chat_id": test_chat.id, "message_id": 777, }, } await channel_layer.group_send( f"user_{user.id}", { "type": "notification_message", "message": payload, }, ) notification_event = await comm.receive_json_from(timeout=5) assert notification_event["type"] == "notification" assert notification_event["notification"]["id"] == notification.id notification = await database_sync_to_async(Notification.objects.get)( id=notification.id ) assert notification.unread is True await comm.disconnect() @pytest.mark.asyncio async def test_notification_read_sync_between_sessions(self, ws_communicator, user): comm1 = await ws_communicator(user=user) comm2 = await ws_communicator(user=user) connected1, _ = await comm1.connect() connected2, _ = await comm2.connect() assert connected1 and connected2 channel_layer = get_channel_layer() await channel_layer.group_send( f"user_{user.id}", { "type": "notification_read", "notification_id": 321, "unread_count": 4, }, ) response1 = await comm1.receive_json_from(timeout=5) response2 = await comm2.receive_json_from(timeout=5) assert response1["type"] == "notification_read" assert response1["notification_id"] == 321 assert response1["unread_count"] == 4 assert response2["type"] == "notification_read" await comm1.disconnect() await comm2.disconnect() @pytest.mark.asyncio async def test_notifications_read_all_sync_between_sessions( self, ws_communicator, user ): comm1 = await ws_communicator(user=user) comm2 = await ws_communicator(user=user) connected1, _ = await comm1.connect() connected2, _ = await comm2.connect() assert connected1 and connected2 channel_layer = get_channel_layer() await channel_layer.group_send( f"user_{user.id}", { "type": "notifications_read_all", "notification_ids": [11, 12, 13], "category": "Заявки", "unread_count": 0, }, ) response1 = await comm1.receive_json_from(timeout=5) response2 = await comm2.receive_json_from(timeout=5) assert response1["type"] == "notifications_read_all" assert response1["notification_ids"] == [11, 12, 13] assert response1["category"] == "Заявки" assert response1["unread_count"] == 0 assert response2["type"] == "notifications_read_all" await comm1.disconnect() await comm2.disconnect()