import gzip
import importlib
import json
import logging
import sys
import time
import unittest
import zlib

import six

if six.PY3:
    from unittest import mock
else:
    import mock

from engineio import exceptions
from engineio import packet
from engineio import payload
from engineio import server
import pytest


original_import_module = importlib.import_module


def _mock_import(module, *args, **kwargs):
    if module.startswith('engineio.'):
        return original_import_module(module, *args, **kwargs)
    return module


class TestServer(unittest.TestCase):
    _mock_async = mock.MagicMock()
    _mock_async._async = {
        'thread': 't',
        'queue': 'q',
        'queue_empty': RuntimeError,
        'websocket': 'w',
    }

    def _get_mock_socket(self):
        mock_socket = mock.MagicMock()
        mock_socket.closed = False
        mock_socket.closing = False
        mock_socket.upgraded = False
        mock_socket.session = {}
        return mock_socket

    @classmethod
    def setUpClass(cls):
        server.Server._default_monitor_clients = False

    @classmethod
    def tearDownClass(cls):
        server.Server._default_monitor_clients = True

    def setUp(self):
        logging.getLogger('engineio').setLevel(logging.NOTSET)

    def tearDown(self):
        # restore JSON encoder, in case a test changed it
        packet.Packet.json = json

    def test_is_asyncio_based(self):
        s = server.Server()
        assert not s.is_asyncio_based()

    def test_async_modes(self):
        s = server.Server()
        assert s.async_modes() == [
            'eventlet',
            'gevent_uwsgi',
            'gevent',
            'threading',
        ]

    def test_create(self):
        kwargs = {
            'ping_timeout': 1,
            'ping_interval': 2,
            'max_http_buffer_size': 3,
            'allow_upgrades': False,
            'http_compression': False,
            'compression_threshold': 4,
            'cookie': 'foo',
            'cors_allowed_origins': ['foo', 'bar', 'baz'],
            'cors_credentials': False,
            'async_handlers': False,
        }
        s = server.Server(**kwargs)
        for arg in six.iterkeys(kwargs):
            assert getattr(s, arg) == kwargs[arg]
        assert s.ping_interval_grace_period == 5

    def test_create_with_grace_period(self):
        s = server.Server(ping_interval=(1, 2))
        assert s.ping_interval == 1
        assert s.ping_interval_grace_period == 2

    def test_create_ignores_kwargs(self):
        server.Server(foo='bar')  # this should not raise

    def test_async_mode_threading(self):
        s = server.Server(async_mode='threading')
        assert s.async_mode == 'threading'

        import threading

        try:
            import queue
        except ImportError:
            import Queue as queue

        assert s._async['thread'] == threading.Thread
        assert s._async['queue'] == queue.Queue
        assert s._async['websocket'] is None

    def test_async_mode_eventlet(self):
        s = server.Server(async_mode='eventlet')
        assert s.async_mode == 'eventlet'

        from eventlet.green import threading
        from eventlet import queue
        from engineio.async_drivers import eventlet as async_eventlet

        assert s._async['thread'] == threading.Thread
        assert s._async['queue'] == queue.Queue
        assert s._async['websocket'] == async_eventlet.WebSocketWSGI

    @mock.patch('importlib.import_module', side_effect=_mock_import)
    def test_async_mode_gevent_uwsgi(self, import_module):
        sys.modules['gevent'] = mock.MagicMock()
        sys.modules['gevent'].queue = mock.MagicMock()
        sys.modules['gevent.queue'] = sys.modules['gevent'].queue
        sys.modules['gevent.queue'].JoinableQueue = 'foo'
        sys.modules['gevent.queue'].Empty = RuntimeError
        sys.modules['gevent.event'] = mock.MagicMock()
        sys.modules['gevent.event'].Event = 'bar'
        sys.modules['uwsgi'] = mock.MagicMock()
        s = server.Server(async_mode='gevent_uwsgi')
        assert s.async_mode == 'gevent_uwsgi'

        from engineio.async_drivers import gevent_uwsgi as async_gevent_uwsgi

        assert s._async['thread'] == async_gevent_uwsgi.Thread
        assert s._async['queue'] == 'foo'
        assert s._async['queue_empty'] == RuntimeError
        assert s._async['event'] == 'bar'
        assert s._async['websocket'] == async_gevent_uwsgi.uWSGIWebSocket
        del sys.modules['gevent']
        del sys.modules['gevent.queue']
        del sys.modules['gevent.event']
        del sys.modules['uwsgi']
        del sys.modules['engineio.async_drivers.gevent_uwsgi']

    @mock.patch('importlib.import_module', side_effect=_mock_import)
    def test_async_mode_gevent_uwsgi_without_uwsgi(self, import_module):
        sys.modules['gevent'] = mock.MagicMock()
        sys.modules['gevent'].queue = mock.MagicMock()
        sys.modules['gevent.queue'] = sys.modules['gevent'].queue
        sys.modules['gevent.queue'].JoinableQueue = 'foo'
        sys.modules['gevent.queue'].Empty = RuntimeError
        sys.modules['gevent.event'] = mock.MagicMock()
        sys.modules['gevent.event'].Event = 'bar'
        sys.modules['uwsgi'] = None
        with pytest.raises(ValueError):
            server.Server(async_mode='gevent_uwsgi')
        del sys.modules['gevent']
        del sys.modules['gevent.queue']
        del sys.modules['gevent.event']
        del sys.modules['uwsgi']

    @mock.patch('importlib.import_module', side_effect=_mock_import)
    def test_async_mode_gevent_uwsgi_without_websocket(self, import_module):
        sys.modules['gevent'] = mock.MagicMock()
        sys.modules['gevent'].queue = mock.MagicMock()
        sys.modules['gevent.queue'] = sys.modules['gevent'].queue
        sys.modules['gevent.queue'].JoinableQueue = 'foo'
        sys.modules['gevent.queue'].Empty = RuntimeError
        sys.modules['gevent.event'] = mock.MagicMock()
        sys.modules['gevent.event'].Event = 'bar'
        sys.modules['uwsgi'] = mock.MagicMock()
        del sys.modules['uwsgi'].websocket_handshake
        s = server.Server(async_mode='gevent_uwsgi')
        assert s.async_mode == 'gevent_uwsgi'

        from engineio.async_drivers import gevent_uwsgi as async_gevent_uwsgi

        assert s._async['thread'] == async_gevent_uwsgi.Thread
        assert s._async['queue'] == 'foo'
        assert s._async['queue_empty'] == RuntimeError
        assert s._async['event'] == 'bar'
        assert s._async['websocket'] is None
        del sys.modules['gevent']
        del sys.modules['gevent.queue']
        del sys.modules['gevent.event']
        del sys.modules['uwsgi']
        del sys.modules['engineio.async_drivers.gevent_uwsgi']

    @mock.patch('importlib.import_module', side_effect=_mock_import)
    def test_async_mode_gevent(self, import_module):
        sys.modules['gevent'] = mock.MagicMock()
        sys.modules['gevent'].queue = mock.MagicMock()
        sys.modules['gevent.queue'] = sys.modules['gevent'].queue
        sys.modules['gevent.queue'].JoinableQueue = 'foo'
        sys.modules['gevent.queue'].Empty = RuntimeError
        sys.modules['gevent.event'] = mock.MagicMock()
        sys.modules['gevent.event'].Event = 'bar'
        sys.modules['geventwebsocket'] = 'geventwebsocket'
        s = server.Server(async_mode='gevent')
        assert s.async_mode == 'gevent'

        from engineio.async_drivers import gevent as async_gevent

        assert s._async['thread'] == async_gevent.Thread
        assert s._async['queue'] == 'foo'
        assert s._async['queue_empty'] == RuntimeError
        assert s._async['event'] == 'bar'
        assert s._async['websocket'] == async_gevent.WebSocketWSGI
        del sys.modules['gevent']
        del sys.modules['gevent.queue']
        del sys.modules['gevent.event']
        del sys.modules['geventwebsocket']
        del sys.modules['engineio.async_drivers.gevent']

    @mock.patch('importlib.import_module', side_effect=_mock_import)
    def test_async_mode_gevent_without_websocket(self, import_module):
        sys.modules['gevent'] = mock.MagicMock()
        sys.modules['gevent'].queue = mock.MagicMock()
        sys.modules['gevent.queue'] = sys.modules['gevent'].queue
        sys.modules['gevent.queue'].JoinableQueue = 'foo'
        sys.modules['gevent.queue'].Empty = RuntimeError
        sys.modules['gevent.event'] = mock.MagicMock()
        sys.modules['gevent.event'].Event = 'bar'
        sys.modules['geventwebsocket'] = None
        s = server.Server(async_mode='gevent')
        assert s.async_mode == 'gevent'

        from engineio.async_drivers import gevent as async_gevent

        assert s._async['thread'] == async_gevent.Thread
        assert s._async['queue'] == 'foo'
        assert s._async['queue_empty'] == RuntimeError
        assert s._async['event'] == 'bar'
        assert s._async['websocket'] is None
        del sys.modules['gevent']
        del sys.modules['gevent.queue']
        del sys.modules['gevent.event']
        del sys.modules['geventwebsocket']
        del sys.modules['engineio.async_drivers.gevent']

    @unittest.skipIf(sys.version_info < (3, 5), 'only for Python 3.5+')
    @mock.patch('importlib.import_module', side_effect=_mock_import)
    def test_async_mode_aiohttp(self, import_module):
        sys.modules['aiohttp'] = mock.MagicMock()
        with pytest.raises(ValueError):
            server.Server(async_mode='aiohttp')

    @mock.patch('importlib.import_module', side_effect=[ImportError])
    def test_async_mode_invalid(self, import_module):
        with pytest.raises(ValueError):
            server.Server(async_mode='foo')

    @mock.patch('importlib.import_module', side_effect=[_mock_async])
    def test_async_mode_auto_eventlet(self, import_module):
        s = server.Server()
        assert s.async_mode == 'eventlet'

    @mock.patch(
        'importlib.import_module', side_effect=[ImportError, _mock_async]
    )
    def test_async_mode_auto_gevent_uwsgi(self, import_module):
        s = server.Server()
        assert s.async_mode == 'gevent_uwsgi'

    @mock.patch(
        'importlib.import_module',
        side_effect=[ImportError, ImportError, _mock_async],
    )
    def test_async_mode_auto_gevent(self, import_module):
        s = server.Server()
        assert s.async_mode == 'gevent'

    @mock.patch(
        'importlib.import_module',
        side_effect=[ImportError, ImportError, ImportError, _mock_async],
    )
    def test_async_mode_auto_threading(self, import_module):
        s = server.Server()
        assert s.async_mode == 'threading'

    def test_generate_id(self):
        s = server.Server()
        assert s._generate_id() != s._generate_id()

    def test_on_event(self):
        s = server.Server()

        @s.on('connect')
        def foo():
            pass

        s.on('disconnect', foo)

        assert s.handlers['connect'] == foo
        assert s.handlers['disconnect'] == foo

    def test_on_event_invalid(self):
        s = server.Server()
        with pytest.raises(ValueError):
            s.on('invalid')

    def test_trigger_event(self):
        s = server.Server()
        f = {}

        @s.on('connect')
        def foo(sid, environ):
            return sid + environ

        @s.on('message')
        def bar(sid, data):
            f['bar'] = sid + data
            return 'bar'

        r = s._trigger_event('connect', 1, 2, run_async=False)
        assert r == 3
        r = s._trigger_event('message', 3, 4, run_async=True)
        r.join()
        assert f['bar'] == 7
        r = s._trigger_event('message', 5, 6)
        assert r == 'bar'

    def test_trigger_event_error(self):
        s = server.Server()

        @s.on('connect')
        def foo(sid, environ):
            return 1 / 0

        @s.on('message')
        def bar(sid, data):
            return 1 / 0

        r = s._trigger_event('connect', 1, 2, run_async=False)
        assert not r
        r = s._trigger_event('message', 3, 4, run_async=False)
        assert r is None

    def test_session(self):
        s = server.Server()
        mock_socket = self._get_mock_socket()
        s.sockets['foo'] = mock_socket
        with s.session('foo') as session:
            assert session == {}
            session['username'] = 'bar'
        assert s.get_session('foo') == {'username': 'bar'}

    def test_close_one_socket(self):
        s = server.Server()
        mock_socket = self._get_mock_socket()
        s.sockets['foo'] = mock_socket
        s.disconnect('foo')
        assert mock_socket.close.call_count == 1
        assert 'foo' not in s.sockets

    def test_close_all_sockets(self):
        s = server.Server()
        mock_sockets = {}
        for sid in ['foo', 'bar', 'baz']:
            mock_sockets[sid] = self._get_mock_socket()
            s.sockets[sid] = mock_sockets[sid]
        s.disconnect()
        for socket in six.itervalues(mock_sockets):
            assert socket.close.call_count == 1
        assert s.sockets == {}

    def test_upgrades(self):
        s = server.Server()
        s.sockets['foo'] = self._get_mock_socket()
        assert s._upgrades('foo', 'polling') == ['websocket']
        assert s._upgrades('foo', 'websocket') == []
        s.sockets['foo'].upgraded = True
        assert s._upgrades('foo', 'polling') == []
        assert s._upgrades('foo', 'websocket') == []
        s.allow_upgrades = False
        s.sockets['foo'].upgraded = True
        assert s._upgrades('foo', 'polling') == []
        assert s._upgrades('foo', 'websocket') == []

    def test_transport(self):
        s = server.Server()
        s.sockets['foo'] = self._get_mock_socket()
        s.sockets['foo'].upgraded = False
        s.sockets['bar'] = self._get_mock_socket()
        s.sockets['bar'].upgraded = True
        assert s.transport('foo') == 'polling'
        assert s.transport('bar') == 'websocket'

    def test_bad_session(self):
        s = server.Server()
        s.sockets['foo'] = 'client'
        with pytest.raises(KeyError):
            s._get_socket('bar')

    def test_closed_socket(self):
        s = server.Server()
        s.sockets['foo'] = self._get_mock_socket()
        s.sockets['foo'].closed = True
        with pytest.raises(KeyError):
            s._get_socket('foo')

    def test_jsonp_with_bad_index(self):
        s = server.Server()
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3&j=abc'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '400 BAD REQUEST'

    def test_jsonp_index(self):
        s = server.Server()
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3&j=233'}
        start_response = mock.MagicMock()
        r = s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '200 OK'
        assert r[0].startswith(b'___eio[233]("')
        assert r[0].endswith(b'");')

    def test_connect(self):
        s = server.Server()
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        r = s.handle_request(environ, start_response)
        assert len(s.sockets) == 1
        assert start_response.call_count == 1
        assert start_response.call_args[0][0] == '200 OK'
        assert (
            'Content-Type',
            'application/octet-stream',
        ) in start_response.call_args[0][1]
        assert len(r) == 1
        packets = payload.Payload(encoded_payload=r[0]).packets
        assert len(packets) == 1
        assert packets[0].packet_type == packet.OPEN
        assert 'upgrades' in packets[0].data
        assert packets[0].data['upgrades'] == ['websocket']
        assert 'sid' in packets[0].data

    def test_connect_no_upgrades(self):
        s = server.Server(allow_upgrades=False)
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        r = s.handle_request(environ, start_response)
        packets = payload.Payload(encoded_payload=r[0]).packets
        assert packets[0].data['upgrades'] == []

    def test_connect_bad_eio_version(self):
        s = server.Server()
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=5'}
        start_response = mock.MagicMock()
        r = s.handle_request(environ, start_response)
        assert start_response.call_args[0][0], '400 BAD REQUEST'
        assert b'unsupported version' in r[0]

    def test_connect_b64_with_1(self):
        s = server.Server(allow_upgrades=False)
        s._generate_id = mock.MagicMock(return_value='1')
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3&b64=1'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0], '200 OK'
        assert (
            'Content-Type',
            'text/plain; charset=UTF-8',
        ) in start_response.call_args[0][1]
        s.send('1', b'\x00\x01\x02', binary=True)
        environ = {'REQUEST_METHOD': 'GET',
                   'QUERY_STRING': 'EIO=3&sid=1&b64=1'}
        r = s.handle_request(environ, start_response)
        assert r[0] == b'6:b4AAEC'

    def test_connect_b64_with_true(self):
        s = server.Server(allow_upgrades=False)
        s._generate_id = mock.MagicMock(return_value='1')
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3&b64=true'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0], '200 OK'
        assert (
            'Content-Type',
            'text/plain; charset=UTF-8',
        ) in start_response.call_args[0][1]
        s.send('1', b'\x00\x01\x02', binary=True)
        environ = {'REQUEST_METHOD': 'GET',
                   'QUERY_STRING': 'EIO=3&sid=1&b64=true'}
        r = s.handle_request(environ, start_response)
        assert r[0] == b'6:b4AAEC'

    def test_connect_b64_with_0(self):
        s = server.Server(allow_upgrades=False)
        s._generate_id = mock.MagicMock(return_value='1')
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3&b64=0'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0], '200 OK'
        assert (
            'Content-Type',
            'application/octet-stream',
        ) in start_response.call_args[0][1]
        s.send('1', b'\x00\x01\x02', binary=True)
        environ = {'REQUEST_METHOD': 'GET',
                   'QUERY_STRING': 'EIO=3&sid=1&b64=0'}
        r = s.handle_request(environ, start_response)
        assert r[0] == b'\x01\x04\xff\x04\x00\x01\x02'

    def test_connect_b64_with_false(self):
        s = server.Server(allow_upgrades=False)
        s._generate_id = mock.MagicMock(return_value='1')
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3&b64=false'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0], '200 OK'
        assert (
            'Content-Type',
            'application/octet-stream',
        ) in start_response.call_args[0][1]
        s.send('1', b'\x00\x01\x02', binary=True)
        environ = {'REQUEST_METHOD': 'GET',
                   'QUERY_STRING': 'EIO=3&sid=1&b64=false'}
        r = s.handle_request(environ, start_response)
        assert r[0] == b'\x01\x04\xff\x04\x00\x01\x02'

    def test_connect_custom_ping_times(self):
        s = server.Server(ping_timeout=123, ping_interval=456)
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        r = s.handle_request(environ, start_response)
        packets = payload.Payload(encoded_payload=r[0]).packets
        assert packets[0].data['pingTimeout'] == 123000
        assert packets[0].data['pingInterval'] == 456000

    @mock.patch(
        'engineio.socket.Socket.poll', side_effect=exceptions.QueueEmpty
    )
    def test_connect_bad_poll(self, poll):
        s = server.Server()
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '400 BAD REQUEST'

    @mock.patch(
        'engineio.socket.Socket',
        return_value=mock.MagicMock(connected=False, closed=False),
    )
    def test_connect_transport_websocket(self, Socket):
        s = server.Server()
        s._generate_id = mock.MagicMock(return_value='123')
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3&transport=websocket',
            'HTTP_UPGRADE': 'websocket',
        }
        start_response = mock.MagicMock()
        # force socket to stay open, so that we can check it later
        Socket().closed = False
        s.handle_request(environ, start_response)
        assert s.sockets['123'].send.call_args[0][0].packet_type == packet.OPEN

    @mock.patch(
        'engineio.socket.Socket',
        return_value=mock.MagicMock(connected=False, closed=False),
    )
    def test_connect_transport_websocket_closed(self, Socket):
        s = server.Server()
        s._generate_id = mock.MagicMock(return_value='123')
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3&transport=websocket',
            'HTTP_UPGRADE': 'websocket',
        }
        start_response = mock.MagicMock()

        def mock_handle(environ, start_response):
            s.sockets['123'].closed = True

        Socket().handle_get_request = mock_handle
        s.handle_request(environ, start_response)
        assert '123' not in s.sockets

    def test_connect_transport_invalid(self):
        s = server.Server()
        environ = {'REQUEST_METHOD': 'GET',
                   'QUERY_STRING': 'EIO=3&transport=foo'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '400 BAD REQUEST'

    def test_connect_transport_websocket_without_upgrade(self):
        s = server.Server()
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3&transport=websocket',
        }
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '400 BAD REQUEST'

    def test_connect_cors_headers(self):
        s = server.Server()
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '200 OK'
        headers = start_response.call_args[0][1]
        assert ('Access-Control-Allow-Credentials', 'true') in headers

    def test_connect_cors_allowed_origin(self):
        s = server.Server(cors_allowed_origins=['a', 'b'])
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3',
            'HTTP_ORIGIN': 'b',
        }
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '200 OK'
        headers = start_response.call_args[0][1]
        assert ('Access-Control-Allow-Origin', 'b') in headers

    def test_connect_cors_not_allowed_origin(self):
        s = server.Server(cors_allowed_origins=['a', 'b'])
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3',
            'HTTP_ORIGIN': 'c',
        }
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '400 BAD REQUEST'
        headers = start_response.call_args[0][1]
        assert ('Access-Control-Allow-Origin', 'c') not in headers
        assert ('Access-Control-Allow-Origin', '*') not in headers

    def test_connect_cors_headers_all_origins(self):
        s = server.Server(cors_allowed_origins='*')
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3',
            'HTTP_ORIGIN': 'foo',
        }
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '200 OK'
        headers = start_response.call_args[0][1]
        assert ('Access-Control-Allow-Origin', 'foo') in headers
        assert ('Access-Control-Allow-Credentials', 'true') in headers

    def test_connect_cors_headers_one_origin(self):
        s = server.Server(cors_allowed_origins='a')
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3',
            'HTTP_ORIGIN': 'a',
        }
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '200 OK'
        headers = start_response.call_args[0][1]
        assert ('Access-Control-Allow-Origin', 'a') in headers
        assert ('Access-Control-Allow-Credentials', 'true') in headers

    def test_connect_cors_headers_one_origin_not_allowed(self):
        s = server.Server(cors_allowed_origins='a')
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3',
            'HTTP_ORIGIN': 'b',
        }
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '400 BAD REQUEST'
        headers = start_response.call_args[0][1]
        assert ('Access-Control-Allow-Origin', 'b') not in headers
        assert ('Access-Control-Allow-Origin', '*') not in headers

    def test_connect_cors_headers_default_origin(self):
        s = server.Server()
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3',
            'wsgi.url_scheme': 'http',
            'HTTP_HOST': 'foo',
            'HTTP_ORIGIN': 'http://foo',
        }
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '200 OK'
        headers = start_response.call_args[0][1]
        assert ('Access-Control-Allow-Origin', 'http://foo') in headers

    def test_connect_cors_headers_default_origin_proxy_server(self):
        s = server.Server()
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3',
            'wsgi.url_scheme': 'http',
            'HTTP_HOST': 'foo',
            'HTTP_ORIGIN': 'https://bar',
            'HTTP_X_FORWARDED_PROTO': 'https, ftp',
            'HTTP_X_FORWARDED_HOST': 'bar , baz',
        }
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '200 OK'
        headers = start_response.call_args[0][1]
        assert ('Access-Control-Allow-Origin', 'https://bar') in headers

    def test_connect_cors_no_credentials(self):
        s = server.Server(cors_credentials=False)
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '200 OK'
        headers = start_response.call_args[0][1]
        assert ('Access-Control-Allow-Credentials', 'true') not in headers

    def test_cors_options(self):
        s = server.Server()
        environ = {'REQUEST_METHOD': 'OPTIONS', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '200 OK'
        headers = start_response.call_args[0][1]
        assert (
            'Access-Control-Allow-Methods',
            'OPTIONS, GET, POST',
        ) in headers

    def test_cors_request_headers(self):
        s = server.Server()
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3',
            'HTTP_ACCESS_CONTROL_REQUEST_HEADERS': 'Foo, Bar',
        }
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '200 OK'
        headers = start_response.call_args[0][1]
        assert ('Access-Control-Allow-Headers', 'Foo, Bar') in headers

    def test_connect_cors_disabled(self):
        s = server.Server(cors_allowed_origins=[])
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3',
            'HTTP_ORIGIN': 'http://foo',
        }
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '200 OK'
        headers = start_response.call_args[0][1]
        for header in headers:
            assert not header[0].startswith('Access-Control-')

    def test_connect_cors_default_no_origin(self):
        s = server.Server()
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        headers = start_response.call_args[0][1]
        for header in headers:
            assert header[0] != 'Access-Control-Allow-Origin'

    def test_connect_cors_all_no_origin(self):
        s = server.Server(cors_allowed_origins='*')
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        headers = start_response.call_args[0][1]
        for header in headers:
            assert header[0] != 'Access-Control-Allow-Origin'

    def test_connect_cors_disabled_no_origin(self):
        s = server.Server(cors_allowed_origins=[])
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        headers = start_response.call_args[0][1]
        for header in headers:
            assert header[0] != 'Access-Control-Allow-Origin'

    def test_connect_event(self):
        s = server.Server()
        s._generate_id = mock.MagicMock(return_value='123')
        mock_event = mock.MagicMock(return_value=None)
        s.on('connect')(mock_event)
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        mock_event.assert_called_once_with('123', environ)
        assert len(s.sockets) == 1

    def test_connect_event_rejects(self):
        s = server.Server()
        s._generate_id = mock.MagicMock(return_value='123')
        mock_event = mock.MagicMock(return_value=False)
        s.on('connect')(mock_event)
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        ret = s.handle_request(environ, start_response)
        assert len(s.sockets) == 0
        assert start_response.call_args[0][0] == '401 UNAUTHORIZED'
        assert ret == [b'"Unauthorized"']

    def test_connect_event_rejects_with_message(self):
        s = server.Server()
        s._generate_id = mock.MagicMock(return_value='123')
        mock_event = mock.MagicMock(return_value='not allowed')
        s.on('connect')(mock_event)
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        ret = s.handle_request(environ, start_response)
        assert len(s.sockets) == 0
        assert start_response.call_args[0][0] == '401 UNAUTHORIZED'
        assert ret == [b'"not allowed"']

    def test_method_not_found(self):
        s = server.Server()
        environ = {'REQUEST_METHOD': 'PUT', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '405 METHOD NOT FOUND'

    def test_get_request_with_bad_sid(self):
        s = server.Server()
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3&sid=foo'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '400 BAD REQUEST'

    def test_post_request_with_bad_sid(self):
        s = server.Server()
        environ = {'REQUEST_METHOD': 'POST', 'QUERY_STRING': 'EIO=3&sid=foo'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '400 BAD REQUEST'

    def test_send(self):
        s = server.Server()
        mock_socket = self._get_mock_socket()
        s.sockets['foo'] = mock_socket
        s.send('foo', 'hello')
        assert mock_socket.send.call_count == 1
        assert mock_socket.send.call_args[0][0].packet_type == packet.MESSAGE
        assert mock_socket.send.call_args[0][0].data == 'hello'

    def test_send_unknown_socket(self):
        s = server.Server()
        # just ensure no exceptions are raised
        s.send('foo', 'hello')

    def test_get_request(self):
        s = server.Server()
        mock_socket = self._get_mock_socket()
        mock_socket.handle_get_request = mock.MagicMock(
            return_value=[packet.Packet(packet.MESSAGE, data='hello')]
        )
        s.sockets['foo'] = mock_socket
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3&sid=foo'}
        start_response = mock.MagicMock()
        r = s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '200 OK'
        assert len(r) == 1
        packets = payload.Payload(encoded_payload=r[0]).packets
        assert len(packets) == 1
        assert packets[0].packet_type == packet.MESSAGE

    def test_get_request_custom_response(self):
        s = server.Server()
        mock_socket = self._get_mock_socket()
        mock_socket.handle_get_request = mock.MagicMock(side_effect=['resp'])
        s.sockets['foo'] = mock_socket
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3&sid=foo'}
        start_response = mock.MagicMock()
        assert s.handle_request(environ, start_response) == 'resp'

    def test_get_request_closes_socket(self):
        s = server.Server()
        mock_socket = self._get_mock_socket()

        def mock_get_request(*args, **kwargs):
            mock_socket.closed = True
            return 'resp'

        mock_socket.handle_get_request = mock_get_request
        s.sockets['foo'] = mock_socket
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3&sid=foo'}
        start_response = mock.MagicMock()
        assert s.handle_request(environ, start_response) == 'resp'
        assert 'foo' not in s.sockets

    def test_get_request_error(self):
        s = server.Server()
        mock_socket = self._get_mock_socket()
        mock_socket.handle_get_request = mock.MagicMock(
            side_effect=[exceptions.QueueEmpty]
        )
        s.sockets['foo'] = mock_socket
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3&sid=foo'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '400 BAD REQUEST'
        assert len(s.sockets) == 0

    def test_post_request(self):
        s = server.Server()
        mock_socket = self._get_mock_socket()
        mock_socket.handle_post_request = mock.MagicMock()
        s.sockets['foo'] = mock_socket
        environ = {'REQUEST_METHOD': 'POST', 'QUERY_STRING': 'EIO=3&sid=foo'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '200 OK'

    def test_post_request_error(self):
        s = server.Server()
        mock_socket = self._get_mock_socket()
        mock_socket.handle_post_request = mock.MagicMock(
            side_effect=[exceptions.EngineIOError]
        )
        s.sockets['foo'] = mock_socket
        environ = {'REQUEST_METHOD': 'POST', 'QUERY_STRING': 'EIO=3&sid=foo'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert start_response.call_args[0][0] == '400 BAD REQUEST'
        assert 'foo' not in s.sockets

    @staticmethod
    def _gzip_decompress(b):
        bytesio = six.BytesIO(b)
        with gzip.GzipFile(fileobj=bytesio, mode='r') as gz:
            return gz.read()

    def test_gzip_compression(self):
        s = server.Server(compression_threshold=0)
        mock_socket = self._get_mock_socket()
        mock_socket.handle_get_request = mock.MagicMock(
            return_value=[packet.Packet(packet.MESSAGE, data='hello')]
        )
        s.sockets['foo'] = mock_socket
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3&sid=foo',
            'HTTP_ACCEPT_ENCODING': 'gzip,deflate',
        }
        start_response = mock.MagicMock()
        r = s.handle_request(environ, start_response)
        assert ('Content-Encoding', 'gzip') in start_response.call_args[0][1]
        self._gzip_decompress(r[0])

    def test_deflate_compression(self):
        s = server.Server(compression_threshold=0)
        mock_socket = self._get_mock_socket()
        mock_socket.handle_get_request = mock.MagicMock(
            return_value=[packet.Packet(packet.MESSAGE, data='hello')]
        )
        s.sockets['foo'] = mock_socket
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3&sid=foo',
            'HTTP_ACCEPT_ENCODING': 'deflate;q=1,gzip',
        }
        start_response = mock.MagicMock()
        r = s.handle_request(environ, start_response)
        assert ('Content-Encoding', 'deflate') in start_response.call_args[0][
            1
        ]
        zlib.decompress(r[0])

    def test_gzip_compression_threshold(self):
        s = server.Server(compression_threshold=1000)
        mock_socket = self._get_mock_socket()
        mock_socket.handle_get_request = mock.MagicMock(
            return_value=[packet.Packet(packet.MESSAGE, data='hello')]
        )
        s.sockets['foo'] = mock_socket
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3&sid=foo',
            'HTTP_ACCEPT_ENCODING': 'gzip',
        }
        start_response = mock.MagicMock()
        r = s.handle_request(environ, start_response)
        for header, value in start_response.call_args[0][1]:
            assert header != 'Content-Encoding'
        with pytest.raises(IOError):
            self._gzip_decompress(r[0])

    def test_compression_disabled(self):
        s = server.Server(http_compression=False, compression_threshold=0)
        mock_socket = self._get_mock_socket()
        mock_socket.handle_get_request = mock.MagicMock(
            return_value=[packet.Packet(packet.MESSAGE, data='hello')]
        )
        s.sockets['foo'] = mock_socket
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3&sid=foo',
            'HTTP_ACCEPT_ENCODING': 'gzip',
        }
        start_response = mock.MagicMock()
        r = s.handle_request(environ, start_response)
        for header, value in start_response.call_args[0][1]:
            assert header != 'Content-Encoding'
        with pytest.raises(IOError):
            self._gzip_decompress(r[0])

    def test_compression_unknown(self):
        s = server.Server(compression_threshold=0)
        mock_socket = self._get_mock_socket()
        mock_socket.handle_get_request = mock.MagicMock(
            return_value=[packet.Packet(packet.MESSAGE, data='hello')]
        )
        s.sockets['foo'] = mock_socket
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3&sid=foo',
            'HTTP_ACCEPT_ENCODING': 'rar',
        }
        start_response = mock.MagicMock()
        r = s.handle_request(environ, start_response)
        for header, value in start_response.call_args[0][1]:
            assert header != 'Content-Encoding'
        with pytest.raises(IOError):
            self._gzip_decompress(r[0])

    def test_compression_no_encoding(self):
        s = server.Server(compression_threshold=0)
        mock_socket = self._get_mock_socket()
        mock_socket.handle_get_request = mock.MagicMock(
            return_value=[packet.Packet(packet.MESSAGE, data='hello')]
        )
        s.sockets['foo'] = mock_socket
        environ = {
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'EIO=3&sid=foo',
            'HTTP_ACCEPT_ENCODING': '',
        }
        start_response = mock.MagicMock()
        r = s.handle_request(environ, start_response)
        for header, value in start_response.call_args[0][1]:
            assert header != 'Content-Encoding'
        with pytest.raises(IOError):
            self._gzip_decompress(r[0])

    def test_cookie(self):
        s = server.Server(cookie='sid')
        s._generate_id = mock.MagicMock(return_value='123')
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert ('Set-Cookie', 'sid=123; path=/; SameSite=Lax') \
            in start_response.call_args[0][1]

    def test_cookie_dict(self):
        def get_path():
            return '/a'

        s = server.Server(cookie={
            'name': 'test',
            'path': get_path,
            'SameSite': 'None',
            'Secure': True,
            'HttpOnly': True
        })
        s._generate_id = mock.MagicMock(return_value='123')
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        assert ('Set-Cookie', 'test=123; path=/a; SameSite=None; Secure; '
                'HttpOnly') in start_response.call_args[0][1]

    def test_no_cookie(self):
        s = server.Server(cookie=None)
        s._generate_id = mock.MagicMock(return_value='123')
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        for header, value in start_response.call_args[0][1]:
            assert header != 'Set-Cookie'

    def test_logger(self):
        s = server.Server(logger=False)
        assert s.logger.getEffectiveLevel() == logging.ERROR
        s.logger.setLevel(logging.NOTSET)
        s = server.Server(logger=True)
        assert s.logger.getEffectiveLevel() == logging.INFO
        s.logger.setLevel(logging.WARNING)
        s = server.Server(logger=True)
        assert s.logger.getEffectiveLevel() == logging.WARNING
        s.logger.setLevel(logging.NOTSET)
        my_logger = logging.Logger('foo')
        s = server.Server(logger=my_logger)
        assert s.logger == my_logger

    def test_custom_json(self):
        # Warning: this test cannot run in parallel with other tests, as it
        # changes the JSON encoding/decoding functions

        class CustomJSON(object):
            @staticmethod
            def dumps(*args, **kwargs):
                return '*** encoded ***'

            @staticmethod
            def loads(*args, **kwargs):
                return '+++ decoded +++'

        server.Server(json=CustomJSON)
        pkt = packet.Packet(packet.MESSAGE, data={'foo': 'bar'})
        assert pkt.encode() == b'4*** encoded ***'
        pkt2 = packet.Packet(encoded_packet=pkt.encode())
        assert pkt2.data == '+++ decoded +++'

        # restore the default JSON module
        packet.Packet.json = json

    def test_background_tasks(self):
        flag = {}

        def bg_task():
            flag['task'] = True

        s = server.Server()
        task = s.start_background_task(bg_task)
        task.join()
        assert 'task' in flag
        assert flag['task']

    def test_sleep(self):
        s = server.Server()
        t = time.time()
        s.sleep(0.1)
        assert time.time() - t > 0.1

    def test_create_queue(self):
        s = server.Server()
        q = s.create_queue()
        empty = s.get_queue_empty_exception()
        with pytest.raises(empty):
            q.get(timeout=0.01)

    def test_create_event(self):
        s = server.Server()
        e = s.create_event()
        assert not e.is_set()
        e.set()
        assert e.is_set()

    def test_log_error_once(self):
        s = server.Server(logger=mock.MagicMock())
        s._log_error_once('foo', 'foo-key')
        s._log_error_once('foo', 'foo-key')
        s.logger.error.assert_called_with(
            'foo (further occurrences of this error will be logged with '
            'level INFO)')
        s.logger.info.assert_called_with('foo')

    def test_service_task_started(self):
        s = server.Server(monitor_clients=True)
        s._service_task = mock.MagicMock()
        environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=3'}
        start_response = mock.MagicMock()
        s.handle_request(environ, start_response)
        s._service_task.assert_called_once_with()
