django/daphne
1
1
Fork 0
mirror of https://github.com/django/daphne.git synced 2024-11-21 23:46:33 +03:00
Code Issues Packages Projects Releases Wiki Activity

Full HTTP request test suite

Browse Source
This commit is contained in:
Andrew Godwin 2017-11-25 23:19:27 -08:00
parent b3115e8dcf
commit e0e60e4117
2 changed files with 145 additions and 125 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
tests/http_base.py
View File

@ -1,5 +1,5 @@
from urllib import parse from urllib import parse
import requests from http.client import HTTPConnection
import socket import socket
import subprocess import subprocess
import time import time
@ -34,7 +34,7 @@ class DaphneTestCase(unittest.TestCase):
finally: finally:
s.close() s.close()
def run_daphne(self, method, path, params, data, responses, headers=None, timeout=1): def run_daphne(self, method, path, params, body, responses, headers=None, timeout=1, xff=False):
""" """
Runs Daphne with the given request callback (given the base URL) Runs Daphne with the given request callback (given the base URL)
and response messages. and response messages.
@ -52,7 +52,11 @@ class DaphneTestCase(unittest.TestCase):
else: else:
raise RuntimeError("Cannot find a free port to test on") raise RuntimeError("Cannot find a free port to test on")
# Launch daphne on that port # Launch daphne on that port
process = subprocess.Popen(["daphne", "-p", str(port), "daphne.test_utils:TestApplication"]) daphne_args = ["daphne", "-p", str(port), "-v", "0"]
if xff:
# Optionally enable X-Forwarded-For support.
daphne_args += ["--proxy-headers"]
process = subprocess.Popen(daphne_args + ["daphne.test_utils:TestApplication"])
try: try:
for _ in range(100): for _ in range(100):
time.sleep(0.1) time.sleep(0.1)
@ -60,9 +64,25 @@ class DaphneTestCase(unittest.TestCase):
break break
else: else:
raise RuntimeError("Daphne never came up.") raise RuntimeError("Daphne never came up.")
# Send it the request # Send it the request. We have to do this the long way to allow
url = "http://127.0.0.1:%i%s" % (port, path) # duplicate headers.
response = requests.request(method, url, params=params, data=data, headers=headers, timeout=timeout) conn = HTTPConnection("127.0.0.1", 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)
# Manually send over headers (encoding any non-safe values as best we can)
if headers:
for header_name, header_value in headers:
conn.putheader(header_name.encode("utf8"), header_value.encode("utf8"))
# Send body if provided.
if body:
conn.putheader("Content-Length", str(len(body)))
conn.endheaders(message_body=body)
else:
conn.endheaders()
response = conn.getresponse()
finally: finally:
# Shut down daphne # Shut down daphne
process.terminate() process.terminate()
@ -71,19 +91,18 @@ class DaphneTestCase(unittest.TestCase):
# Return the inner result and the response # Return the inner result and the response
return inner_result, response return inner_result, response
def run_daphne_request(self, method, path, params=None, data=None, headers=None): def run_daphne_request(self, method, path, params=None, body=None, headers=None, xff=False):
""" """
Convenience method for just testing request handling. Convenience method for just testing request handling.
Returns (scope, messages) Returns (scope, messages)
""" """
if headers is not None:
headers = dict(headers)
inner_result, _ = self.run_daphne( inner_result, _ = self.run_daphne(
method=method, method=method,
path=path, path=path,
params=params, params=params,
data=data, body=body,
headers=headers, headers=headers,
xff=xff,
responses=[{"type": "http.response", "status": 200, "content": b"OK"}], responses=[{"type": "http.response", "status": 200, "content": b"OK"}],
) )
return inner_result["scope"], inner_result["messages"] return inner_result["scope"], inner_result["messages"]

231
tests/test_http_request.py
View File

@ -130,7 +130,7 @@ class TestHTTPRequestSpec(DaphneTestCase):
""" """
Tests a typical HTTP POST request, with a path and body. Tests a typical HTTP POST request, with a path and body.
""" """
scope, messages = self.run_daphne_request("POST", request_path, data=request_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_scope(scope, "POST", request_path)
self.assert_valid_http_request_message(messages[0], body=request_body) self.assert_valid_http_request_message(messages[0], body=request_body)
@ -145,123 +145,124 @@ class TestHTTPRequestSpec(DaphneTestCase):
self.assert_valid_http_scope(scope, "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"") self.assert_valid_http_request_message(messages[0], body=b"")
# @given(request_headers=http_strategies.headers()) @given(request_headers=http_strategies.headers())
# def test_duplicate_headers(self, request_headers): @settings(max_examples=5, deadline=2000)
# """ def test_duplicate_headers(self, request_headers):
# Tests that duplicate header values are preserved """
# """ Tests that duplicate header values are preserved
# assume(len(request_headers) >= 2) """
# # Set all header field names to the same value # Make sure there's duplicate headers
# header_name = request_headers[0][0] assume(len(request_headers) >= 2)
# duplicated_headers = [(header_name, header[1]) for header in request_headers] 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"")
# request_method, request_path = "OPTIONS", "/te st-à/" @given(
# message = message_for_request(request_method, request_path, headers=duplicated_headers) 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=2000)
def test_kitchen_sink(
self,
request_method,
request_path,
request_params,
request_headers,
request_body,
):
"""
Throw everything at channels 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)
# self.assert_valid_http_request_message( def test_headers_are_lowercased_and_stripped(self):
# message, request_method, request_path, request_headers=duplicated_headers) """
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 scope["headers"] == [[b"mycustomheader", b"foobar"]]
# @given( @given(daphne_path=http_strategies.http_path())
# request_method=http_strategies.http_method(), @settings(max_examples=5, deadline=2000)
# request_path=http_strategies.http_path(), def test_root_path_header(self, daphne_path):
# request_params=http_strategies.query_params(), """
# request_headers=http_strategies.headers(), Tests root_path handling.
# request_body=http_strategies.http_body(), """
# ) # Daphne-Root-Path must be URL encoded when submitting as HTTP header field
# # This test is slow enough that on Travis, hypothesis sometimes complains. headers = [("Daphne-Root-Path", parse.quote(daphne_path.encode("utf8")))]
# @settings(suppress_health_check=[HealthCheck.too_slow]) scope, messages = self.run_daphne_request("GET", "/", headers=headers)
# def test_kitchen_sink( # Daphne-Root-Path is not included in the returned 'headers' section. So we expect
# self, request_method, request_path, request_params, request_headers, request_body): # empty headers.
# """ self.assert_valid_http_scope(scope, "GET", "/", headers=[])
# Throw everything at channels that we dare. The idea is that if a combination self.assert_valid_http_request_message(messages[0], body=b"")
# of method/path/headers/body would break the spec, hypothesis will eventually find it. # And what we're looking for, root_path being set.
# """ assert scope["root_path"] == daphne_path
# request_headers.append(content_length_header(request_body))
# message = message_for_request(
# request_method, request_path, request_params, request_headers, request_body)
# self.assert_valid_http_request_message( def test_x_forwarded_for_ignored(self):
# message, request_method, request_path, request_params, request_headers, request_body) """
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_headers_are_lowercased_and_stripped(self): def test_x_forwarded_for_parsed(self):
# request_method, request_path = "GET", "/" """
# headers = [("MYCUSTOMHEADER", " foobar ")] When X-Forwarded-For is enabled, make sure it is respected.
# message = message_for_request(request_method, request_path, headers=headers) """
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])
# self.assert_valid_http_request_message( def test_x_forwarded_for_no_port(self):
# message, request_method, request_path, request_headers=headers) """
# # Note that Daphne returns a list of tuples here, which is fine, because the spec When X-Forwarded-For is enabled but only the host is passed, make sure
# # asks to treat them interchangeably. that at least makes it through.
# assert message["headers"] == [(b"mycustomheader", b"foobar")] """
headers = [
# @given(daphne_path=http_strategies.http_path()) ["X-Forwarded-For", "10.1.2.3"],
# def test_root_path_header(self, daphne_path): ]
# """ scope, messages = self.run_daphne_request("GET", "/", headers=headers, xff=True)
# Tests root_path handling. self.assert_valid_http_scope(scope, "GET", "/", headers=headers)
# """ self.assert_valid_http_request_message(messages[0], body=b"")
# request_method, request_path = "GET", "/" # It should now appear in the client scope item
# # Daphne-Root-Path must be URL encoded when submitting as HTTP header field self.assertEqual(scope["client"], ["10.1.2.3", 0])
# headers = [("Daphne-Root-Path", parse.quote(daphne_path.encode("utf8")))]
# message = message_for_request(request_method, request_path, headers=headers)
# # Daphne-Root-Path is not included in the returned 'headers' section. So we expect
# # empty headers.
# expected_headers = []
# self.assert_valid_http_request_message(
# message, request_method, request_path, request_headers=expected_headers)
# # And what we're looking for, root_path being set.
# assert message["root_path"] == daphne_path
# class TestProxyHandling(unittest.TestCase):
# """
# Tests that concern interaction of Daphne with proxies.
# They live in a separate test case, because they're not part of the spec.
# """
# def setUp(self):
# self.channel_layer = ChannelLayer()
# self.factory = HTTPFactory(self.channel_layer, send_channel="test!")
# self.proto = self.factory.buildProtocol(("127.0.0.1", 0))
# self.tr = proto_helpers.StringTransport()
# self.proto.makeConnection(self.tr)
# def test_x_forwarded_for_ignored(self):
# self.proto.dataReceived(
# b"GET /te%20st-%C3%A0/?foo=+bar HTTP/1.1\r\n" +
# b"Host: somewhere.com\r\n" +
# b"X-Forwarded-For: 10.1.2.3\r\n" +
# b"X-Forwarded-Port: 80\r\n" +
# b"\r\n"
# )
# # Get the resulting message off of the channel layer
# _, message = self.channel_layer.receive(["http.request"])
# self.assertEqual(message["client"], ["192.168.1.1", 54321])
# def test_x_forwarded_for_parsed(self):
# self.factory.proxy_forwarded_address_header = "X-Forwarded-For"
# self.factory.proxy_forwarded_port_header = "X-Forwarded-Port"
# self.proto.dataReceived(
# b"GET /te%20st-%C3%A0/?foo=+bar HTTP/1.1\r\n" +
# b"Host: somewhere.com\r\n" +
# b"X-Forwarded-For: 10.1.2.3\r\n" +
# b"X-Forwarded-Port: 80\r\n" +
# b"\r\n"
# )
# # Get the resulting message off of the channel layer
# _, message = self.channel_layer.receive(["http.request"])
# self.assertEqual(message["client"], ["10.1.2.3", 80])
# def test_x_forwarded_for_port_missing(self):
# self.factory.proxy_forwarded_address_header = "X-Forwarded-For"
# self.factory.proxy_forwarded_port_header = "X-Forwarded-Port"
# self.proto.dataReceived(
# b"GET /te%20st-%C3%A0/?foo=+bar HTTP/1.1\r\n" +
# b"Host: somewhere.com\r\n" +
# b"X-Forwarded-For: 10.1.2.3\r\n" +
# b"\r\n"
# )
# # Get the resulting message off of the channel layer
# _, message = self.channel_layer.receive(["http.request"])
# self.assertEqual(message["client"], ["10.1.2.3", 0])