mirror of
https://github.com/django/daphne.git
synced 2024-11-14 03:56:34 +03:00
268 lines
8.5 KiB
Python
268 lines
8.5 KiB
Python
import logging
|
|
import os
|
|
from argparse import ArgumentError
|
|
from unittest import TestCase, skipUnless
|
|
|
|
from daphne.cli import CommandLineInterface
|
|
from daphne.endpoints import build_endpoint_description_strings as build
|
|
|
|
|
|
class TestEndpointDescriptions(TestCase):
|
|
"""
|
|
Tests that the endpoint parsing/generation works as intended.
|
|
"""
|
|
|
|
def testBasics(self):
|
|
self.assertEqual(build(), [], msg="Empty list returned when no kwargs given")
|
|
|
|
def testTcpPortBindings(self):
|
|
self.assertEqual(
|
|
build(port=1234, host="example.com"),
|
|
["tcp:port=1234:interface=example.com"],
|
|
)
|
|
|
|
self.assertEqual(
|
|
build(port=8000, host="127.0.0.1"), ["tcp:port=8000:interface=127.0.0.1"]
|
|
)
|
|
|
|
self.assertEqual(
|
|
build(port=8000, host="[200a::1]"), [r"tcp:port=8000:interface=200a\:\:1"]
|
|
)
|
|
|
|
self.assertEqual(
|
|
build(port=8000, host="200a::1"), [r"tcp:port=8000:interface=200a\:\:1"]
|
|
)
|
|
|
|
# incomplete port/host kwargs raise errors
|
|
self.assertRaises(ValueError, build, port=123)
|
|
self.assertRaises(ValueError, build, host="example.com")
|
|
|
|
def testUnixSocketBinding(self):
|
|
self.assertEqual(
|
|
build(unix_socket="/tmp/daphne.sock"), ["unix:/tmp/daphne.sock"]
|
|
)
|
|
|
|
def testFileDescriptorBinding(self):
|
|
self.assertEqual(build(file_descriptor=5), ["fd:fileno=5"])
|
|
|
|
def testMultipleEnpoints(self):
|
|
self.assertEqual(
|
|
sorted(
|
|
build(
|
|
file_descriptor=123,
|
|
unix_socket="/tmp/daphne.sock",
|
|
port=8080,
|
|
host="10.0.0.1",
|
|
)
|
|
),
|
|
sorted(
|
|
[
|
|
"tcp:port=8080:interface=10.0.0.1",
|
|
"unix:/tmp/daphne.sock",
|
|
"fd:fileno=123",
|
|
]
|
|
),
|
|
)
|
|
|
|
|
|
class TestCLIInterface(TestCase):
|
|
"""
|
|
Tests the overall CLI class.
|
|
"""
|
|
|
|
class TestedCLI(CommandLineInterface):
|
|
"""
|
|
CommandLineInterface subclass that we used for testing (has a fake
|
|
server subclass).
|
|
"""
|
|
|
|
class TestedServer:
|
|
"""
|
|
Mock server object for testing.
|
|
"""
|
|
|
|
def __init__(self, **kwargs):
|
|
self.init_kwargs = kwargs
|
|
|
|
def run(self):
|
|
pass
|
|
|
|
server_class = TestedServer
|
|
|
|
def setUp(self):
|
|
logging.disable(logging.CRITICAL)
|
|
|
|
def tearDown(self):
|
|
logging.disable(logging.NOTSET)
|
|
|
|
def assertCLI(self, args, server_kwargs):
|
|
"""
|
|
Asserts that the CLI class passes the right args to the server class.
|
|
Passes in a fake application automatically.
|
|
"""
|
|
cli = self.TestedCLI()
|
|
cli.run(
|
|
args + ["daphne:__version__"]
|
|
) # We just pass something importable as app
|
|
# Check the server got all arguments as intended
|
|
for key, value in server_kwargs.items():
|
|
# Get the value and sort it if it's a list (for endpoint checking)
|
|
actual_value = cli.server.init_kwargs.get(key)
|
|
if isinstance(actual_value, list):
|
|
actual_value.sort()
|
|
# Check values
|
|
self.assertEqual(
|
|
value,
|
|
actual_value,
|
|
"Wrong value for server kwarg %s: %r != %r"
|
|
% (key, value, actual_value),
|
|
)
|
|
|
|
def testCLIBasics(self):
|
|
"""
|
|
Tests basic endpoint generation.
|
|
"""
|
|
self.assertCLI([], {"endpoints": ["tcp:port=8000:interface=127.0.0.1"]})
|
|
self.assertCLI(
|
|
["-p", "123"], {"endpoints": ["tcp:port=123:interface=127.0.0.1"]}
|
|
)
|
|
self.assertCLI(
|
|
["-b", "10.0.0.1"], {"endpoints": ["tcp:port=8000:interface=10.0.0.1"]}
|
|
)
|
|
self.assertCLI(
|
|
["-b", "200a::1"], {"endpoints": [r"tcp:port=8000:interface=200a\:\:1"]}
|
|
)
|
|
self.assertCLI(
|
|
["-b", "[200a::1]"], {"endpoints": [r"tcp:port=8000:interface=200a\:\:1"]}
|
|
)
|
|
self.assertCLI(
|
|
["-p", "8080", "-b", "example.com"],
|
|
{"endpoints": ["tcp:port=8080:interface=example.com"]},
|
|
)
|
|
|
|
def testUnixSockets(self):
|
|
self.assertCLI(
|
|
["-p", "8080", "-u", "/tmp/daphne.sock"],
|
|
{
|
|
"endpoints": [
|
|
"tcp:port=8080:interface=127.0.0.1",
|
|
"unix:/tmp/daphne.sock",
|
|
]
|
|
},
|
|
)
|
|
self.assertCLI(
|
|
["-b", "example.com", "-u", "/tmp/daphne.sock"],
|
|
{
|
|
"endpoints": [
|
|
"tcp:port=8000:interface=example.com",
|
|
"unix:/tmp/daphne.sock",
|
|
]
|
|
},
|
|
)
|
|
self.assertCLI(
|
|
["-u", "/tmp/daphne.sock", "--fd", "5"],
|
|
{"endpoints": ["fd:fileno=5", "unix:/tmp/daphne.sock"]},
|
|
)
|
|
|
|
def testMixedCLIEndpointCreation(self):
|
|
"""
|
|
Tests mixing the shortcut options with the endpoint string options.
|
|
"""
|
|
self.assertCLI(
|
|
["-p", "8080", "-e", "unix:/tmp/daphne.sock"],
|
|
{
|
|
"endpoints": [
|
|
"tcp:port=8080:interface=127.0.0.1",
|
|
"unix:/tmp/daphne.sock",
|
|
]
|
|
},
|
|
)
|
|
self.assertCLI(
|
|
["-p", "8080", "-e", "tcp:port=8080:interface=127.0.0.1"],
|
|
{
|
|
"endpoints": [
|
|
"tcp:port=8080:interface=127.0.0.1",
|
|
"tcp:port=8080:interface=127.0.0.1",
|
|
]
|
|
},
|
|
)
|
|
|
|
def testCustomEndpoints(self):
|
|
"""
|
|
Tests entirely custom endpoints
|
|
"""
|
|
self.assertCLI(["-e", "imap:"], {"endpoints": ["imap:"]})
|
|
|
|
def test_default_proxyheaders(self):
|
|
"""
|
|
Passing `--proxy-headers` without a parameter will use the
|
|
`X-Forwarded-For` header.
|
|
"""
|
|
self.assertCLI(
|
|
["--proxy-headers"], {"proxy_forwarded_address_header": "X-Forwarded-For"}
|
|
)
|
|
|
|
def test_custom_proxyhost(self):
|
|
"""
|
|
Passing `--proxy-headers-host` will set the used host header to
|
|
the passed one, and `--proxy-headers` is mandatory.
|
|
"""
|
|
self.assertCLI(
|
|
["--proxy-headers", "--proxy-headers-host", "blah"],
|
|
{"proxy_forwarded_address_header": "blah"},
|
|
)
|
|
with self.assertRaises(expected_exception=ArgumentError) as exc:
|
|
self.assertCLI(
|
|
["--proxy-headers-host", "blah"],
|
|
{"proxy_forwarded_address_header": "blah"},
|
|
)
|
|
self.assertEqual(exc.exception.argument_name, "--proxy-headers-host")
|
|
self.assertEqual(
|
|
exc.exception.message,
|
|
"--proxy-headers has to be passed for this parameter.",
|
|
)
|
|
|
|
def test_custom_proxyport(self):
|
|
"""
|
|
Passing `--proxy-headers-port` will set the used port header to
|
|
the passed one, and `--proxy-headers` is mandatory.
|
|
"""
|
|
self.assertCLI(
|
|
["--proxy-headers", "--proxy-headers-port", "blah2"],
|
|
{"proxy_forwarded_port_header": "blah2"},
|
|
)
|
|
with self.assertRaises(expected_exception=ArgumentError) as exc:
|
|
self.assertCLI(
|
|
["--proxy-headers-port", "blah2"],
|
|
{"proxy_forwarded_address_header": "blah2"},
|
|
)
|
|
self.assertEqual(exc.exception.argument_name, "--proxy-headers-port")
|
|
self.assertEqual(
|
|
exc.exception.message,
|
|
"--proxy-headers has to be passed for this parameter.",
|
|
)
|
|
|
|
def test_custom_servername(self):
|
|
"""
|
|
Passing `--server-name` will set the default server header
|
|
from 'daphne' to the passed one.
|
|
"""
|
|
self.assertCLI([], {"server_name": "daphne"})
|
|
self.assertCLI(["--server-name", ""], {"server_name": ""})
|
|
self.assertCLI(["--server-name", "python"], {"server_name": "python"})
|
|
|
|
def test_no_servername(self):
|
|
"""
|
|
Passing `--no-server-name` will set server name to '' (empty string)
|
|
"""
|
|
self.assertCLI(["--no-server-name"], {"server_name": ""})
|
|
|
|
|
|
@skipUnless(os.getenv("ASGI_THREADS"), "ASGI_THREADS environment variable not set.")
|
|
class TestASGIThreads(TestCase):
|
|
def test_default_executor(self):
|
|
from daphne.server import twisted_loop
|
|
|
|
executor = twisted_loop._default_executor
|
|
self.assertEqual(executor._max_workers, int(os.getenv("ASGI_THREADS")))
|