From 333f4644d1a5a167c85eca967a52e4e10a7db0bd Mon Sep 17 00:00:00 2001 From: Simon Willison Date: Wed, 3 Jul 2019 11:22:03 -0700 Subject: [PATCH] Added support for raw_path in scope. (#268) 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 --- daphne/http_protocol.py | 1 + daphne/ws_protocol.py | 1 + tests/http_base.py | 7 +------ tests/test_http_request.py | 15 ++++++++++++--- tests/test_websocket.py | 19 ++++++++++++++++--- 5 files changed, 31 insertions(+), 12 deletions(-) diff --git a/daphne/http_protocol.py b/daphne/http_protocol.py index 0116db4..ccbfdb9 100755 --- a/daphne/http_protocol.py +++ b/daphne/http_protocol.py @@ -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, diff --git a/daphne/ws_protocol.py b/daphne/ws_protocol.py index edf1254..1962450 100755 --- a/daphne/ws_protocol.py +++ b/daphne/ws_protocol.py @@ -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, diff --git a/tests/http_base.py b/tests/http_base.py index e6fc92b..71d9618 100644 --- a/tests/http_base.py +++ b/tests/http_base.py @@ -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)) diff --git a/tests/test_http_request.py b/tests/test_http_request.py index c1efd64..0bd2d82 100644 --- a/tests/test_http_request.py +++ b/tests/test_http_request.py @@ -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 ) diff --git a/tests/test_websocket.py b/tests/test_websocket.py index 69a54f8..9ec2c0d 100644 --- a/tests/test_websocket.py +++ b/tests/test_websocket.py @@ -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.