django/daphne
1
1
Fork 0
mirror of https://github.com/django/daphne.git synced 2025-10-30 23:37:25 +03:00
Code Issues Packages Projects Releases Wiki Activity

Added support for raw_path in scope. (#268)

Browse Source 
As per https://github.com/django/asgiref/pull/92

Required valid URI path fragments to be used in tests:
- Test case must ensure paths are correctly quoted before calling
   run_daphne_request() & co.

Co-authored-by: Carlton Gibson <carlton.gibson@noumenal.es>
This commit is contained in:
Simon Willison 2019-07-03 11:22:03 -07:00 committed by Carlton Gibson
parent ffd949f2ce
commit 333f4644d1
5 changed files with 31 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
daphne/http_protocol.py
View File

@ -171,6 +171,7 @@ class WebRequest(http.Request):
),
"method": self.method.decode("ascii"),
"path": unquote(self.path.decode("ascii")),
"raw_path": self.path,
"root_path": self.root_path,
"scheme": self.client_scheme,
"query_string": self.query_string,

1
daphne/ws_protocol.py
View File

@ -75,6 +75,7 @@ class WebSocketProtocol(WebSocketServerProtocol):
{
"type": "websocket",
"path": unquote(self.path.decode("ascii")),
"raw_path": self.path,
"headers": self.clean_headers,
"query_string": self._raw_query_string, # Passed by HTTP protocol
"client": self.client_addr,

7
tests/http_base.py
View File

@ -32,8 +32,6 @@ class DaphneTestCase(unittest.TestCase):
# Send it the request. We have to do this the long way to allow
# duplicate headers.
conn = HTTPConnection(test_app.host, test_app.port, timeout=timeout)
# Make sure path is urlquoted and add any params
path = parse.quote(path)
if params:
path += "?" + parse.urlencode(params, doseq=True)
conn.putrequest(method, path, skip_accept_encoding=True, skip_host=True)
@ -128,8 +126,6 @@ class DaphneTestCase(unittest.TestCase):
# Send it the request. We have to do this the long way to allow
# duplicate headers.
conn = HTTPConnection(test_app.host, test_app.port, timeout=timeout)
# Make sure path is urlquoted and add any params
path = parse.quote(path)
if params:
path += "?" + parse.urlencode(params, doseq=True)
conn.putrequest("GET", path, skip_accept_encoding=True, skip_host=True)
@ -247,12 +243,11 @@ class DaphneTestCase(unittest.TestCase):
# Assert that no other keys are present
self.assertEqual(set(), present_keys - required_keys - optional_keys)
def assert_valid_path(self, path, request_path):
def assert_valid_path(self, path):
"""
Checks the path is valid and already url-decoded.
"""
self.assertIsInstance(path, str)
self.assertEqual(path, request_path)
# Assert that it's already url decoded
self.assertEqual(path, parse.unquote(path))

15
tests/test_http_request.py
View File

@ -28,6 +28,7 @@ class TestHTTPRequest(DaphneTestCase):
"http_version",
"method",
"path",
"raw_path",
"query_string",
"headers",
},
@ -40,7 +41,7 @@ class TestHTTPRequest(DaphneTestCase):
self.assertIsInstance(scope["method"], str)
self.assertEqual(scope["method"], method.upper())
# Path
self.assert_valid_path(scope["path"], path)
self.assert_valid_path(scope["path"])
# HTTP version
self.assertIn(scope["http_version"], ["1.0", "1.1", "1.2"])
# Scheme
@ -134,13 +135,21 @@ class TestHTTPRequest(DaphneTestCase):
self.assert_valid_http_scope(scope, "POST", request_path)
self.assert_valid_http_request_message(messages[0], body=request_body)
def test_raw_path(self):
"""
Tests that /foo%2Fbar produces raw_path and a decoded path
"""
scope, _ = self.run_daphne_request("GET", "/foo%2Fbar")
self.assertEqual(scope["path"], "/foo/bar")
self.assertEqual(scope["raw_path"], b"/foo%2Fbar")
@given(request_headers=http_strategies.headers())
@settings(max_examples=5, deadline=5000)
def test_headers(self, request_headers):
"""
Tests that HTTP header fields are handled as specified
"""
request_path = "/te st-à/"
request_path = parse.quote("/te st-à/")
scope, messages = self.run_daphne_request(
"OPTIONS", request_path, headers=request_headers
)
@ -160,7 +169,7 @@ class TestHTTPRequest(DaphneTestCase):
header_name = request_headers[0][0]
duplicated_headers = [(header_name, header[1]) for header in request_headers]
# Run the request
request_path = "/te st-à/"
request_path = parse.quote("/te st-à/")
scope, messages = self.run_daphne_request(
"OPTIONS", request_path, headers=duplicated_headers
)

19
tests/test_websocket.py
View File

@ -24,14 +24,14 @@ class TestWebsocket(DaphneTestCase):
"""
# Check overall keys
self.assert_key_sets(
required_keys={"type", "path", "query_string", "headers"},
required_keys={"type", "path", "raw_path", "query_string", "headers"},
optional_keys={"scheme", "root_path", "client", "server", "subprotocols"},
actual_keys=scope.keys(),
)
# Check that it is the right type
self.assertEqual(scope["type"], "websocket")
# Path
self.assert_valid_path(scope["path"], path)
self.assert_valid_path(scope["path"])
# Scheme
self.assertIn(scope.get("scheme", "ws"), ["ws", "wss"])
if scheme:
@ -161,7 +161,7 @@ class TestWebsocket(DaphneTestCase):
test_app.add_send_messages([{"type": "websocket.accept"}])
self.websocket_handshake(
test_app,
path=request_path,
path=parse.quote(request_path),
params=request_params,
headers=request_headers,
)
@ -172,6 +172,19 @@ class TestWebsocket(DaphneTestCase):
)
self.assert_valid_websocket_connect_message(messages[0])
def test_raw_path(self):
"""
Tests that /foo%2Fbar produces raw_path and a decoded path
"""
with DaphneTestingInstance() as test_app:
test_app.add_send_messages([{"type": "websocket.accept"}])
self.websocket_handshake(test_app, path="/foo%2Fbar")
# Validate the scope and messages we got
scope, _ = test_app.get_received()
self.assertEqual(scope["path"], "/foo/bar")
self.assertEqual(scope["raw_path"], b"/foo%2Fbar")
def test_text_frames(self):
"""
Tests we can send and receive text frames.