Full test suite for HTTP requests (#91)

* Add Hypothesis for property-based tests

Hypothesis:
"It works by letting you write tests that assert that
something should be true for every case, not just the ones you happen to
think of."

I think it's well suited for the task of ensuring Daphne conforms to the
ASGI specification.

* Fix accidental cast to byte string under Python 2

While grepping for calls to str(), I found this bit which looks like a
cast to unicode was intended under Python 2.

* ASGITestCase - checking channel messages for spec conformance

This commit introduces a new test case class, with it's main method
assert_valid_http_request_message. The idea is
that this method is a translation of the ASGI spec to code, and can be
used to check channel messages for conformance with that part of the
spec.

I plan to add further methods for other parts of the spec.

* Add Hypothesis strategies for generating HTTP requests

Hypothesis, our framework for test data generation, contains only
general so-called strategies for generating data. This commit adds a few
which will be useful for generating the data for our tests.

Alos see http://hypothesis.readthedocs.io/en/latest/data.html.

* Add and convert tests for HTTP requests

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.

* Fix byte string issue in Python 2

Twisted seems to return a byte string for the client and server IP
address. It is easily rectified by casting to the required unicode
string. Also added a test to ensure this is also handled correctly in
the X-Forwarded-For header parsing.

* Check order of header values

I'm in the process of updating the ASGI spec to require that the order
of header values is kept. To match that work, I'm adding matching
assertions to the tests.

The code unfortunately is not as elegant as I'd like, but then it's only
a result of the underlying HTTP spec.

* Suppress warning about slow test

The kitchen sink test expectedly can be slow. So far it wasn't slow
enough for hypothesis to trigger a warning, but today Travis must've had
a bad day. I think for this test is is acceptable to exceed hypothesis'
warning threshold.
This commit is contained in:
Maik Hoepfel 2017-03-14 16:12:07 -05:00 committed by Andrew Godwin
parent c55bc8a94b
commit 7f92a48293
12 changed files with 549 additions and 74 deletions

2
.gitignore vendored
View File

@ -4,3 +4,5 @@ __pycache__
dist/
build/
/.tox
.hypothesis
.cache

View File

@ -11,6 +11,6 @@ env:
- TWISTED_RELEASE="twisted"
- TWISTED_RELEASE="twisted==16.0.0"
install: pip install $TWISTED_RELEASE -e .
install: pip install $TWISTED_RELEASE -e .[tests]
script: python -m unittest discover

View File

@ -63,8 +63,10 @@ class WebRequest(http.Request):
upgrade_header = self.requestHeaders.getRawHeaders(b"Upgrade")[0]
# Get client address if possible
if hasattr(self.client, "host") and hasattr(self.client, "port"):
self.client_addr = [self.client.host, self.client.port]
self.server_addr = [self.host.host, self.host.port]
# client.host and host.host are byte strings in Python 2, but spec
# requires unicode string.
self.client_addr = [six.text_type(self.client.host), self.client.port]
self.server_addr = [six.text_type(self.host.host), self.host.port]
else:
self.client_addr = None
self.server_addr = None
@ -273,7 +275,7 @@ class WebRequest(http.Request):
(b"Content-Type", b"text/html; charset=utf-8"),
],
"content": (self.error_template % {
"title": str(status) + " " + status_text.decode("ascii"),
"title": six.text_type(status) + " " + status_text.decode("ascii"),
"body": body,
}).encode("utf8"),
})

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))

View File

@ -0,0 +1,112 @@
"""
Assorted Hypothesis strategies useful for generating HTTP requests and responses
"""
from __future__ import unicode_literals
from six.moves.urllib import parse
import string
from hypothesis import strategies
HTTP_METHODS = ['OPTIONS', 'GET', 'HEAD', 'POST', 'PUT', 'DELETE', 'TRACE', 'CONNECT']
# Unicode characters of the "Letter" category
letters = strategies.characters(whitelist_categories=('Lu', 'Ll', 'Lt', 'Lm', 'Lo', 'Nl'))
def http_method():
return strategies.sampled_from(HTTP_METHODS)
def http_path():
"""
Returns a URL path (not encoded).
"""
alphabet = string.ascii_letters + string.digits + '-._~/'
return strategies.text(min_size=0, max_size=255, alphabet=alphabet).map(lambda s: '/' + s)
def http_body():
"""
Returns random printable ASCII characters. This may be exceeding what HTTP allows,
but seems to not cause an issue so far.
"""
return strategies.text(alphabet=string.printable, min_size=0, average_size=600, max_size=1500)
def valid_bidi(value):
"""
Rejects strings which nonsensical Unicode text direction flags.
Relying on random Unicode characters means that some combinations don't make sense, from a
direction of text point of view. This little helper just rejects those.
"""
try:
value.encode('idna')
except UnicodeError:
return False
else:
return True
def _domain_label():
return strategies.text(
alphabet=letters, min_size=1, average_size=6, max_size=63).filter(valid_bidi)
def international_domain_name():
"""
Returns a byte string of a domain name, IDNA-encoded.
"""
return strategies.lists(
_domain_label(), min_size=2, average_size=2).map(lambda s: ('.'.join(s)).encode('idna'))
def _query_param():
return strategies.text(alphabet=letters, min_size=1, average_size=10, max_size=255).\
map(lambda s: s.encode('utf8'))
def query_params():
"""
Returns a list of two-tuples byte strings, ready for encoding with urlencode.
We're aiming for a total length of a URL below 2083 characters, so this strategy
ensures that the total urlencoded query string is not longer than 1500 characters.
"""
return strategies.lists(
strategies.tuples(_query_param(), _query_param()), min_size=0, average_size=5).\
filter(lambda x: len(parse.urlencode(x)) < 1500)
def header_name():
"""
Strategy returning something that looks like a HTTP header field
https://en.wikipedia.org/wiki/List_of_HTTP_header_fields suggests they are between 4
and 20 characters long
"""
return strategies.text(
alphabet=string.ascii_letters + string.digits + '-', min_size=1, max_size=30)
def header_value():
"""
Strategy returning something that looks like a HTTP header value
"For example, the Apache 2.3 server by default limits the size of each field to 8190 bytes"
https://en.wikipedia.org/wiki/List_of_HTTP_header_fields
"""
return strategies.text(
alphabet=string.ascii_letters + string.digits + string.punctuation + ' /t',
min_size=1, average_size=40, max_size=8190).filter(lambda s: len(s.encode('utf8')) < 8190)
def headers():
"""
Strategy returning a list of tuples, containing HTTP header fields and their values.
"[Apache 2.3] there can be at most 100 header fields in a single request."
https://en.wikipedia.org/wiki/List_of_HTTP_header_fields
"""
return strategies.lists(
strategies.tuples(header_name(), header_value()),
min_size=0, average_size=10, max_size=100)

View File

@ -0,0 +1,197 @@
# 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, settings, HealthCheck
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(),
)
# This test is slow enough that on Travis, hypothesis sometimes complains.
@settings(suppress_health_check=[HealthCheck.too_slow])
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])

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])

View File

@ -1,6 +1,7 @@
# coding: utf8
from __future__ import unicode_literals
from unittest import TestCase
import six
from twisted.web.http_headers import Headers
@ -17,10 +18,9 @@ class TestXForwardedForHttpParsing(TestCase):
b'X-Forwarded-For': [b'10.1.2.3'],
b'X-Forwarded-Port': [b'1234']
})
self.assertEqual(
parse_x_forwarded_for(headers),
['10.1.2.3', 1234]
)
result = parse_x_forwarded_for(headers)
self.assertEqual(result, ['10.1.2.3', 1234])
self.assertIsInstance(result[0], six.text_type)
def test_address_only(self):
headers = Headers({

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):

124
daphne/tests/testcases.py Normal file
View File

@ -0,0 +1,124 @@
"""
Contains a test case class to allow verifying ASGI messages
"""
from __future__ import unicode_literals
from collections import defaultdict
import six
import socket
from six.moves.urllib import parse
import unittest
class ASGITestCase(unittest.TestCase):
"""
Test case with helpers for ASGI message verification
"""
def assert_is_ip_address(self, address):
"""
Tests whether a given address string is a valid IPv4 or IPv6 address.
"""
try:
socket.inet_aton(address)
except socket.error:
self.fail("'%s' is not a valid IP address." % address)
def assert_valid_http_request_message(
self, channel_message, request_method, request_path,
request_params=None, request_headers=None, request_body=None):
"""
Asserts that a given channel message conforms to the HTTP request section of the ASGI spec.
"""
self.assertTrue(channel_message)
# == General assertions about expected dictionary keys being present ==
message_keys = set(channel_message.keys())
required_message_keys = {
'reply_channel', 'http_version', 'method', 'path', 'query_string', 'headers',
}
optional_message_keys = {
'scheme', 'root_path', 'body', 'body_channel', 'client', 'server'
}
self.assertTrue(required_message_keys <= message_keys)
# Assert that no other keys are present
self.assertEqual(set(), message_keys - required_message_keys - optional_message_keys)
# == Assertions about required channel_message fields ==
reply_channel = channel_message['reply_channel']
self.assertIsInstance(reply_channel, six.text_type)
self.assertTrue(reply_channel.startswith('http.response!'))
http_version = channel_message['http_version']
self.assertIsInstance(http_version, six.text_type)
self.assertIn(http_version, ['1.0', '1.1', '1.2'])
method = channel_message['method']
self.assertIsInstance(method, six.text_type)
self.assertTrue(method.isupper())
self.assertEqual(channel_message['method'], request_method)
path = channel_message['path']
self.assertIsInstance(path, six.text_type)
self.assertEqual(path, request_path)
# Assert that it's already url decoded
self.assertEqual(path, parse.unquote(path))
query_string = channel_message['query_string']
# Assert that query_string is a byte string and still url encoded
self.assertIsInstance(query_string, six.binary_type)
self.assertEqual(query_string, parse.urlencode(request_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_message_headers = defaultdict(list)
for name, value in channel_message['headers']:
transformed_message_headers[name].append(value)
transformed_request_headers = defaultdict(list)
for name, value in (request_headers or []):
expected_name = name.lower().strip().encode('ascii')
expected_value = value.strip().encode('ascii')
transformed_request_headers[expected_name].append(expected_value)
self.assertEqual(transformed_message_headers, transformed_request_headers)
# == Assertions about optional channel_message fields ==
scheme = channel_message.get('scheme')
if scheme is not None:
self.assertIsInstance(scheme, six.text_type)
self.assertTrue(scheme) # May not be empty
root_path = channel_message.get('root_path')
if root_path is not None:
self.assertIsInstance(root_path, six.text_type)
body = channel_message.get('body')
# Ensure we test for presence of 'body' if a request body was given
if request_body is not None or body is not None:
self.assertIsInstance(body, six.binary_type)
self.assertEqual(body, (request_body or '').encode('ascii'))
body_channel = channel_message.get('body_channel')
if body_channel is not None:
self.assertIsInstance(body_channel, six.text_type)
self.assertIn('?', body_channel)
client = channel_message.get('client')
if client is not None:
client_host, client_port = client
self.assertIsInstance(client_host, six.text_type)
self.assert_is_ip_address(client_host)
self.assertIsInstance(client_port, int)
server = channel_message.get('server')
if server is not None:
server_host, server_port = channel_message['server']
self.assertIsInstance(server_host, six.text_type)
self.assert_is_ip_address(server_host)
self.assertIsInstance(server_port, int)

View File

@ -27,6 +27,9 @@ setup(
'twisted>=16.0',
'autobahn>=0.12',
],
extras_require={
'tests': ['hypothesis', 'tox']
},
entry_points={'console_scripts': [
'daphne = daphne.cli:CommandLineInterface.entrypoint',
]},

View File

@ -3,6 +3,7 @@
envlist = py{27,34,35}-twisted-{old,new}
[testenv]
extras = tests
deps =
twisted-old: twisted==16.0.0
commands = python -m unittest discover