1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
|
import asyncio
from urllib.parse import unquote
import pytest
from django.urls import path
from channels.consumer import AsyncConsumer
from channels.generic.websocket import WebsocketConsumer
from channels.routing import URLRouter
from channels.testing import HttpCommunicator, WebsocketCommunicator
class SimpleHttpApp(AsyncConsumer):
"""
Barebones HTTP ASGI app for testing.
"""
async def http_request(self, event):
assert self.scope["path"] == "/test/"
assert self.scope["method"] == "GET"
assert self.scope["query_string"] == b"foo=bar"
await self.send({"type": "http.response.start", "status": 200, "headers": []})
await self.send({"type": "http.response.body", "body": b"test response"})
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_http_communicator():
"""
Tests that the HTTP communicator class works at a basic level.
"""
communicator = HttpCommunicator(SimpleHttpApp(), "GET", "/test/?foo=bar")
response = await communicator.get_response()
assert response["body"] == b"test response"
assert response["status"] == 200
class SimpleWebsocketApp(WebsocketConsumer):
"""
Barebones WebSocket ASGI app for testing.
"""
def connect(self):
assert self.scope["path"] == "/testws/"
self.accept()
def receive(self, text_data=None, bytes_data=None):
self.send(text_data=text_data, bytes_data=bytes_data)
class AcceptCloseWebsocketApp(WebsocketConsumer):
def connect(self):
assert self.scope["path"] == "/testws/"
self.accept()
self.close()
class ErrorWebsocketApp(WebsocketConsumer):
"""
Barebones WebSocket ASGI app for error testing.
"""
def receive(self, text_data=None, bytes_data=None):
pass
class KwargsWebSocketApp(WebsocketConsumer):
"""
WebSocket ASGI app used for testing the kwargs arguments in the url_route.
"""
def connect(self):
self.accept()
self.send(text_data=self.scope["url_route"]["kwargs"]["message"])
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_websocket_communicator():
"""
Tests that the WebSocket communicator class works at a basic level.
"""
communicator = WebsocketCommunicator(SimpleWebsocketApp(), "/testws/")
# Test connection
connected, subprotocol = await communicator.connect()
assert connected
assert subprotocol is None
# Test sending text
await communicator.send_to(text_data="hello")
response = await communicator.receive_from()
assert response == "hello"
# Test sending bytes
await communicator.send_to(bytes_data=b"w\0\0\0")
response = await communicator.receive_from()
assert response == b"w\0\0\0"
# Test sending JSON
await communicator.send_json_to({"hello": "world"})
response = await communicator.receive_json_from()
assert response == {"hello": "world"}
# Close out
await communicator.disconnect()
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_websocket_incorrect_read_json():
"""
When using an invalid communicator method, an assertion error will be raised with
informative message.
In this test, the server accepts and then immediately closes the connection so
the server is not in a valid state to handle "receive_from".
"""
communicator = WebsocketCommunicator(AcceptCloseWebsocketApp(), "/testws/")
await communicator.connect()
with pytest.raises(AssertionError) as exception_info:
await communicator.receive_from()
assert (
str(exception_info.value)
== "Expected type 'websocket.send', but was 'websocket.close'"
)
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_websocket_application():
"""
Tests that the WebSocket communicator class works with the
URLRoute application.
"""
application = URLRouter([path("testws/<str:message>/", KwargsWebSocketApp())])
communicator = WebsocketCommunicator(application, "/testws/test/")
connected, subprotocol = await communicator.connect()
# Test connection
assert connected
assert subprotocol is None
message = await communicator.receive_from()
assert message == "test"
await communicator.disconnect()
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_timeout_disconnect():
"""
Tests that communicator.disconnect() raises after a timeout. (Application
is finished.)
"""
communicator = WebsocketCommunicator(ErrorWebsocketApp(), "/testws/")
# Test connection
connected, subprotocol = await communicator.connect()
assert connected
assert subprotocol is None
# Test sending text (will error internally)
await communicator.send_to(text_data="hello")
with pytest.raises(asyncio.TimeoutError):
await communicator.receive_from()
with pytest.raises(asyncio.exceptions.CancelledError):
await communicator.disconnect()
class ConnectionScopeValidator(WebsocketConsumer):
"""
Tests ASGI specification for the connection scope.
"""
def connect(self):
assert self.scope["type"] == "websocket"
# check if path is a unicode string
assert isinstance(self.scope["path"], str)
# check if path has percent escapes decoded
assert self.scope["path"] == unquote(self.scope["path"])
# check if query_string is a bytes sequence
assert isinstance(self.scope["query_string"], bytes)
self.accept()
paths = [
"user:pass@example.com:8080/p/a/t/h?query=string#hash",
"wss://user:pass@example.com:8080/p/a/t/h?query=string#hash",
(
"ws://www.example.com/%E9%A6%96%E9%A1%B5/index.php?"
"foo=%E9%A6%96%E9%A1%B5&spam=eggs"
),
]
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
@pytest.mark.parametrize("path", paths)
async def test_connection_scope(path):
"""
Tests ASGI specification for the the connection scope.
"""
communicator = WebsocketCommunicator(ConnectionScopeValidator(), path)
connected, _ = await communicator.connect()
assert connected
await communicator.disconnect()
|
