daphne/channels/tests/test_management.py

from __future__ import unicode_literals

import logging

from asgiref.inmemory import ChannelLayer
from django.core.management import CommandError, call_command
from django.test import TestCase, mock
from six import StringIO

from channels.management.commands import runserver
from channels.staticfiles import StaticFilesConsumer


class FakeChannelLayer(ChannelLayer):
    '''
    Dummy class to bypass the 'inmemory' string check.
    '''
    pass


@mock.patch('channels.management.commands.runworker.Worker')
class RunWorkerTests(TestCase):

    def setUp(self):
        import channels.log
        self.stream = StringIO()
        channels.log.handler = logging.StreamHandler(self.stream)

    def test_runworker_no_local_only(self, mock_worker):
        """
        Runworker should fail with the default "inmemory" worker.
        """
        with self.assertRaises(CommandError):
            call_command('runworker')

    def test_debug(self, mock_worker):
        """
        Test that the StaticFilesConsumer is used in debug mode.
        """
        with self.settings(DEBUG=True, STATIC_URL='/static/'):
            # Use 'fake_channel' that bypasses the 'inmemory' check
            call_command('runworker', '--layer', 'fake_channel')
            mock_worker.assert_called_with(
                only_channels=None,
                exclude_channels=None,
                callback=None,
                channel_layer=mock.ANY,
            )

            channel_layer = mock_worker.call_args[1]['channel_layer']
            static_consumer = channel_layer.router.root.routing[0].consumer
            self.assertIsInstance(static_consumer, StaticFilesConsumer)

    def test_runworker(self, mock_worker):
        # Use 'fake_channel' that bypasses the 'inmemory' check
        call_command('runworker', '--layer', 'fake_channel')
        mock_worker.assert_called_with(
            callback=None,
            only_channels=None,
            channel_layer=mock.ANY,
            exclude_channels=None,
        )

    def test_runworker_verbose(self, mocked_worker):
        # Use 'fake_channel' that bypasses the 'inmemory' check
        call_command('runworker', '--layer', 'fake_channel', '--verbosity', '2')

        # Verify the callback is set
        mocked_worker.assert_called_with(
            callback=mock.ANY,
            only_channels=None,
            channel_layer=mock.ANY,
            exclude_channels=None,
        )


class RunServerTests(TestCase):

    def setUp(self):
        import channels.log
        self.stream = StringIO()
        # Capture the logging of the channels moduel to match against the
        # output.
        channels.log.handler = logging.StreamHandler(self.stream)

    @mock.patch('channels.management.commands.runserver.sys.stdout', new_callable=StringIO)
    @mock.patch('channels.management.commands.runserver.Server')
    @mock.patch('channels.management.commands.runworker.Worker')
    def test_runserver_basic(self, mocked_worker, mocked_server, mock_stdout):
        # Django's autoreload util uses threads and this is not needed
        # in the test envirionment.
        # See:
        # https://github.com/django/django/blob/master/django/core/management/commands/runserver.py#L105
        call_command('runserver', '--noreload')
        mocked_server.assert_called_with(
            port=8000,
            signal_handlers=True,
            http_timeout=60,
            host='127.0.0.1',
            action_logger=mock.ANY,
            channel_layer=mock.ANY,
            ws_protocols=None,
            root_path='',
        )

    @mock.patch('channels.management.commands.runserver.sys.stdout', new_callable=StringIO)
    @mock.patch('channels.management.commands.runserver.Server')
    @mock.patch('channels.management.commands.runworker.Worker')
    def test_runserver_debug(self, mocked_worker, mocked_server, mock_stdout):
        """
        Test that the server runs with `DEBUG=True`.
        """
        # Debug requires the static url is set.
        with self.settings(DEBUG=True, STATIC_URL='/static/'):
            call_command('runserver', '--noreload')
            mocked_server.assert_called_with(
                port=8000,
                signal_handlers=True,
                http_timeout=60,
                host='127.0.0.1',
                action_logger=mock.ANY,
                channel_layer=mock.ANY,
                ws_protocols=None,
                root_path='',
            )

            call_command('runserver', '--noreload', 'localhost:8001')
            mocked_server.assert_called_with(
                port=8001,
                signal_handlers=True,
                http_timeout=60,
                host='localhost',
                action_logger=mock.ANY,
                channel_layer=mock.ANY,
                ws_protocols=None,
                root_path='',
            )

        self.assertFalse(
            mocked_worker.called,
            "The worker should not be called with '--noworker'",
        )

    @mock.patch('channels.management.commands.runserver.sys.stdout', new_callable=StringIO)
    @mock.patch('channels.management.commands.runserver.Server')
    @mock.patch('channels.management.commands.runworker.Worker')
    def test_runserver_noworker(self, mocked_worker, mocked_server, mock_stdout):
        '''
        Test that the Worker is not called when using the `--noworker` parameter.
        '''
        call_command('runserver', '--noreload', '--noworker')
        mocked_server.assert_called_with(
            port=8000,
            signal_handlers=True,
            http_timeout=60,
            host='127.0.0.1',
            action_logger=mock.ANY,
            channel_layer=mock.ANY,
            ws_protocols=None,
            root_path='',
        )
        self.assertFalse(
            mocked_worker.called,
            "The worker should not be called with '--noworker'",
        )

    @mock.patch('channels.management.commands.runserver.sys.stderr', new_callable=StringIO)
    def test_log_action(self, mocked_stderr):
        cmd = runserver.Command()
        test_actions = [
            (100, 'http', 'complete', 'HTTP GET /a-path/ 100 [0.12, a-client]'),
            (200, 'http', 'complete', 'HTTP GET /a-path/ 200 [0.12, a-client]'),
            (300, 'http', 'complete', 'HTTP GET /a-path/ 300 [0.12, a-client]'),
            (304, 'http', 'complete', 'HTTP GET /a-path/ 304 [0.12, a-client]'),
            (400, 'http', 'complete', 'HTTP GET /a-path/ 400 [0.12, a-client]'),
            (404, 'http', 'complete', 'HTTP GET /a-path/ 404 [0.12, a-client]'),
            (500, 'http', 'complete', 'HTTP GET /a-path/ 500 [0.12, a-client]'),
            (None, 'websocket', 'connected', 'WebSocket CONNECT /a-path/ [a-client]'),
            (None, 'websocket', 'disconnected', 'WebSocket DISCONNECT /a-path/ [a-client]'),
            (None, 'websocket', 'something', ''),  # This shouldn't happen
        ]

        for status_code, protocol, action, output in test_actions:
            details = {
                'status': status_code,
                'method': 'GET',
                'path': '/a-path/',
                'time_taken': 0.12345,
                'client': 'a-client',
            }
            cmd.log_action(protocol, action, details)
            self.assertIn(output, mocked_stderr.getvalue())
            # Clear previous output
            mocked_stderr.truncate(0)