django/daphne
1
1
Fork 0
mirror of https://github.com/django/daphne.git synced 2025-04-21 01:02:06 +03:00
Code Issues Packages Projects Releases Wiki Activity

Add and convert tests for HTTP requests

Browse Source 
This commit introduces a few Hypothesis tests to test the HTTP request
part of the specification. To keep things organized, I split the
existing tests module into two: one concerned with requests, and one
concerned with responses. I anticipate that we'll also add modules for
chunks and server push later.

daphne already had tests for the HTTP protocol. Some of them I converted
to Hypothesis tests to increase what was tested. Some were also
concerned with HTTP responses, so they were moved to the new response
module. And three tests were concerned with proxy behaviour, which I
wasn't sure about, and I just kept them as-is, but also moved them
to the request tests.
This commit is contained in:
Maik Hoepfel 2017-03-07 12:27:38 +08:00
parent bcd7015844
commit 549df9d3ec
4 changed files with 295 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

92
daphne/tests/factories.py Normal file
View File

@ -0,0 +1,92 @@
from __future__ import unicode_literals
import six
from six.moves.urllib import parse
from asgiref.inmemory import ChannelLayer
from twisted.test import proto_helpers
from daphne.http_protocol import HTTPFactory
def message_for_request(method, path, params=None, headers=None, body=None):
"""
Constructs a HTTP request according to the given parameters, runs
that through daphne and returns the emitted channel message.
"""
request = _build_request(method, path, params, headers, body)
return _run_through_daphne(request, 'http.request')
def _build_request(method, path, params=None, headers=None, body=None):
"""
Takes request parameters and returns a byte string of a valid HTTP/1.1 request.
We really shouldn't manually build a HTTP request, and instead try to capture
what e.g. urllib or requests would do. But that is non-trivial, so meanwhile
we hope that our request building doesn't mask any errors.
This code is messy, because urllib behaves rather different between Python 2
and 3. Readability is further obstructed by the fact that Python 3.4 doesn't
support % formatting for bytes, so we need to concat everything.
If we run into more issues with this, the python-future library has a backport
of Python 3's urllib.
:param method: ASCII string of HTTP method.
:param path: unicode string of URL path.
:param params: List of two-tuples of bytestrings, ready for consumption for
urlencode. Encode to utf8 if necessary.
:param headers: List of two-tuples ASCII strings of HTTP header, value.
:param body: ASCII string of request body.
ASCII string is short for a unicode string containing only ASCII characters,
or a byte string with ASCII encoding.
"""
if headers is None:
headers = []
else:
headers = headers[:]
if six.PY3:
quoted_path = parse.quote(path)
if params:
quoted_path += '?' + parse.urlencode(params)
quoted_path = quoted_path.encode('ascii')
else:
quoted_path = parse.quote(path.encode('utf8'))
if params:
quoted_path += b'?' + parse.urlencode(params)
request = method.encode('ascii') + b' ' + quoted_path + b" HTTP/1.1\r\n"
for k, v in headers:
request += k.encode('ascii') + b': ' + v.encode('ascii') + b"\r\n"
request += b'\r\n'
if body:
request += body.encode('ascii')
return request
def _run_through_daphne(request, channel_name):
"""
Returns Daphne's channel message for a given request.
This helper requires a fair bit of scaffolding and can certainly be improved,
but it works for now.
"""
channel_layer = ChannelLayer()
factory = HTTPFactory(channel_layer)
proto = factory.buildProtocol(('127.0.0.1', 0))
tr = proto_helpers.StringTransport()
proto.makeConnection(tr)
proto.dataReceived(request)
_, message = channel_layer.receive([channel_name])
return message
def content_length_header(body):
"""
Returns an appropriate Content-Length HTTP header for a given body.
"""
return 'Content-Length', six.text_type(len(body))

195
daphne/tests/test_http_request.py Normal file
View File

@ -0,0 +1,195 @@
# coding: utf8
"""
Tests for the HTTP request section of the ASGI spec
"""
from __future__ import unicode_literals
import unittest
from six.moves.urllib import parse
from asgiref.inmemory import ChannelLayer
from hypothesis import given, assume
from twisted.test import proto_helpers
from daphne.http_protocol import HTTPFactory
from daphne.tests import testcases, http_strategies
from daphne.tests.factories import message_for_request, content_length_header
class TestHTTPRequestSpec(testcases.ASGITestCase):
"""
Tests which try to pour the HTTP request section of the ASGI spec into code.
The heavy lifting is done by the assert_valid_http_request_message function,
the tests mostly serve to wire up hypothesis so that it exercise it's power to find
edge cases.
"""
def test_minimal_request(self):
"""
Smallest viable example. Mostly verifies that our request building works.
"""
request_method, request_path = 'GET', '/'
message = message_for_request(request_method, request_path)
self.assert_valid_http_request_message(message, request_method, request_path)
@given(
request_path=http_strategies.http_path(),
request_params=http_strategies.query_params()
)
def test_get_request(self, request_path, request_params):
"""
Tests a typical HTTP GET request, with a path and query parameters
"""
request_method = 'GET'
message = message_for_request(request_method, request_path, request_params)
self.assert_valid_http_request_message(
message, request_method, request_path, request_params=request_params)
@given(
request_path=http_strategies.http_path(),
request_body=http_strategies.http_body()
)
def test_post_request(self, request_path, request_body):
"""
Tests a typical POST request, submitting some data in a body.
"""
request_method = 'POST'
headers = [content_length_header(request_body)]
message = message_for_request(
request_method, request_path, headers=headers, body=request_body)
self.assert_valid_http_request_message(
message, request_method, request_path,
request_headers=headers, request_body=request_body)
@given(request_headers=http_strategies.headers())
def test_headers(self, request_headers):
"""
Tests that HTTP header fields are handled as specified
"""
request_method, request_path = 'OPTIONS', '/te st-à/'
message = message_for_request(request_method, request_path, headers=request_headers)
self.assert_valid_http_request_message(
message, request_method, request_path, request_headers=request_headers)
@given(request_headers=http_strategies.headers())
def test_duplicate_headers(self, request_headers):
"""
Tests that duplicate header values are preserved
"""
assume(len(request_headers) >= 2)
# Set all header field names to the same value
header_name = request_headers[0][0]
duplicated_headers = [(header_name, header[1]) for header in request_headers]
request_method, request_path = 'OPTIONS', '/te st-à/'
message = message_for_request(request_method, request_path, headers=duplicated_headers)
self.assert_valid_http_request_message(
message, request_method, request_path, request_headers=duplicated_headers)
@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(),
)
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.
"""
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(
message, request_method, request_path, request_params, request_headers, request_body)
def test_headers_are_lowercased_and_stripped(self):
request_method, request_path = 'GET', '/'
headers = [('MYCUSTOMHEADER', ' foobar ')]
message = message_for_request(request_method, request_path, headers=headers)
self.assert_valid_http_request_message(
message, request_method, request_path, request_headers=headers)
# Note that Daphne returns a list of tuples here, which is fine, because the spec
# asks to treat them interchangeably.
assert message['headers'] == [(b'mycustomheader', b'foobar')]
@given(daphne_path=http_strategies.http_path())
def test_root_path_header(self, daphne_path):
"""
Tests root_path handling.
"""
request_method, request_path = 'GET', '/'
# Daphne-Root-Path must be URL encoded when submitting as HTTP header field
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)
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])

70
daphne/tests/test_http.py → daphne/tests/test_http_response.py
View File

@ -1,13 +1,18 @@
# coding: utf8
"""
Tests for the HTTP response section of the ASGI spec
"""
from __future__ import unicode_literals
from unittest import TestCase
from asgiref.inmemory import ChannelLayer
from twisted.test import proto_helpers
from ..http_protocol import HTTPFactory
class TestHTTPProtocol(TestCase):
class TestHTTPResponse(TestCase):
"""
Tests that the HTTP protocol class correctly generates and parses messages.
"""
@ -52,21 +57,6 @@ class TestHTTPProtocol(TestCase):
# Make sure that comes back right on the protocol
self.assertEqual(self.tr.value(), b"HTTP/1.1 201 Created\r\nTransfer-Encoding: chunked\r\nX-Test: Boom!\r\n\r\n6\r\nOH HAI\r\n0\r\n\r\n")
def test_root_path_header(self):
"""
Tests root path header handling
"""
# Send a simple request to the protocol
self.proto.dataReceived(
b"GET /te%20st-%C3%A0/?foo=bar HTTP/1.1\r\n" +
b"Host: somewhere.com\r\n" +
b"Daphne-Root-Path: /foobar%20/bar\r\n" +
b"\r\n"
)
# Get the resulting message off of the channel layer, check root_path
_, message = self.channel_layer.receive(["http.request"])
self.assertEqual(message['root_path'], "/foobar /bar")
def test_http_disconnect_sets_path_key(self):
"""
Tests http disconnect has the path key set, see https://channels.readthedocs.io/en/latest/asgi.html#disconnect
@ -93,51 +83,3 @@ class TestHTTPProtocol(TestCase):
# Get the disconnection notification
_, disconnect_message = self.channel_layer.receive(["http.disconnect"])
self.assertEqual(disconnect_message['path'], "/te st-à/")
def test_x_forwarded_for_ignored(self):
"""
Tests basic HTTP parsing
"""
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):
"""
Tests basic HTTP parsing
"""
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):
"""
Tests basic HTTP parsing
"""
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])

4
daphne/tests/test_ws.py
View File

@ -4,12 +4,12 @@ from unittest import TestCase
from asgiref.inmemory import ChannelLayer
from twisted.test import proto_helpers
from ..http_protocol import HTTPFactory
from daphne.http_protocol import HTTPFactory
class TestWebSocketProtocol(TestCase):
"""
Tests that the WebSocket protocol class correctly generates and parses messages.
Tests that the WebSocket protocol class correcly generates and parses messages.
"""
def setUp(self):