88 lines
3.2 KiB
Python
88 lines
3.2 KiB
Python
import pytest
|
|
from asgiref.sync import async_to_sync
|
|
from channels.layers import get_channel_layer
|
|
from channels.routing import URLRouter
|
|
from channels.testing import WebsocketCommunicator
|
|
from django.urls import path
|
|
|
|
from accounts.models import User
|
|
from contest.consumers import MatchConsumer
|
|
from contest.models import Contest, Question, RealtimeMatch
|
|
|
|
|
|
@pytest.mark.django_db(transaction=True)
|
|
def test_match_consumer_双方连接并同步答题进度():
|
|
first = User.objects.create_user(
|
|
username="socket_player_one",
|
|
password="StrongPass_2026",
|
|
nickname="WS 玩家一",
|
|
)
|
|
second = User.objects.create_user(
|
|
username="socket_player_two",
|
|
password="StrongPass_2026",
|
|
nickname="WS 玩家二",
|
|
)
|
|
outsider = User.objects.create_user(
|
|
username="socket_outsider",
|
|
password="StrongPass_2026",
|
|
nickname="WS 局外人",
|
|
)
|
|
contest = Contest.objects.create(
|
|
slug="socket-contest",
|
|
title="WebSocket 联机赛",
|
|
kind=Contest.Kind.REALTIME,
|
|
track=Question.Track.STANDARD,
|
|
status=Contest.Status.PUBLISHED,
|
|
)
|
|
match = RealtimeMatch.objects.create(
|
|
contest=contest,
|
|
player_one=first,
|
|
player_two=second,
|
|
player_one_rating=first.rating,
|
|
player_two_rating=second.rating,
|
|
status=RealtimeMatch.Status.ACTIVE,
|
|
)
|
|
application = URLRouter(
|
|
[
|
|
path(
|
|
"ws/test/<uuid:match_id>/",
|
|
MatchConsumer.as_asgi(),
|
|
)
|
|
]
|
|
)
|
|
|
|
async def scenario():
|
|
outsider_socket = WebsocketCommunicator(application, f"/ws/test/{match.id}/")
|
|
outsider_socket.scope["user"] = outsider
|
|
outsider_connected, close_code = await outsider_socket.connect()
|
|
assert not outsider_connected
|
|
assert close_code == 4403
|
|
|
|
first_socket = WebsocketCommunicator(application, f"/ws/test/{match.id}/")
|
|
second_socket = WebsocketCommunicator(application, f"/ws/test/{match.id}/")
|
|
first_socket.scope["user"] = first
|
|
second_socket.scope["user"] = second
|
|
first_connected, _ = await first_socket.connect()
|
|
second_connected, _ = await second_socket.connect()
|
|
assert first_connected and second_connected
|
|
assert (await first_socket.receive_json_from())["type"] == "connected"
|
|
assert (await second_socket.receive_json_from())["type"] == "connected"
|
|
|
|
await first_socket.send_json_to({"type": "progress", "answered_count": 3})
|
|
first_progress = await first_socket.receive_json_from()
|
|
second_progress = await second_socket.receive_json_from()
|
|
assert first_progress["answered_count"] == 3
|
|
assert second_progress["answered_count"] == 3
|
|
assert second_progress["user_id"] == str(first.id)
|
|
|
|
await get_channel_layer().group_send(
|
|
f"match_{match.id}",
|
|
{"type": "match.state", "reason": "completed"},
|
|
)
|
|
assert (await first_socket.receive_json_from())["reason"] == "completed"
|
|
assert (await second_socket.receive_json_from())["reason"] == "completed"
|
|
await first_socket.disconnect()
|
|
await second_socket.disconnect()
|
|
|
|
async_to_sync(scenario)()
|