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. """ abort_start = False 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")))