79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
from datetime import timedelta
|
|
|
|
import pytest
|
|
from asgiref.sync import async_to_sync
|
|
from channels.routing import URLRouter
|
|
from channels.testing import WebsocketCommunicator
|
|
from django.urls import path
|
|
from django.utils import timezone
|
|
|
|
from accounts.models import User
|
|
from toolbox.consumers import BoardConsumer
|
|
from toolbox.models import BoardSession
|
|
|
|
|
|
@pytest.mark.django_db(transaction=True)
|
|
def test_board_consumer_参与者同步画布并拒绝局外人():
|
|
host = User.objects.create_user(
|
|
username="board_socket_host",
|
|
password="StrongPass_2026",
|
|
nickname="WS 房主",
|
|
)
|
|
guest = User.objects.create_user(
|
|
username="board_socket_guest",
|
|
password="StrongPass_2026",
|
|
nickname="WS 访客",
|
|
)
|
|
outsider = User.objects.create_user(
|
|
username="board_socket_outsider",
|
|
password="StrongPass_2026",
|
|
nickname="WS 局外人",
|
|
)
|
|
session = BoardSession.objects.create(
|
|
code="ABC234",
|
|
host=host,
|
|
guest=guest,
|
|
status=BoardSession.Status.ACTIVE,
|
|
expires_at=timezone.now() + timedelta(minutes=30),
|
|
)
|
|
application = URLRouter(
|
|
[
|
|
path(
|
|
"ws/test/<uuid:session_id>/",
|
|
BoardConsumer.as_asgi(),
|
|
)
|
|
]
|
|
)
|
|
|
|
async def scenario():
|
|
outsider_socket = WebsocketCommunicator(
|
|
application,
|
|
f"/ws/test/{session.id}/",
|
|
)
|
|
outsider_socket.scope["user"] = outsider
|
|
connected, close_code = await outsider_socket.connect()
|
|
assert not connected
|
|
assert close_code == 4403
|
|
|
|
host_socket = WebsocketCommunicator(application, f"/ws/test/{session.id}/")
|
|
guest_socket = WebsocketCommunicator(application, f"/ws/test/{session.id}/")
|
|
host_socket.scope["user"] = host
|
|
guest_socket.scope["user"] = guest
|
|
assert (await host_socket.connect())[0]
|
|
assert (await guest_socket.connect())[0]
|
|
assert (await host_socket.receive_json_from())["type"] == "connected"
|
|
assert (await guest_socket.receive_json_from())["type"] == "connected"
|
|
|
|
payload = "data:image/png;base64,AAAA"
|
|
await host_socket.send_json_to({"type": "canvas", "payload": payload})
|
|
host_event = await host_socket.receive_json_from()
|
|
guest_event = await guest_socket.receive_json_from()
|
|
assert host_event["payload"] == payload
|
|
assert guest_event["payload"] == payload
|
|
assert guest_event["user_id"] == str(host.id)
|
|
|
|
await host_socket.disconnect()
|
|
await guest_socket.disconnect()
|
|
|
|
async_to_sync(scenario)()
|