daphne/tests/test_cli.py

268 lines
8.5 KiB
Python
Raw Permalink Normal View History

2017-11-26 00:39:46 +03:00
import logging
import os
from argparse import ArgumentError
from unittest import TestCase, skipUnless
2017-11-26 00:39:46 +03:00
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):
2017-11-26 00:40:15 +03:00
self.assertEqual(build(), [], msg="Empty list returned when no kwargs given")
2017-11-26 00:39:46 +03:00
def testTcpPortBindings(self):
self.assertEqual(
2017-11-26 00:40:15 +03:00
build(port=1234, host="example.com"),
2018-08-27 05:27:32 +03:00
["tcp:port=1234:interface=example.com"],
2017-11-26 00:39:46 +03:00
)
self.assertEqual(
2018-08-27 05:27:32 +03:00
build(port=8000, host="127.0.0.1"), ["tcp:port=8000:interface=127.0.0.1"]
2017-11-26 00:39:46 +03:00
)
self.assertEqual(
2018-08-27 05:27:32 +03:00
build(port=8000, host="[200a::1]"), [r"tcp:port=8000:interface=200a\:\:1"]
2017-11-26 00:39:46 +03:00
)
self.assertEqual(
2018-08-27 05:27:32 +03:00
build(port=8000, host="200a::1"), [r"tcp:port=8000:interface=200a\:\:1"]
2017-11-26 00:39:46 +03:00
)
# incomplete port/host kwargs raise errors
2018-08-27 05:27:32 +03:00
self.assertRaises(ValueError, build, port=123)
self.assertRaises(ValueError, build, host="example.com")
2017-11-26 00:39:46 +03:00
def testUnixSocketBinding(self):
self.assertEqual(
2018-08-27 05:27:32 +03:00
build(unix_socket="/tmp/daphne.sock"), ["unix:/tmp/daphne.sock"]
2017-11-26 00:39:46 +03:00
)
def testFileDescriptorBinding(self):
2018-08-27 05:27:32 +03:00
self.assertEqual(build(file_descriptor=5), ["fd:fileno=5"])
2017-11-26 00:39:46 +03:00
def testMultipleEnpoints(self):
self.assertEqual(
sorted(
build(
file_descriptor=123,
2017-11-26 00:40:15 +03:00
unix_socket="/tmp/daphne.sock",
2017-11-26 00:39:46 +03:00
port=8080,
2018-08-27 05:27:32 +03:00
host="10.0.0.1",
2017-11-26 00:39:46 +03:00
)
),
2018-08-27 05:27:32 +03:00
sorted(
[
"tcp:port=8080:interface=10.0.0.1",
"unix:/tmp/daphne.sock",
"fd:fileno=123",
]
),
2017-11-26 00:39:46 +03:00
)
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()
2018-08-27 05:27:32 +03:00
cli.run(
args + ["daphne:__version__"]
) # We just pass something importable as app
2017-11-26 00:39:46 +03:00
# 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,
2018-08-27 05:27:32 +03:00
"Wrong value for server kwarg %s: %r != %r"
% (key, value, actual_value),
2017-11-26 00:39:46 +03:00
)
def testCLIBasics(self):
"""
Tests basic endpoint generation.
"""
2018-08-27 05:27:32 +03:00
self.assertCLI([], {"endpoints": ["tcp:port=8000:interface=127.0.0.1"]})
2017-11-26 00:39:46 +03:00
self.assertCLI(
2018-08-27 05:27:32 +03:00
["-p", "123"], {"endpoints": ["tcp:port=123:interface=127.0.0.1"]}
2017-11-26 00:39:46 +03:00
)
self.assertCLI(
2018-08-27 05:27:32 +03:00
["-b", "10.0.0.1"], {"endpoints": ["tcp:port=8000:interface=10.0.0.1"]}
2017-11-26 00:39:46 +03:00
)
self.assertCLI(
2018-08-27 05:27:32 +03:00
["-b", "200a::1"], {"endpoints": [r"tcp:port=8000:interface=200a\:\:1"]}
2017-11-26 00:39:46 +03:00
)
self.assertCLI(
2018-08-27 05:27:32 +03:00
["-b", "[200a::1]"], {"endpoints": [r"tcp:port=8000:interface=200a\:\:1"]}
2017-11-26 00:39:46 +03:00
)
self.assertCLI(
2017-11-26 00:40:15 +03:00
["-p", "8080", "-b", "example.com"],
2018-08-27 05:27:32 +03:00
{"endpoints": ["tcp:port=8080:interface=example.com"]},
2017-11-26 00:39:46 +03:00
)
def testUnixSockets(self):
self.assertCLI(
2017-11-26 00:40:15 +03:00
["-p", "8080", "-u", "/tmp/daphne.sock"],
2017-11-26 00:39:46 +03:00
{
2017-11-26 00:40:15 +03:00
"endpoints": [
"tcp:port=8080:interface=127.0.0.1",
"unix:/tmp/daphne.sock",
2018-08-27 05:27:32 +03:00
]
2017-11-26 00:39:46 +03:00
},
)
self.assertCLI(
2017-11-26 00:40:15 +03:00
["-b", "example.com", "-u", "/tmp/daphne.sock"],
2017-11-26 00:39:46 +03:00
{
2017-11-26 00:40:15 +03:00
"endpoints": [
"tcp:port=8000:interface=example.com",
"unix:/tmp/daphne.sock",
2018-08-27 05:27:32 +03:00
]
2017-11-26 00:39:46 +03:00
},
)
self.assertCLI(
2017-11-26 00:40:15 +03:00
["-u", "/tmp/daphne.sock", "--fd", "5"],
2018-08-27 05:27:32 +03:00
{"endpoints": ["fd:fileno=5", "unix:/tmp/daphne.sock"]},
2017-11-26 00:39:46 +03:00
)
def testMixedCLIEndpointCreation(self):
"""
Tests mixing the shortcut options with the endpoint string options.
"""
self.assertCLI(
2017-11-26 00:40:15 +03:00
["-p", "8080", "-e", "unix:/tmp/daphne.sock"],
2017-11-26 00:39:46 +03:00
{
2017-11-26 00:40:15 +03:00
"endpoints": [
"tcp:port=8080:interface=127.0.0.1",
2018-08-27 05:27:32 +03:00
"unix:/tmp/daphne.sock",
]
2017-11-26 00:39:46 +03:00
},
)
self.assertCLI(
2017-11-26 00:40:15 +03:00
["-p", "8080", "-e", "tcp:port=8080:interface=127.0.0.1"],
2017-11-26 00:39:46 +03:00
{
2017-11-26 00:40:15 +03:00
"endpoints": [
"tcp:port=8080:interface=127.0.0.1",
"tcp:port=8080:interface=127.0.0.1",
2018-08-27 05:27:32 +03:00
]
2017-11-26 00:39:46 +03:00
},
)
def testCustomEndpoints(self):
"""
Tests entirely custom endpoints
"""
2018-08-27 05:27:32 +03:00
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")))