mirror of
https://github.com/django/daphne.git
synced 2025-11-01 00:07:26 +03:00
262 lines
11 KiB
Python
262 lines
11 KiB
Python
# coding: utf8
|
|
|
|
import collections
|
|
from urllib import parse
|
|
|
|
from hypothesis import assume, given, settings
|
|
|
|
import http_strategies
|
|
from http_base import DaphneTestCase
|
|
|
|
|
|
class TestHTTPRequest(DaphneTestCase):
|
|
"""
|
|
Tests the HTTP request handling.
|
|
"""
|
|
|
|
def assert_valid_http_scope(
|
|
self,
|
|
scope,
|
|
method,
|
|
path,
|
|
params=None,
|
|
headers=None,
|
|
scheme=None,
|
|
):
|
|
"""
|
|
Checks that the passed scope is a valid ASGI HTTP scope regarding types
|
|
and some urlencoding things.
|
|
"""
|
|
# Check overall keys
|
|
self.assert_key_sets(
|
|
required_keys={"type", "http_version", "method", "path", "query_string", "headers"},
|
|
optional_keys={"scheme", "root_path", "client", "server"},
|
|
actual_keys=scope.keys(),
|
|
)
|
|
# Check that it is the right type
|
|
self.assertEqual(scope["type"], "http")
|
|
# Method (uppercased unicode string)
|
|
self.assertIsInstance(scope["method"], str)
|
|
self.assertEqual(scope["method"], method.upper())
|
|
# Path
|
|
self.assert_valid_path(scope["path"], path)
|
|
# HTTP version
|
|
self.assertIn(scope["http_version"], ["1.0", "1.1", "1.2"])
|
|
# Scheme
|
|
self.assertIn(scope["scheme"], ["http", "https"])
|
|
if scheme:
|
|
self.assertEqual(scheme, scope["scheme"])
|
|
# Query string (byte string and still url encoded)
|
|
query_string = scope["query_string"]
|
|
self.assertIsInstance(query_string, bytes)
|
|
if params:
|
|
self.assertEqual(query_string, parse.urlencode(params or []).encode("ascii"))
|
|
# Ordering of header names is not important, but the order of values for a header
|
|
# name is. To assert whether that order is kept, we transform both the request
|
|
# headers and the channel message headers into a dictionary
|
|
# {name: [value1, value2, ...]} and check if they're equal.
|
|
transformed_scope_headers = collections.defaultdict(list)
|
|
for name, value in scope["headers"]:
|
|
transformed_scope_headers[name].append(value)
|
|
transformed_request_headers = collections.defaultdict(list)
|
|
for name, value in (headers or []):
|
|
expected_name = name.lower().strip().encode("ascii")
|
|
expected_value = value.strip().encode("ascii")
|
|
transformed_request_headers[expected_name].append(expected_value)
|
|
for name, value in transformed_request_headers.items():
|
|
self.assertIn(name, transformed_scope_headers)
|
|
self.assertEqual(value, transformed_scope_headers[name])
|
|
# Root path
|
|
self.assertIsInstance(scope.get("root_path", ""), str)
|
|
# Client and server addresses
|
|
client = scope.get("client")
|
|
if client is not None:
|
|
self.assert_valid_address_and_port(client)
|
|
server = scope.get("server")
|
|
if server is not None:
|
|
self.assert_valid_address_and_port(server)
|
|
|
|
def assert_valid_http_request_message(self, message, body=None):
|
|
"""
|
|
Asserts that a message is a valid http.request message
|
|
"""
|
|
# Check overall keys
|
|
self.assert_key_sets(
|
|
required_keys={"type"},
|
|
optional_keys={"body", "more_body"},
|
|
actual_keys=message.keys(),
|
|
)
|
|
# Check that it is the right type
|
|
self.assertEqual(message["type"], "http.request")
|
|
# If there's a body present, check its type
|
|
self.assertIsInstance(message.get("body", b""), bytes)
|
|
if body is not None:
|
|
self.assertEqual(body, message.get("body", b""))
|
|
|
|
def test_minimal_request(self):
|
|
"""
|
|
Smallest viable example. Mostly verifies that our request building works.
|
|
"""
|
|
scope, messages = self.run_daphne_request("GET", "/")
|
|
self.assert_valid_http_scope(scope, "GET", "/")
|
|
self.assert_valid_http_request_message(messages[0], body=b"")
|
|
|
|
@given(
|
|
request_path=http_strategies.http_path(),
|
|
request_params=http_strategies.query_params()
|
|
)
|
|
@settings(max_examples=5, deadline=5000)
|
|
def test_get_request(self, request_path, request_params):
|
|
"""
|
|
Tests a typical HTTP GET request, with a path and query parameters
|
|
"""
|
|
scope, messages = self.run_daphne_request("GET", request_path, params=request_params)
|
|
self.assert_valid_http_scope(scope, "GET", request_path, params=request_params)
|
|
self.assert_valid_http_request_message(messages[0], body=b"")
|
|
|
|
@given(
|
|
request_path=http_strategies.http_path(),
|
|
request_body=http_strategies.http_body()
|
|
)
|
|
@settings(max_examples=5, deadline=5000)
|
|
def test_post_request(self, request_path, request_body):
|
|
"""
|
|
Tests a typical HTTP POST request, with a path and body.
|
|
"""
|
|
scope, messages = self.run_daphne_request("POST", request_path, body=request_body)
|
|
self.assert_valid_http_scope(scope, "POST", request_path)
|
|
self.assert_valid_http_request_message(messages[0], body=request_body)
|
|
|
|
@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-à/"
|
|
scope, messages = self.run_daphne_request("OPTIONS", request_path, headers=request_headers)
|
|
self.assert_valid_http_scope(scope, "OPTIONS", request_path, headers=request_headers)
|
|
self.assert_valid_http_request_message(messages[0], body=b"")
|
|
|
|
@given(request_headers=http_strategies.headers())
|
|
@settings(max_examples=5, deadline=5000)
|
|
def test_duplicate_headers(self, request_headers):
|
|
"""
|
|
Tests that duplicate header values are preserved
|
|
"""
|
|
# Make sure there's duplicate headers
|
|
assume(len(request_headers) >= 2)
|
|
header_name = request_headers[0][0]
|
|
duplicated_headers = [(header_name, header[1]) for header in request_headers]
|
|
# Run the request
|
|
request_path = "/te st-à/"
|
|
scope, messages = self.run_daphne_request("OPTIONS", request_path, headers=duplicated_headers)
|
|
self.assert_valid_http_scope(scope, "OPTIONS", request_path, headers=duplicated_headers)
|
|
self.assert_valid_http_request_message(messages[0], body=b"")
|
|
|
|
@given(
|
|
request_method=http_strategies.http_method(),
|
|
request_path=http_strategies.http_path(),
|
|
request_params=http_strategies.query_params(),
|
|
request_headers=http_strategies.headers(),
|
|
request_body=http_strategies.http_body(),
|
|
)
|
|
@settings(max_examples=5, deadline=5000)
|
|
def test_kitchen_sink(
|
|
self,
|
|
request_method,
|
|
request_path,
|
|
request_params,
|
|
request_headers,
|
|
request_body,
|
|
):
|
|
"""
|
|
Throw everything at Daphne that we dare. The idea is that if a combination
|
|
of method/path/headers/body would break the spec, hypothesis will eventually find it.
|
|
"""
|
|
scope, messages = self.run_daphne_request(
|
|
request_method,
|
|
request_path,
|
|
params=request_params,
|
|
headers=request_headers,
|
|
body=request_body,
|
|
)
|
|
self.assert_valid_http_scope(
|
|
scope,
|
|
request_method,
|
|
request_path,
|
|
params=request_params,
|
|
headers=request_headers,
|
|
)
|
|
self.assert_valid_http_request_message(messages[0], body=request_body)
|
|
|
|
def test_headers_are_lowercased_and_stripped(self):
|
|
"""
|
|
Make sure headers are normalized as the spec says they are.
|
|
"""
|
|
headers = [("MYCUSTOMHEADER", " foobar ")]
|
|
scope, messages = self.run_daphne_request("GET", "/", headers=headers)
|
|
self.assert_valid_http_scope(scope, "GET", "/", headers=headers)
|
|
self.assert_valid_http_request_message(messages[0], body=b"")
|
|
# Note that Daphne returns a list of tuples here, which is fine, because the spec
|
|
# asks to treat them interchangeably.
|
|
assert [list(x) for x in scope["headers"]] == [[b"mycustomheader", b"foobar"]]
|
|
|
|
@given(daphne_path=http_strategies.http_path())
|
|
@settings(max_examples=5, deadline=5000)
|
|
def test_root_path_header(self, daphne_path):
|
|
"""
|
|
Tests root_path handling.
|
|
"""
|
|
# Daphne-Root-Path must be URL encoded when submitting as HTTP header field
|
|
headers = [("Daphne-Root-Path", parse.quote(daphne_path.encode("utf8")))]
|
|
scope, messages = self.run_daphne_request("GET", "/", headers=headers)
|
|
# Daphne-Root-Path is not included in the returned 'headers' section. So we expect
|
|
# empty headers.
|
|
self.assert_valid_http_scope(scope, "GET", "/", headers=[])
|
|
self.assert_valid_http_request_message(messages[0], body=b"")
|
|
# And what we're looking for, root_path being set.
|
|
assert scope["root_path"] == daphne_path
|
|
|
|
def test_x_forwarded_for_ignored(self):
|
|
"""
|
|
Make sure that, by default, X-Forwarded-For is ignored.
|
|
"""
|
|
headers = [
|
|
["X-Forwarded-For", "10.1.2.3"],
|
|
["X-Forwarded-Port", "80"],
|
|
]
|
|
scope, messages = self.run_daphne_request("GET", "/", headers=headers)
|
|
self.assert_valid_http_scope(scope, "GET", "/", headers=headers)
|
|
self.assert_valid_http_request_message(messages[0], body=b"")
|
|
# It should NOT appear in the client scope item
|
|
self.assertNotEqual(scope["client"], ["10.1.2.3", 80])
|
|
|
|
def test_x_forwarded_for_parsed(self):
|
|
"""
|
|
When X-Forwarded-For is enabled, make sure it is respected.
|
|
"""
|
|
headers = [
|
|
["X-Forwarded-For", "10.1.2.3"],
|
|
["X-Forwarded-Port", "80"],
|
|
]
|
|
scope, messages = self.run_daphne_request("GET", "/", headers=headers, xff=True)
|
|
self.assert_valid_http_scope(scope, "GET", "/", headers=headers)
|
|
self.assert_valid_http_request_message(messages[0], body=b"")
|
|
# It should now appear in the client scope item
|
|
self.assertEqual(scope["client"], ["10.1.2.3", 80])
|
|
|
|
def test_x_forwarded_for_no_port(self):
|
|
"""
|
|
When X-Forwarded-For is enabled but only the host is passed, make sure
|
|
that at least makes it through.
|
|
"""
|
|
headers = [
|
|
["X-Forwarded-For", "10.1.2.3"],
|
|
]
|
|
scope, messages = self.run_daphne_request("GET", "/", headers=headers, xff=True)
|
|
self.assert_valid_http_scope(scope, "GET", "/", headers=headers)
|
|
self.assert_valid_http_request_message(messages[0], body=b"")
|
|
# It should now appear in the client scope item
|
|
self.assertEqual(scope["client"], ["10.1.2.3", 0])
|