diff options
| author | Paulo Henrique de Lima Santana (phls) <phls@debian.org> | 2024-04-09 16:16:45 -0300 |
|---|---|---|
| committer | git-ubuntu importer <ubuntu-devel-discuss@lists.ubuntu.com> | 2024-04-10 04:34:59 +0000 |
| commit | 49574c074decf4c2404cb93cd0b6d582dabbf71f (patch) | |
| tree | e4698d727f5b4706ec2c5b818c370abc422307c1 /tests | |
| parent | b7f96168e33c438b8f45adc4bbf79f7b773f4a62 (diff) | |
4.9.0-1 (patches unapplied)import/4.9.0-1
Imported using git-ubuntu import.
Notes
Notes:
* New upstream version 4.9.0.
* debian/control: bumped Standards-Version to 4.7.0.
* debian/copyright.
- Added Alexandre Detiste to packaging copyright.
- Removed files were deleted by upstream.
- Updated packaging copyright years.
* debian/docs: removed README.rst.
Diffstat (limited to 'tests')
27 files changed, 7837 insertions, 0 deletions
diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/tests/__init__.py diff --git a/tests/async/__init__.py b/tests/async/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/tests/async/__init__.py diff --git a/tests/async/index.html b/tests/async/index.html new file mode 100644 index 0000000..18ecdcb --- /dev/null +++ b/tests/async/index.html @@ -0,0 +1 @@ +<html></html> diff --git a/tests/async/test_aiohttp.py b/tests/async/test_aiohttp.py new file mode 100644 index 0000000..f89fb4b --- /dev/null +++ b/tests/async/test_aiohttp.py @@ -0,0 +1,53 @@ +import unittest +from unittest import mock + +from engineio.async_drivers import aiohttp as async_aiohttp + + +class AiohttpTests(unittest.TestCase): + def test_create_route(self): + app = mock.MagicMock() + mock_server = mock.MagicMock() + async_aiohttp.create_route(app, mock_server, '/foo') + app.router.add_get.assert_any_call('/foo', mock_server.handle_request) + app.router.add_post.assert_any_call('/foo', mock_server.handle_request) + + def test_translate_request(self): + request = mock.MagicMock() + request._message.method = 'PUT' + request._message.path = '/foo/bar?baz=1' + request._message.version = (1, 1) + request._message.headers = { + 'a': 'b', + 'c-c': 'd', + 'c_c': 'e', + 'content-type': 'application/json', + 'content-length': 123, + } + request._payload = b'hello world' + environ = async_aiohttp.translate_request(request) + expected_environ = { + 'REQUEST_METHOD': 'PUT', + 'PATH_INFO': '/foo/bar', + 'QUERY_STRING': 'baz=1', + 'CONTENT_TYPE': 'application/json', + 'CONTENT_LENGTH': 123, + 'HTTP_A': 'b', + # 'HTTP_C_C': 'd,e', + 'RAW_URI': '/foo/bar?baz=1', + 'SERVER_PROTOCOL': 'HTTP/1.1', + 'wsgi.input': b'hello world', + 'aiohttp.request': request, + } + for k, v in expected_environ.items(): + assert v == environ[k] + assert environ['HTTP_C_C'] == 'd,e' or environ['HTTP_C_C'] == 'e,d' + + # @mock.patch('async_aiohttp.aiohttp.web.Response') + def test_make_response(self): + rv = async_aiohttp.make_response( + '202 ACCEPTED', {'foo': 'bar'}, b'payload', {} + ) + assert rv.status == 202 + assert rv.headers['foo'] == 'bar' + assert rv.body == b'payload' diff --git a/tests/async/test_asgi.py b/tests/async/test_asgi.py new file mode 100644 index 0000000..d51c38b --- /dev/null +++ b/tests/async/test_asgi.py @@ -0,0 +1,516 @@ +import asyncio +import os +import unittest +from unittest import mock + +from engineio.async_drivers import asgi as async_asgi + + +def AsyncMock(*args, **kwargs): + """Return a mock asynchronous function.""" + m = mock.MagicMock(*args, **kwargs) + + async def mock_coro(*args, **kwargs): + return m(*args, **kwargs) + + mock_coro.mock = m + return mock_coro + + +def _run(coro): + """Run the given coroutine.""" + return asyncio.get_event_loop().run_until_complete(coro) + + +class AsgiTests(unittest.TestCase): + def test_create_app(self): + app = async_asgi.ASGIApp( + 'eio', + 'other_app', + static_files='static_files', + engineio_path='/foo/', + ) + assert app.engineio_server == 'eio' + assert app.other_asgi_app == 'other_app' + assert app.static_files == 'static_files' + assert app.engineio_path == '/foo/' + + def test_engineio_routing(self): + mock_server = mock.MagicMock() + mock_server.handle_request = AsyncMock() + app = async_asgi.ASGIApp(mock_server) + scope = {'type': 'http', 'path': '/engine.io/'} + _run(app(scope, 'receive', 'send')) + mock_server.handle_request.mock.assert_called_once_with( + scope, 'receive', 'send' + ) + + def test_other_app_routing(self): + other_app = AsyncMock() + app = async_asgi.ASGIApp('eio', other_app) + scope = {'type': 'http', 'path': '/foo'} + _run(app(scope, 'receive', 'send')) + other_app.mock.assert_called_once_with(scope, 'receive', 'send') + + def test_other_app_lifespan_routing(self): + other_app = AsyncMock() + app = async_asgi.ASGIApp('eio', other_app) + scope = {'type': 'lifespan'} + _run(app(scope, 'receive', 'send')) + other_app.mock.assert_called_once_with(scope, 'receive', 'send') + + def test_static_file_routing(self): + root_dir = os.path.dirname(__file__) + app = async_asgi.ASGIApp( + 'eio', + static_files={ + '/': root_dir + '/index.html', + '/foo': { + 'content_type': 'text/plain', + 'filename': root_dir + '/index.html', + }, + '/static': root_dir, + '/static/test/': root_dir + '/', + '/static2/test/': {'filename': root_dir + '/', + 'content_type': 'image/gif'}, + }, + ) + + def check_path(path, status_code, content_type, body): + scope = {'type': 'http', 'path': path} + receive = AsyncMock(return_value={'type': 'http.request'}) + send = AsyncMock() + _run(app(scope, receive, send)) + send.mock.assert_any_call( + { + 'type': 'http.response.start', + 'status': status_code, + 'headers': [ + (b'Content-Type', content_type.encode('utf-8')) + ], + } + ) + send.mock.assert_any_call( + {'type': 'http.response.body', 'body': body.encode('utf-8')} + ) + + check_path('/', 200, 'text/html', '<html></html>\n') + check_path('/foo', 200, 'text/plain', '<html></html>\n') + check_path('/foo/bar', 404, 'text/plain', 'Not Found') + check_path('/static/index.html', 200, 'text/html', '<html></html>\n') + check_path('/static/foo.bar', 404, 'text/plain', 'Not Found') + check_path( + '/static/test/index.html', 200, 'text/html', '<html></html>\n' + ) + check_path('/static/test/index.html', 200, 'text/html', + '<html></html>\n') + check_path('/static/test/files/', 200, 'text/html', + '<html>file</html>\n') + check_path('/static/test/files/file.txt', 200, 'text/plain', + 'file\n') + check_path('/static/test/files/x.html', 404, 'text/plain', + 'Not Found') + check_path('/static2/test/', 200, 'image/gif', '<html></html>\n') + check_path('/static2/test/index.html', 200, 'image/gif', + '<html></html>\n') + check_path('/static2/test/files/', 200, 'image/gif', + '<html>file</html>\n') + check_path('/static2/test/files/file.txt', 200, 'image/gif', + 'file\n') + check_path('/static2/test/files/x.html', 404, 'text/plain', + 'Not Found') + check_path('/bar/foo', 404, 'text/plain', 'Not Found') + check_path('', 404, 'text/plain', 'Not Found') + + app.static_files[''] = 'index.html' + check_path('/static/test/', 200, 'text/html', '<html></html>\n') + + app.static_files[''] = {'filename': 'index.html'} + check_path('/static/test/', 200, 'text/html', '<html></html>\n') + + app.static_files[''] = { + 'filename': 'index.html', + 'content_type': 'image/gif', + } + check_path('/static/test/', 200, 'image/gif', '<html></html>\n') + + app.static_files[''] = {'filename': 'test.gif'} + check_path('/static/test/', 404, 'text/plain', 'Not Found') + + app.static_files = {} + check_path('/static/test/index.html', 404, 'text/plain', 'Not Found') + + def test_lifespan_startup(self): + app = async_asgi.ASGIApp('eio') + scope = {'type': 'lifespan'} + receive = AsyncMock(side_effect=[{'type': 'lifespan.startup'}, + {'type': 'lifespan.shutdown'}]) + send = AsyncMock() + _run(app(scope, receive, send)) + send.mock.assert_any_call( + {'type': 'lifespan.startup.complete'} + ) + + def test_lifespan_startup_sync_function(self): + up = False + + def startup(): + nonlocal up + up = True + + app = async_asgi.ASGIApp('eio', on_startup=startup) + scope = {'type': 'lifespan'} + receive = AsyncMock(side_effect=[{'type': 'lifespan.startup'}, + {'type': 'lifespan.shutdown'}]) + send = AsyncMock() + _run(app(scope, receive, send)) + send.mock.assert_any_call( + {'type': 'lifespan.startup.complete'} + ) + assert up + + def test_lifespan_startup_async_function(self): + up = False + + async def startup(): + nonlocal up + up = True + + app = async_asgi.ASGIApp('eio', on_startup=startup) + scope = {'type': 'lifespan'} + receive = AsyncMock(side_effect=[{'type': 'lifespan.startup'}, + {'type': 'lifespan.shutdown'}]) + send = AsyncMock() + _run(app(scope, receive, send)) + send.mock.assert_any_call( + {'type': 'lifespan.startup.complete'} + ) + assert up + + def test_lifespan_startup_function_exception(self): + up = False + + def startup(): + raise Exception + + app = async_asgi.ASGIApp('eio', on_startup=startup) + scope = {'type': 'lifespan'} + receive = AsyncMock(side_effect=[{'type': 'lifespan.startup'}]) + send = AsyncMock() + _run(app(scope, receive, send)) + send.mock.assert_called_once_with({'type': 'lifespan.startup.failed'}) + assert not up + + def test_lifespan_shutdown(self): + app = async_asgi.ASGIApp('eio') + scope = {'type': 'lifespan'} + receive = AsyncMock(return_value={'type': 'lifespan.shutdown'}) + send = AsyncMock() + _run(app(scope, receive, send)) + send.mock.assert_called_once_with( + {'type': 'lifespan.shutdown.complete'} + ) + + def test_lifespan_shutdown_sync_function(self): + down = False + + def shutdown(): + nonlocal down + down = True + + app = async_asgi.ASGIApp('eio', on_shutdown=shutdown) + scope = {'type': 'lifespan'} + receive = AsyncMock(return_value={'type': 'lifespan.shutdown'}) + send = AsyncMock() + _run(app(scope, receive, send)) + send.mock.assert_called_once_with( + {'type': 'lifespan.shutdown.complete'} + ) + assert down + + def test_lifespan_shutdown_async_function(self): + down = False + + async def shutdown(): + nonlocal down + down = True + + app = async_asgi.ASGIApp('eio', on_shutdown=shutdown) + scope = {'type': 'lifespan'} + receive = AsyncMock(return_value={'type': 'lifespan.shutdown'}) + send = AsyncMock() + _run(app(scope, receive, send)) + send.mock.assert_called_once_with( + {'type': 'lifespan.shutdown.complete'} + ) + assert down + + def test_lifespan_shutdown_function_exception(self): + down = False + + def shutdown(): + raise Exception + + app = async_asgi.ASGIApp('eio', on_shutdown=shutdown) + scope = {'type': 'lifespan'} + receive = AsyncMock(return_value={'type': 'lifespan.shutdown'}) + send = AsyncMock() + _run(app(scope, receive, send)) + send.mock.assert_called_once_with({'type': 'lifespan.shutdown.failed'}) + assert not down + + def test_lifespan_invalid(self): + app = async_asgi.ASGIApp('eio') + scope = {'type': 'lifespan'} + receive = AsyncMock(side_effect=[{'type': 'lifespan.foo'}, + {'type': 'lifespan.shutdown'}]) + send = AsyncMock() + _run(app(scope, receive, send)) + send.mock.assert_called_once_with( + {'type': 'lifespan.shutdown.complete'} + ) + + def test_not_found(self): + app = async_asgi.ASGIApp('eio') + scope = {'type': 'http', 'path': '/foo'} + receive = AsyncMock(return_value={'type': 'http.request'}) + send = AsyncMock() + _run(app(scope, receive, send)) + send.mock.assert_any_call( + { + 'type': 'http.response.start', + 'status': 404, + 'headers': [(b'Content-Type', b'text/plain')], + } + ) + send.mock.assert_any_call( + {'type': 'http.response.body', 'body': b'Not Found'} + ) + + def test_translate_request(self): + receive = AsyncMock( + return_value={'type': 'http.request', 'body': b'hello world'} + ) + send = AsyncMock() + environ = _run( + async_asgi.translate_request( + { + 'type': 'http', + 'method': 'PUT', + 'headers': [ + (b'a', b'b'), + (b'c-c', b'd'), + (b'c_c', b'e'), + (b'content-type', b'application/json'), + (b'content-length', b'123'), + ], + 'path': '/foo/bar', + 'query_string': b'baz=1', + }, + receive, + send, + ) + ) + expected_environ = { + 'REQUEST_METHOD': 'PUT', + 'PATH_INFO': '/foo/bar', + 'QUERY_STRING': 'baz=1', + 'CONTENT_TYPE': 'application/json', + 'CONTENT_LENGTH': '123', + 'HTTP_A': 'b', + # 'HTTP_C_C': 'd,e', + 'RAW_URI': '/foo/bar?baz=1', + 'SERVER_PROTOCOL': 'HTTP/1.1', + 'asgi.receive': receive, + 'asgi.send': send, + } + for k, v in expected_environ.items(): + assert v == environ[k] + assert environ['HTTP_C_C'] == 'd,e' or environ['HTTP_C_C'] == 'e,d' + body = _run(environ['wsgi.input'].read()) + assert body == b'hello world' + + def test_translate_request_no_query_string(self): + receive = AsyncMock( + return_value={'type': 'http.request', 'body': b'hello world'} + ) + send = AsyncMock() + environ = _run( + async_asgi.translate_request( + { + 'type': 'http', + 'method': 'PUT', + 'headers': [ + (b'a', b'b'), + (b'c-c', b'd'), + (b'c_c', b'e'), + (b'content-type', b'application/json'), + (b'content-length', b'123'), + ], + 'path': '/foo/bar', + }, + receive, + send, + ) + ) + expected_environ = { + 'REQUEST_METHOD': 'PUT', + 'PATH_INFO': '/foo/bar', + 'QUERY_STRING': '', + 'CONTENT_TYPE': 'application/json', + 'CONTENT_LENGTH': '123', + 'HTTP_A': 'b', + # 'HTTP_C_C': 'd,e', + 'RAW_URI': '/foo/bar', + 'SERVER_PROTOCOL': 'HTTP/1.1', + 'asgi.receive': receive, + 'asgi.send': send, + } + for k, v in expected_environ.items(): + assert v == environ[k] + assert environ['HTTP_C_C'] == 'd,e' or environ['HTTP_C_C'] == 'e,d' + body = _run(environ['wsgi.input'].read()) + assert body == b'hello world' + + def test_translate_request_with_large_body(self): + receive = AsyncMock( + side_effect=[ + {'type': 'http.request', 'body': b'hello ', 'more_body': True}, + {'type': 'http.request', 'body': b'world', 'more_body': True}, + {'type': 'foo.bar'}, # should stop parsing here + {'type': 'http.request', 'body': b'!!!'}, + ] + ) + send = AsyncMock() + environ = _run( + async_asgi.translate_request( + { + 'type': 'http', + 'method': 'PUT', + 'headers': [ + (b'a', b'b'), + (b'c-c', b'd'), + (b'c_c', b'e'), + (b'content-type', b'application/json'), + (b'content-length', b'123'), + ], + 'path': '/foo/bar', + 'query_string': b'baz=1', + }, + receive, + send, + ) + ) + expected_environ = { + 'REQUEST_METHOD': 'PUT', + 'PATH_INFO': '/foo/bar', + 'QUERY_STRING': 'baz=1', + 'CONTENT_TYPE': 'application/json', + 'CONTENT_LENGTH': '123', + 'HTTP_A': 'b', + # 'HTTP_C_C': 'd,e', + 'RAW_URI': '/foo/bar?baz=1', + 'SERVER_PROTOCOL': 'HTTP/1.1', + 'asgi.receive': receive, + 'asgi.send': send, + } + for k, v in expected_environ.items(): + assert v == environ[k] + assert environ['HTTP_C_C'] == 'd,e' or environ['HTTP_C_C'] == 'e,d' + body = _run(environ['wsgi.input'].read()) + assert body == b'hello world' + + def test_translate_websocket_request(self): + receive = AsyncMock(return_value={'type': 'websocket.connect'}) + send = AsyncMock() + _run( + async_asgi.translate_request( + { + 'type': 'websocket', + 'headers': [ + (b'a', b'b'), + (b'c-c', b'd'), + (b'c_c', b'e'), + (b'content-type', b'application/json'), + (b'content-length', b'123'), + ], + 'path': '/foo/bar', + 'query_string': b'baz=1', + }, + receive, + send, + ) + ) + send.mock.assert_not_called() + + def test_translate_unknown_request(self): + receive = AsyncMock(return_value={'type': 'http.foo'}) + send = AsyncMock() + environ = _run( + async_asgi.translate_request( + {'type': 'http', 'path': '/foo/bar', 'query_string': b'baz=1'}, + receive, + send, + ) + ) + assert environ == {} + + def test_make_response(self): + environ = {'asgi.send': AsyncMock(), 'asgi.scope': {'type': 'http'}} + _run( + async_asgi.make_response( + '202 ACCEPTED', [('foo', 'bar')], b'payload', environ + ) + ) + environ['asgi.send'].mock.assert_any_call( + { + 'type': 'http.response.start', + 'status': 202, + 'headers': [(b'foo', b'bar')], + } + ) + environ['asgi.send'].mock.assert_any_call( + {'type': 'http.response.body', 'body': b'payload'} + ) + + def test_make_response_websocket_accept(self): + environ = { + 'asgi.send': AsyncMock(), + 'asgi.scope': {'type': 'websocket'}, + } + _run( + async_asgi.make_response( + '200 OK', [('foo', 'bar')], b'payload', environ + ) + ) + environ['asgi.send'].mock.assert_called_with( + {'type': 'websocket.accept', 'headers': [(b'foo', b'bar')]} + ) + + def test_make_response_websocket_reject(self): + environ = { + 'asgi.send': AsyncMock(), + 'asgi.scope': {'type': 'websocket'}, + } + _run( + async_asgi.make_response( + '401 UNAUTHORIZED', [('foo', 'bar')], b'payload', environ + ) + ) + environ['asgi.send'].mock.assert_called_with( + {'type': 'websocket.close', 'reason': 'payload'} + ) + + def test_make_response_websocket_reject_no_payload(self): + environ = { + 'asgi.send': AsyncMock(), + 'asgi.scope': {'type': 'websocket'}, + } + _run( + async_asgi.make_response( + '401 UNAUTHORIZED', [('foo', 'bar')], None, environ + ) + ) + environ['asgi.send'].mock.assert_called_with( + {'type': 'websocket.close'} + ) diff --git a/tests/async/test_client.py b/tests/async/test_client.py new file mode 100644 index 0000000..ee5f8ef --- /dev/null +++ b/tests/async/test_client.py @@ -0,0 +1,1489 @@ +import asyncio +import ssl +import sys +import unittest +from unittest import mock + +try: + import aiohttp +except ImportError: + aiohttp = None +import pytest + +from engineio import async_client +from engineio import base_client +from engineio import exceptions +from engineio import packet +from engineio import payload + + +def AsyncMock(*args, **kwargs): + """Return a mock asynchronous function.""" + m = mock.MagicMock(*args, **kwargs) + + async def mock_coro(*args, **kwargs): + return m(*args, **kwargs) + + mock_coro.mock = m + return mock_coro + + +def _run(coro): + """Run the given coroutine.""" + return asyncio.get_event_loop().run_until_complete(coro) + + +@unittest.skipIf(sys.version_info < (3, 5), 'only for Python 3.5+') +class TestAsyncClient(unittest.TestCase): + def test_is_asyncio_based(self): + c = async_client.AsyncClient() + assert c.is_asyncio_based() + + def test_already_connected(self): + c = async_client.AsyncClient() + c.state = 'connected' + with pytest.raises(ValueError): + _run(c.connect('http://foo')) + + def test_invalid_transports(self): + c = async_client.AsyncClient() + with pytest.raises(ValueError): + _run(c.connect('http://foo', transports=['foo', 'bar'])) + + def test_some_invalid_transports(self): + c = async_client.AsyncClient() + c._connect_websocket = AsyncMock() + _run(c.connect('http://foo', transports=['foo', 'websocket', 'bar'])) + assert c.transports == ['websocket'] + + def test_connect_polling(self): + c = async_client.AsyncClient() + c._connect_polling = AsyncMock(return_value='foo') + assert _run(c.connect('http://foo')) == 'foo' + c._connect_polling.mock.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + + c = async_client.AsyncClient() + c._connect_polling = AsyncMock(return_value='foo') + assert _run(c.connect('http://foo', transports=['polling'])) == 'foo' + c._connect_polling.mock.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + + c = async_client.AsyncClient() + c._connect_polling = AsyncMock(return_value='foo') + assert ( + _run(c.connect('http://foo', transports=['polling', 'websocket'])) + == 'foo' + ) + c._connect_polling.mock.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + + def test_connect_websocket(self): + c = async_client.AsyncClient() + c._connect_websocket = AsyncMock(return_value='foo') + assert _run(c.connect('http://foo', transports=['websocket'])) == 'foo' + c._connect_websocket.mock.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + + c = async_client.AsyncClient() + c._connect_websocket = AsyncMock(return_value='foo') + assert _run(c.connect('http://foo', transports='websocket')) == 'foo' + c._connect_websocket.mock.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + + def test_connect_query_string(self): + c = async_client.AsyncClient() + c._connect_polling = AsyncMock(return_value='foo') + assert _run(c.connect('http://foo?bar=baz')) == 'foo' + c._connect_polling.mock.assert_called_once_with( + 'http://foo?bar=baz', {}, 'engine.io' + ) + + def test_connect_custom_headers(self): + c = async_client.AsyncClient() + c._connect_polling = AsyncMock(return_value='foo') + assert _run(c.connect('http://foo', headers={'Foo': 'Bar'})) == 'foo' + c._connect_polling.mock.assert_called_once_with( + 'http://foo', {'Foo': 'Bar'}, 'engine.io' + ) + + def test_wait(self): + c = async_client.AsyncClient() + done = [] + + async def fake_read_look_task(): + done.append(True) + + c.read_loop_task = fake_read_look_task() + _run(c.wait()) + assert done == [True] + + def test_wait_no_task(self): + c = async_client.AsyncClient() + c.read_loop_task = None + _run(c.wait()) + + def test_send(self): + c = async_client.AsyncClient() + saved_packets = [] + + async def fake_send_packet(pkt): + saved_packets.append(pkt) + + c._send_packet = fake_send_packet + _run(c.send('foo')) + _run(c.send('foo')) + _run(c.send(b'foo')) + assert saved_packets[0].packet_type == packet.MESSAGE + assert saved_packets[0].data == 'foo' + assert not saved_packets[0].binary + assert saved_packets[1].packet_type == packet.MESSAGE + assert saved_packets[1].data == 'foo' + assert not saved_packets[1].binary + assert saved_packets[2].packet_type == packet.MESSAGE + assert saved_packets[2].data == b'foo' + assert saved_packets[2].binary + + def test_disconnect_not_connected(self): + c = async_client.AsyncClient() + c.state = 'foo' + c.sid = 'bar' + _run(c.disconnect()) + assert c.state == 'disconnected' + assert c.sid is None + + def test_disconnect_polling(self): + c = async_client.AsyncClient() + base_client.connected_clients.append(c) + c.state = 'connected' + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.queue.put = AsyncMock() + c.queue.join = AsyncMock() + c.read_loop_task = AsyncMock()() + c.ws = mock.MagicMock() + c.ws.close = AsyncMock() + c._trigger_event = AsyncMock() + _run(c.disconnect()) + c.ws.close.mock.assert_not_called() + assert c not in base_client.connected_clients + c._trigger_event.mock.assert_called_once_with( + 'disconnect', run_async=False + ) + + def test_disconnect_websocket(self): + c = async_client.AsyncClient() + base_client.connected_clients.append(c) + c.state = 'connected' + c.current_transport = 'websocket' + c.queue = mock.MagicMock() + c.queue.put = AsyncMock() + c.queue.join = AsyncMock() + c.read_loop_task = AsyncMock()() + c.ws = mock.MagicMock() + c.ws.close = AsyncMock() + c._trigger_event = AsyncMock() + _run(c.disconnect()) + c.ws.close.mock.assert_called_once_with() + assert c not in base_client.connected_clients + c._trigger_event.mock.assert_called_once_with( + 'disconnect', run_async=False + ) + + def test_disconnect_polling_abort(self): + c = async_client.AsyncClient() + base_client.connected_clients.append(c) + c.state = 'connected' + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.queue.put = AsyncMock() + c.queue.join = AsyncMock() + c.read_loop_task = AsyncMock()() + c.ws = mock.MagicMock() + c.ws.close = AsyncMock() + _run(c.disconnect(abort=True)) + c.queue.join.mock.assert_not_called() + c.ws.close.mock.assert_not_called() + assert c not in base_client.connected_clients + + def test_disconnect_websocket_abort(self): + c = async_client.AsyncClient() + base_client.connected_clients.append(c) + c.state = 'connected' + c.current_transport = 'websocket' + c.queue = mock.MagicMock() + c.queue.put = AsyncMock() + c.queue.join = AsyncMock() + c.read_loop_task = AsyncMock()() + c.ws = mock.MagicMock() + c.ws.close = AsyncMock() + _run(c.disconnect(abort=True)) + c.queue.join.mock.assert_not_called() + c.ws.mock.assert_not_called() + assert c not in base_client.connected_clients + + def test_background_tasks(self): + r = [] + + async def foo(arg): + r.append(arg) + + c = async_client.AsyncClient() + c.start_background_task(foo, 'bar') + pending = asyncio.all_tasks(loop=asyncio.get_event_loop()) \ + if hasattr(asyncio, 'all_tasks') else asyncio.Task.all_tasks() + asyncio.get_event_loop().run_until_complete(asyncio.wait(pending)) + assert r == ['bar'] + + def test_sleep(self): + c = async_client.AsyncClient() + _run(c.sleep(0)) + + def test_create_queue(self): + c = async_client.AsyncClient() + q = c.create_queue() + with pytest.raises(q.Empty): + q.get_nowait() + + def test_create_event(self): + c = async_client.AsyncClient() + e = c.create_event() + assert not e.is_set() + e.set() + assert e.is_set() + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_polling_connection_failed(self, _time): + c = async_client.AsyncClient() + c._send_request = AsyncMock(return_value=None) + with pytest.raises(exceptions.ConnectionError): + _run(c.connect('http://foo', headers={'Foo': 'Bar'})) + c._send_request.mock.assert_called_once_with( + 'GET', + 'http://foo/engine.io/?transport=polling&EIO=4&t=123.456', + headers={'Foo': 'Bar'}, + timeout=5, + ) + + def test_polling_connection_404(self): + c = async_client.AsyncClient() + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 404 + c._send_request.mock.return_value.json = AsyncMock( + return_value={'foo': 'bar'} + ) + try: + _run(c.connect('http://foo')) + except exceptions.ConnectionError as exc: + assert len(exc.args) == 2 + assert ( + exc.args[0] == 'Unexpected status code 404 in server response' + ) + assert exc.args[1] == {'foo': 'bar'} + + def test_polling_connection_404_no_json(self): + c = async_client.AsyncClient() + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 404 + c._send_request.mock.return_value.json = AsyncMock( + side_effect=aiohttp.ContentTypeError('foo', 'bar') + ) + try: + _run(c.connect('http://foo')) + except exceptions.ConnectionError as exc: + assert len(exc.args) == 2 + assert ( + exc.args[0] == 'Unexpected status code 404 in server response' + ) + assert exc.args[1] is None + + def test_polling_connection_invalid_packet(self): + c = async_client.AsyncClient() + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 200 + c._send_request.mock.return_value.read = AsyncMock(return_value=b'foo') + with pytest.raises(exceptions.ConnectionError): + _run(c.connect('http://foo')) + + def test_polling_connection_no_open_packet(self): + c = async_client.AsyncClient() + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 200 + c._send_request.mock.return_value.read = AsyncMock( + return_value=payload.Payload( + packets=[ + packet.Packet( + packet.CLOSE, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 10, + 'pingTimeout': 20, + }, + ) + ] + ).encode().encode('utf-8') + ) + with pytest.raises(exceptions.ConnectionError): + _run(c.connect('http://foo')) + + def test_polling_connection_successful(self): + c = async_client.AsyncClient() + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 200 + c._send_request.mock.return_value.read = AsyncMock( + return_value=payload.Payload( + packets=[ + packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ) + ] + ).encode().encode('utf-8') + ) + c._read_loop_polling = AsyncMock() + c._read_loop_websocket = AsyncMock() + c._write_loop = AsyncMock() + on_connect = AsyncMock() + c.on('connect', on_connect) + _run(c.connect('http://foo')) + + c._read_loop_polling.mock.assert_called_once_with() + c._read_loop_websocket.mock.assert_not_called() + c._write_loop.mock.assert_called_once_with() + on_connect.mock.assert_called_once_with() + assert c in base_client.connected_clients + assert ( + c.base_url + == 'http://foo/engine.io/?transport=polling&EIO=4&sid=123' + ) + assert c.sid == '123' + assert c.ping_interval == 1 + assert c.ping_timeout == 2 + assert c.upgrades == [] + assert c.transport() == 'polling' + + def test_polling_https_noverify_connection_successful(self): + c = async_client.AsyncClient(ssl_verify=False) + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 200 + c._send_request.mock.return_value.read = AsyncMock( + return_value=payload.Payload( + packets=[ + packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ) + ] + ).encode().encode('utf-8') + ) + c._read_loop_polling = AsyncMock() + c._read_loop_websocket = AsyncMock() + c._write_loop = AsyncMock() + on_connect = AsyncMock() + c.on('connect', on_connect) + _run(c.connect('https://foo')) + + c._read_loop_polling.mock.assert_called_once_with() + c._read_loop_websocket.mock.assert_not_called() + c._write_loop.mock.assert_called_once_with() + on_connect.mock.assert_called_once_with() + assert c in base_client.connected_clients + assert ( + c.base_url + == 'https://foo/engine.io/?transport=polling&EIO=4&sid=123' + ) + assert c.sid == '123' + assert c.ping_interval == 1 + assert c.ping_timeout == 2 + assert c.upgrades == [] + assert c.transport() == 'polling' + + def test_polling_connection_with_more_packets(self): + c = async_client.AsyncClient() + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 200 + c._send_request.mock.return_value.read = AsyncMock( + return_value=payload.Payload( + packets=[ + packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ), + packet.Packet(packet.NOOP), + ] + ).encode().encode('utf-8') + ) + c._read_loop_polling = AsyncMock() + c._read_loop_websocket = AsyncMock() + c._write_loop = AsyncMock() + c._receive_packet = AsyncMock() + on_connect = AsyncMock() + c.on('connect', on_connect) + _run(c.connect('http://foo')) + assert c._receive_packet.mock.call_count == 1 + assert ( + c._receive_packet.mock.call_args_list[0][0][0].packet_type + == packet.NOOP + ) + + def test_polling_connection_upgraded(self): + c = async_client.AsyncClient() + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 200 + c._send_request.mock.return_value.read = AsyncMock( + return_value=payload.Payload( + packets=[ + packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': ['websocket'], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ) + ] + ).encode().encode('utf-8') + ) + c._connect_websocket = AsyncMock(return_value=True) + on_connect = mock.MagicMock() + c.on('connect', on_connect) + _run(c.connect('http://foo')) + + c._connect_websocket.mock.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + on_connect.assert_called_once_with() + assert c in base_client.connected_clients + assert ( + c.base_url + == 'http://foo/engine.io/?transport=polling&EIO=4&sid=123' + ) + assert c.sid == '123' + assert c.ping_interval == 1 + assert c.ping_timeout == 2 + assert c.upgrades == ['websocket'] + + def test_polling_connection_not_upgraded(self): + c = async_client.AsyncClient() + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 200 + c._send_request.mock.return_value.read = AsyncMock( + return_value=payload.Payload( + packets=[ + packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': ['websocket'], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ) + ] + ).encode().encode('utf-8') + ) + c._connect_websocket = AsyncMock(return_value=False) + c._read_loop_polling = AsyncMock() + c._read_loop_websocket = AsyncMock() + c._write_loop = AsyncMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + _run(c.connect('http://foo')) + + c._connect_websocket.mock.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + c._read_loop_polling.mock.assert_called_once_with() + c._read_loop_websocket.mock.assert_not_called() + c._write_loop.mock.assert_called_once_with() + on_connect.assert_called_once_with() + assert c in base_client.connected_clients + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_websocket_connection_failed(self, _time): + c = async_client.AsyncClient() + c.http = mock.MagicMock(closed=False) + c.http.ws_connect = AsyncMock( + side_effect=[aiohttp.client_exceptions.ServerConnectionError()] + ) + with pytest.raises(exceptions.ConnectionError): + _run( + c.connect( + 'http://foo', + transports=['websocket'], + headers={'Foo': 'Bar'}, + ) + ) + c.http.ws_connect.mock.assert_called_once_with( + 'ws://foo/engine.io/?transport=websocket&EIO=4&t=123.456', + headers={'Foo': 'Bar'}, + timeout=5 + ) + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_websocket_connection_extra(self, _time): + c = async_client.AsyncClient(websocket_extra_options={ + 'headers': {'Baz': 'Qux'}, + 'timeout': 10 + }) + c.http = mock.MagicMock(closed=False) + c.http.ws_connect = AsyncMock( + side_effect=[aiohttp.client_exceptions.ServerConnectionError()] + ) + with pytest.raises(exceptions.ConnectionError): + _run( + c.connect( + 'http://foo', + transports=['websocket'], + headers={'Foo': 'Bar'}, + ) + ) + c.http.ws_connect.mock.assert_called_once_with( + 'ws://foo/engine.io/?transport=websocket&EIO=4&t=123.456', + headers={'Foo': 'Bar', 'Baz': 'Qux'}, + timeout=10, + ) + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_websocket_upgrade_failed(self, _time): + c = async_client.AsyncClient() + c.http = mock.MagicMock(closed=False) + c.http.ws_connect = AsyncMock( + side_effect=[aiohttp.client_exceptions.ServerConnectionError()] + ) + c.sid = '123' + assert not _run(c.connect('http://foo', transports=['websocket'])) + c.http.ws_connect.mock.assert_called_once_with( + 'ws://foo/engine.io/?transport=websocket&EIO=4&sid=123&t=123.456', + headers={}, + timeout=5, + ) + + def test_websocket_connection_no_open_packet(self): + c = async_client.AsyncClient() + c.http = mock.MagicMock(closed=False) + c.http.ws_connect = AsyncMock() + ws = c.http.ws_connect.mock.return_value + ws.receive = AsyncMock() + ws.receive.mock.return_value.data = packet.Packet( + packet.CLOSE + ).encode() + with pytest.raises(exceptions.ConnectionError): + _run(c.connect('http://foo', transports=['websocket'])) + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_websocket_connection_successful(self, _time): + c = async_client.AsyncClient() + c.http = mock.MagicMock(closed=False) + c.http.ws_connect = AsyncMock() + ws = c.http.ws_connect.mock.return_value + ws.receive = AsyncMock() + ws.receive.mock.return_value.data = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + c._read_loop_polling = AsyncMock() + c._read_loop_websocket = AsyncMock() + c._write_loop = AsyncMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + _run(c.connect('ws://foo', transports=['websocket'])) + + c._read_loop_polling.mock.assert_not_called() + c._read_loop_websocket.mock.assert_called_once_with() + c._write_loop.mock.assert_called_once_with() + on_connect.assert_called_once_with() + assert c in base_client.connected_clients + assert c.base_url == 'ws://foo/engine.io/?transport=websocket&EIO=4' + assert c.sid == '123' + assert c.ping_interval == 1 + assert c.ping_timeout == 2 + assert c.upgrades == [] + assert c.transport() == 'websocket' + assert c.ws == ws + c.http.ws_connect.mock.assert_called_once_with( + 'ws://foo/engine.io/?transport=websocket&EIO=4&t=123.456', + headers={}, + timeout=5, + ) + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_websocket_https_noverify_connection_successful(self, _time): + c = async_client.AsyncClient(ssl_verify=False) + c.http = mock.MagicMock(closed=False) + c.http.ws_connect = AsyncMock() + ws = c.http.ws_connect.mock.return_value + ws.receive = AsyncMock() + ws.receive.mock.return_value.data = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + c._read_loop_polling = AsyncMock() + c._read_loop_websocket = AsyncMock() + c._write_loop = AsyncMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + _run(c.connect('wss://foo', transports=['websocket'])) + + c._read_loop_polling.mock.assert_not_called() + c._read_loop_websocket.mock.assert_called_once_with() + c._write_loop.mock.assert_called_once_with() + on_connect.assert_called_once_with() + assert c in base_client.connected_clients + assert c.base_url == 'wss://foo/engine.io/?transport=websocket&EIO=4' + assert c.sid == '123' + assert c.ping_interval == 1 + assert c.ping_timeout == 2 + assert c.upgrades == [] + assert c.transport() == 'websocket' + assert c.ws == ws + _, kwargs = c.http.ws_connect.mock.call_args + assert 'ssl' in kwargs + assert isinstance(kwargs['ssl'], ssl.SSLContext) + assert kwargs['ssl'].verify_mode == ssl.CERT_NONE + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_websocket_connection_with_cookies(self, _time): + c = async_client.AsyncClient() + c.http = mock.MagicMock(closed=False) + c.http.ws_connect = AsyncMock() + ws = c.http.ws_connect.mock.return_value + ws.receive = AsyncMock() + ws.receive.mock.return_value.data = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + c.http._cookie_jar = [mock.MagicMock(), mock.MagicMock()] + c.http._cookie_jar[0].key = 'key' + c.http._cookie_jar[0].value = 'value' + c.http._cookie_jar[1].key = 'key2' + c.http._cookie_jar[1].value = 'value2' + c._read_loop_polling = AsyncMock() + c._read_loop_websocket = AsyncMock() + c._write_loop = AsyncMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + _run(c.connect('ws://foo', transports=['websocket'])) + c.http.ws_connect.mock.assert_called_once_with( + 'ws://foo/engine.io/?transport=websocket&EIO=4&t=123.456', + headers={}, + timeout=5, + ) + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_websocket_connection_with_cookie_header(self, _time): + c = async_client.AsyncClient() + c.http = mock.MagicMock(closed=False) + c.http.ws_connect = AsyncMock() + ws = c.http.ws_connect.mock.return_value + ws.receive = AsyncMock() + ws.receive.mock.return_value.data = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + c.http._cookie_jar = [] + c._read_loop_polling = AsyncMock() + c._read_loop_websocket = AsyncMock() + c._write_loop = AsyncMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + _run( + c.connect( + 'ws://foo', + headers={'Cookie': 'key=value; key2=value2; key3="value3="'}, + transports=['websocket'], + ) + ) + c.http.ws_connect.mock.assert_called_once_with( + 'ws://foo/engine.io/?transport=websocket&EIO=4&t=123.456', + headers={}, + timeout=5, + ) + c.http.cookie_jar.update_cookies.assert_called_once_with( + {'key': 'value', 'key2': 'value2', 'key3': '"value3="'} + ) + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_websocket_connection_with_cookies_and_headers(self, _time): + c = async_client.AsyncClient() + c.http = mock.MagicMock(closed=False) + c.http.ws_connect = AsyncMock() + ws = c.http.ws_connect.mock.return_value + ws.receive = AsyncMock() + ws.receive.mock.return_value.data = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + c.http._cookie_jar = [mock.MagicMock(), mock.MagicMock()] + c.http._cookie_jar[0].key = 'key' + c.http._cookie_jar[0].value = 'value' + c.http._cookie_jar[1].key = 'key2' + c.http._cookie_jar[1].value = 'value2' + c._read_loop_polling = AsyncMock() + c._read_loop_websocket = AsyncMock() + c._write_loop = AsyncMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + _run( + c.connect( + 'ws://foo', + headers={'Foo': 'Bar', 'Cookie': 'key3=value3'}, + transports=['websocket'], + ) + ) + c.http.ws_connect.mock.assert_called_once_with( + 'ws://foo/engine.io/?transport=websocket&EIO=4&t=123.456', + headers={'Foo': 'Bar'}, + timeout=5, + ) + c.http.cookie_jar.update_cookies.assert_called_once_with( + {'key3': 'value3'} + ) + + def test_websocket_upgrade_no_pong(self): + c = async_client.AsyncClient() + c.http = mock.MagicMock(closed=False) + c.http.ws_connect = AsyncMock() + ws = c.http.ws_connect.mock.return_value + ws.receive = AsyncMock() + ws.receive.mock.return_value.data = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + ws.send_str = AsyncMock() + c.sid = '123' + c.current_transport = 'polling' + c._read_loop_polling = AsyncMock() + c._read_loop_websocket = AsyncMock() + c._write_loop = AsyncMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + assert not _run(c.connect('ws://foo', transports=['websocket'])) + + c._read_loop_polling.mock.assert_not_called() + c._read_loop_websocket.mock.assert_not_called() + c._write_loop.mock.assert_not_called() + on_connect.assert_not_called() + assert c.transport() == 'polling' + ws.send_str.mock.assert_called_once_with('2probe') + + def test_websocket_upgrade_successful(self): + c = async_client.AsyncClient() + c.http = mock.MagicMock(closed=False) + c.http.ws_connect = AsyncMock() + ws = c.http.ws_connect.mock.return_value + ws.receive = AsyncMock() + ws.receive.mock.return_value.data = packet.Packet( + packet.PONG, 'probe' + ).encode() + ws.send_str = AsyncMock() + c.sid = '123' + c.base_url = 'http://foo' + c.current_transport = 'polling' + c._read_loop_polling = AsyncMock() + c._read_loop_websocket = AsyncMock() + c._write_loop = AsyncMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + assert _run(c.connect('ws://foo', transports=['websocket'])) + + c._read_loop_polling.mock.assert_not_called() + c._read_loop_websocket.mock.assert_called_once_with() + c._write_loop.mock.assert_called_once_with() + on_connect.assert_not_called() # was called by polling + assert c not in base_client.connected_clients # was added by polling + assert c.base_url == 'http://foo' # not changed + assert c.sid == '123' # not changed + assert c.transport() == 'websocket' + assert c.ws == ws + assert ws.send_str.mock.call_args_list[0] == (('2probe',),) # ping + assert ws.send_str.mock.call_args_list[1] == (('5',),) # upgrade + + def test_receive_unknown_packet(self): + c = async_client.AsyncClient() + _run(c._receive_packet(packet.Packet(encoded_packet='9'))) + # should be ignored + + def test_receive_noop_packet(self): + c = async_client.AsyncClient() + _run(c._receive_packet(packet.Packet(packet.NOOP))) + # should be ignored + + def test_receive_ping_packet(self): + c = async_client.AsyncClient() + c._send_packet = AsyncMock() + _run(c._receive_packet(packet.Packet(packet.PING))) + assert c._send_packet.mock.call_args_list[0][0][0].encode() == '3' + + def test_receive_message_packet(self): + c = async_client.AsyncClient() + c._trigger_event = AsyncMock() + _run(c._receive_packet(packet.Packet(packet.MESSAGE, {'foo': 'bar'}))) + c._trigger_event.mock.assert_called_once_with( + 'message', {'foo': 'bar'}, run_async=True + ) + + def test_receive_close_packet(self): + c = async_client.AsyncClient() + c.disconnect = AsyncMock() + _run(c._receive_packet(packet.Packet(packet.CLOSE))) + c.disconnect.mock.assert_called_once_with(abort=True) + + def test_send_packet_disconnected(self): + c = async_client.AsyncClient() + c.queue = c.create_queue() + c.state = 'disconnected' + _run(c._send_packet(packet.Packet(packet.NOOP))) + assert c.queue.empty() + + def test_send_packet(self): + c = async_client.AsyncClient() + c.queue = c.create_queue() + c.state = 'connected' + _run(c._send_packet(packet.Packet(packet.NOOP))) + assert not c.queue.empty() + pkt = _run(c.queue.get()) + assert pkt.packet_type == packet.NOOP + + def test_trigger_event_function(self): + result = [] + + def foo_handler(arg): + result.append('ok') + result.append(arg) + + c = async_client.AsyncClient() + c.on('message', handler=foo_handler) + _run(c._trigger_event('message', 'bar')) + assert result == ['ok', 'bar'] + + def test_trigger_event_coroutine(self): + result = [] + + async def foo_handler(arg): + result.append('ok') + result.append(arg) + + c = async_client.AsyncClient() + c.on('message', handler=foo_handler) + _run(c._trigger_event('message', 'bar')) + assert result == ['ok', 'bar'] + + def test_trigger_event_function_error(self): + def connect_handler(arg): + return 1 / 0 + + def foo_handler(arg): + return 1 / 0 + + c = async_client.AsyncClient() + c.on('connect', handler=connect_handler) + c.on('message', handler=foo_handler) + assert not _run(c._trigger_event('connect', '123')) + assert _run(c._trigger_event('message', 'bar')) is None + + def test_trigger_event_coroutine_error(self): + async def connect_handler(arg): + return 1 / 0 + + async def foo_handler(arg): + return 1 / 0 + + c = async_client.AsyncClient() + c.on('connect', handler=connect_handler) + c.on('message', handler=foo_handler) + assert not _run(c._trigger_event('connect', '123')) + assert _run(c._trigger_event('message', 'bar')) is None + + def test_trigger_event_function_async(self): + result = [] + + def foo_handler(arg): + result.append('ok') + result.append(arg) + + c = async_client.AsyncClient() + c.on('message', handler=foo_handler) + fut = _run(c._trigger_event('message', 'bar', run_async=True)) + asyncio.get_event_loop().run_until_complete(fut) + assert result == ['ok', 'bar'] + + def test_trigger_event_coroutine_async(self): + result = [] + + async def foo_handler(arg): + result.append('ok') + result.append(arg) + + c = async_client.AsyncClient() + c.on('message', handler=foo_handler) + fut = _run(c._trigger_event('message', 'bar', run_async=True)) + asyncio.get_event_loop().run_until_complete(fut) + assert result == ['ok', 'bar'] + + def test_trigger_event_function_async_error(self): + result = [] + + def foo_handler(arg): + result.append(arg) + return 1 / 0 + + c = async_client.AsyncClient() + c.on('message', handler=foo_handler) + fut = _run(c._trigger_event('message', 'bar', run_async=True)) + with pytest.raises(ZeroDivisionError): + asyncio.get_event_loop().run_until_complete(fut) + assert result == ['bar'] + + def test_trigger_event_coroutine_async_error(self): + result = [] + + async def foo_handler(arg): + result.append(arg) + return 1 / 0 + + c = async_client.AsyncClient() + c.on('message', handler=foo_handler) + fut = _run(c._trigger_event('message', 'bar', run_async=True)) + with pytest.raises(ZeroDivisionError): + asyncio.get_event_loop().run_until_complete(fut) + assert result == ['bar'] + + def test_trigger_unknown_event(self): + c = async_client.AsyncClient() + _run(c._trigger_event('connect', run_async=False)) + _run(c._trigger_event('message', 123, run_async=True)) + # should do nothing + + def test_read_loop_polling_disconnected(self): + c = async_client.AsyncClient() + c.state = 'disconnected' + c._trigger_event = AsyncMock() + c.write_loop_task = AsyncMock()() + _run(c._read_loop_polling()) + c._trigger_event.mock.assert_not_called() + # should not block + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_read_loop_polling_no_response(self, _time): + c = async_client.AsyncClient() + c.ping_interval = 25 + c.ping_timeout = 5 + c.state = 'connected' + c.base_url = 'http://foo' + c.queue = mock.MagicMock() + c.queue.put = AsyncMock() + c._send_request = AsyncMock(return_value=None) + c._trigger_event = AsyncMock() + c.write_loop_task = AsyncMock()() + _run(c._read_loop_polling()) + assert c.state == 'disconnected' + c.queue.put.mock.assert_called_once_with(None) + c._send_request.mock.assert_called_once_with( + 'GET', 'http://foo&t=123.456', timeout=30 + ) + c._trigger_event.mock.assert_called_once_with( + 'disconnect', run_async=False + ) + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_read_loop_polling_bad_status(self, _time): + c = async_client.AsyncClient() + c.ping_interval = 25 + c.ping_timeout = 5 + c.state = 'connected' + c.base_url = 'http://foo' + c.queue = mock.MagicMock() + c.queue.put = AsyncMock() + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 400 + c.write_loop_task = AsyncMock()() + _run(c._read_loop_polling()) + assert c.state == 'disconnected' + c.queue.put.mock.assert_called_once_with(None) + c._send_request.mock.assert_called_once_with( + 'GET', 'http://foo&t=123.456', timeout=30 + ) + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_read_loop_polling_bad_packet(self, _time): + c = async_client.AsyncClient() + c.ping_interval = 25 + c.ping_timeout = 60 + c.state = 'connected' + c.base_url = 'http://foo' + c.queue = mock.MagicMock() + c.queue.put = AsyncMock() + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 200 + c._send_request.mock.return_value.read = AsyncMock(return_value=b'foo') + c.write_loop_task = AsyncMock()() + _run(c._read_loop_polling()) + assert c.state == 'disconnected' + c.queue.put.mock.assert_called_once_with(None) + c._send_request.mock.assert_called_once_with( + 'GET', 'http://foo&t=123.456', timeout=65 + ) + + def test_read_loop_polling(self): + c = async_client.AsyncClient() + c.ping_interval = 25 + c.ping_timeout = 5 + c.state = 'connected' + c.base_url = 'http://foo' + c.queue = mock.MagicMock() + c.queue.put = AsyncMock() + c._send_request = AsyncMock() + c._send_request.mock.side_effect = [ + mock.MagicMock( + status=200, + read=AsyncMock( + return_value=payload.Payload( + packets=[ + packet.Packet(packet.PING), + packet.Packet(packet.NOOP), + ] + ).encode().encode('utf-8') + ), + ), + None, + ] + c.write_loop_task = AsyncMock()() + c._receive_packet = AsyncMock() + _run(c._read_loop_polling()) + assert c.state == 'disconnected' + c.queue.put.mock.assert_called_once_with(None) + assert c._send_request.mock.call_count == 2 + assert c._receive_packet.mock.call_count == 2 + assert c._receive_packet.mock.call_args_list[0][0][0].encode() == '2' + assert c._receive_packet.mock.call_args_list[1][0][0].encode() == '6' + + def test_read_loop_websocket_disconnected(self): + c = async_client.AsyncClient() + c.state = 'disconnected' + c.write_loop_task = AsyncMock()() + _run(c._read_loop_websocket()) + # should not block + + def test_read_loop_websocket_timeout(self): + c = async_client.AsyncClient() + c.ping_interval = 1 + c.ping_timeout = 2 + c.base_url = 'ws://foo' + c.state = 'connected' + c.queue = mock.MagicMock() + c.queue.put = AsyncMock() + c.ws = mock.MagicMock() + c.ws.receive = AsyncMock(side_effect=asyncio.TimeoutError()) + c.write_loop_task = AsyncMock()() + _run(c._read_loop_websocket()) + assert c.state == 'disconnected' + c.queue.put.mock.assert_called_once_with(None) + + def test_read_loop_websocket_no_response(self): + c = async_client.AsyncClient() + c.ping_interval = 1 + c.ping_timeout = 2 + c.base_url = 'ws://foo' + c.state = 'connected' + c.queue = mock.MagicMock() + c.queue.put = AsyncMock() + c.ws = mock.MagicMock() + c.ws.receive = AsyncMock( + side_effect=aiohttp.client_exceptions.ServerDisconnectedError() + ) + c.write_loop_task = AsyncMock()() + _run(c._read_loop_websocket()) + assert c.state == 'disconnected' + c.queue.put.mock.assert_called_once_with(None) + + def test_read_loop_websocket_unexpected_error(self): + c = async_client.AsyncClient() + c.ping_interval = 1 + c.ping_timeout = 2 + c.base_url = 'ws://foo' + c.state = 'connected' + c.queue = mock.MagicMock() + c.queue.put = AsyncMock() + c.ws = mock.MagicMock() + c.ws.receive = AsyncMock(side_effect=ValueError) + c.write_loop_task = AsyncMock()() + _run(c._read_loop_websocket()) + assert c.state == 'disconnected' + c.queue.put.mock.assert_called_once_with(None) + + def test_read_loop_websocket(self): + c = async_client.AsyncClient() + c.ping_interval = 1 + c.ping_timeout = 2 + c.base_url = 'ws://foo' + c.state = 'connected' + c.queue = mock.MagicMock() + c.queue.put = AsyncMock() + c.ws = mock.MagicMock() + c.ws.receive = AsyncMock( + side_effect=[ + mock.MagicMock(data=packet.Packet(packet.PING).encode()), + ValueError, + ] + ) + c.write_loop_task = AsyncMock()() + c._receive_packet = AsyncMock() + _run(c._read_loop_websocket()) + assert c.state == 'disconnected' + assert c._receive_packet.mock.call_args_list[0][0][0].encode() == '2' + c.queue.put.mock.assert_called_once_with(None) + + def test_write_loop_disconnected(self): + c = async_client.AsyncClient() + c.state = 'disconnected' + _run(c._write_loop()) + # should not block + + def test_write_loop_no_packets(self): + c = async_client.AsyncClient() + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.queue = mock.MagicMock() + c.queue.get = AsyncMock(return_value=None) + _run(c._write_loop()) + c.queue.task_done.assert_called_once_with() + c.queue.get.mock.assert_called_once_with() + + def test_write_loop_empty_queue(self): + c = async_client.AsyncClient() + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get = AsyncMock(side_effect=RuntimeError) + _run(c._write_loop()) + c.queue.get.mock.assert_called_once_with() + + def test_write_loop_polling_one_packet(self): + c = async_client.AsyncClient() + c.base_url = 'http://foo' + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get = AsyncMock( + side_effect=[ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + RuntimeError, + ] + ) + c.queue.get_nowait = mock.MagicMock(side_effect=RuntimeError) + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 200 + _run(c._write_loop()) + assert c.queue.task_done.call_count == 1 + p = payload.Payload( + packets=[packet.Packet(packet.MESSAGE, {'foo': 'bar'})] + ) + c._send_request.mock.assert_called_once_with( + 'POST', + 'http://foo', + body=p.encode(), + headers={'Content-Type': 'text/plain'}, + timeout=5, + ) + + def test_write_loop_polling_three_packets(self): + c = async_client.AsyncClient() + c.base_url = 'http://foo' + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get = AsyncMock( + side_effect=[ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + RuntimeError, + ] + ) + c.queue.get_nowait = mock.MagicMock( + side_effect=[ + packet.Packet(packet.PING), + packet.Packet(packet.NOOP), + RuntimeError, + ] + ) + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 200 + _run(c._write_loop()) + assert c.queue.task_done.call_count == 3 + p = payload.Payload( + packets=[ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + packet.Packet(packet.PING), + packet.Packet(packet.NOOP), + ] + ) + c._send_request.mock.assert_called_once_with( + 'POST', + 'http://foo', + body=p.encode(), + headers={'Content-Type': 'text/plain'}, + timeout=5, + ) + + def test_write_loop_polling_two_packets_done(self): + c = async_client.AsyncClient() + c.base_url = 'http://foo' + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get = AsyncMock( + side_effect=[ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + RuntimeError, + ] + ) + c.queue.get_nowait = mock.MagicMock( + side_effect=[packet.Packet(packet.PING), None] + ) + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 200 + _run(c._write_loop()) + assert c.queue.task_done.call_count == 3 + p = payload.Payload( + packets=[ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + packet.Packet(packet.PING), + ] + ) + c._send_request.mock.assert_called_once_with( + 'POST', + 'http://foo', + body=p.encode(), + headers={'Content-Type': 'text/plain'}, + timeout=5, + ) + assert c.state == 'connected' + + def test_write_loop_polling_bad_connection(self): + c = async_client.AsyncClient() + c.base_url = 'http://foo' + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get = AsyncMock( + side_effect=[packet.Packet(packet.MESSAGE, {'foo': 'bar'})] + ) + c.queue.get_nowait = mock.MagicMock(side_effect=[RuntimeError]) + c._send_request = AsyncMock(return_value=None) + _run(c._write_loop()) + assert c.queue.task_done.call_count == 1 + p = payload.Payload( + packets=[packet.Packet(packet.MESSAGE, {'foo': 'bar'})] + ) + c._send_request.mock.assert_called_once_with( + 'POST', + 'http://foo', + body=p.encode(), + headers={'Content-Type': 'text/plain'}, + timeout=5, + ) + assert c.state == 'connected' + + def test_write_loop_polling_bad_status(self): + c = async_client.AsyncClient() + c.base_url = 'http://foo' + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get = AsyncMock( + side_effect=[packet.Packet(packet.MESSAGE, {'foo': 'bar'})] + ) + c.queue.get_nowait = mock.MagicMock(side_effect=[RuntimeError]) + c._send_request = AsyncMock() + c._send_request.mock.return_value.status = 500 + _run(c._write_loop()) + assert c.queue.task_done.call_count == 1 + p = payload.Payload( + packets=[packet.Packet(packet.MESSAGE, {'foo': 'bar'})] + ) + c._send_request.mock.assert_called_once_with( + 'POST', + 'http://foo', + body=p.encode(), + headers={'Content-Type': 'text/plain'}, + timeout=5, + ) + assert c.state == 'connected' + assert c.write_loop_task is None + + def test_write_loop_websocket_one_packet(self): + c = async_client.AsyncClient() + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'websocket' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get = AsyncMock( + side_effect=[ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + RuntimeError, + ] + ) + c.queue.get_nowait = mock.MagicMock(side_effect=[RuntimeError]) + c.ws = mock.MagicMock() + c.ws.send_str = AsyncMock() + _run(c._write_loop()) + assert c.queue.task_done.call_count == 1 + assert c.ws.send_str.mock.call_count == 1 + c.ws.send_str.mock.assert_called_once_with('4{"foo":"bar"}') + + def test_write_loop_websocket_three_packets(self): + c = async_client.AsyncClient() + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'websocket' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get = AsyncMock( + side_effect=[ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + RuntimeError, + ] + ) + c.queue.get_nowait = mock.MagicMock( + side_effect=[ + packet.Packet(packet.PING), + packet.Packet(packet.NOOP), + RuntimeError, + ] + ) + c.ws = mock.MagicMock() + c.ws.send_str = AsyncMock() + _run(c._write_loop()) + assert c.queue.task_done.call_count == 3 + assert c.ws.send_str.mock.call_count == 3 + assert c.ws.send_str.mock.call_args_list[0][0][0] == '4{"foo":"bar"}' + assert c.ws.send_str.mock.call_args_list[1][0][0] == '2' + assert c.ws.send_str.mock.call_args_list[2][0][0] == '6' + + def test_write_loop_websocket_one_packet_binary(self): + c = async_client.AsyncClient() + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'websocket' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get = AsyncMock( + side_effect=[packet.Packet(packet.MESSAGE, b'foo'), RuntimeError] + ) + c.queue.get_nowait = mock.MagicMock(side_effect=[RuntimeError]) + c.ws = mock.MagicMock() + c.ws.send_bytes = AsyncMock() + _run(c._write_loop()) + assert c.queue.task_done.call_count == 1 + assert c.ws.send_bytes.mock.call_count == 1 + c.ws.send_bytes.mock.assert_called_once_with(b'foo') + + def test_write_loop_websocket_bad_connection(self): + c = async_client.AsyncClient() + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'websocket' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get = AsyncMock( + side_effect=[ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + RuntimeError, + ] + ) + c.queue.get_nowait = mock.MagicMock(side_effect=[RuntimeError]) + c.ws = mock.MagicMock() + c.ws.send_str = AsyncMock( + side_effect=aiohttp.client_exceptions.ServerDisconnectedError() + ) + _run(c._write_loop()) + assert c.state == 'connected' + + @mock.patch('engineio.base_client.original_signal_handler') + def test_signal_handler(self, original_handler): + clients = [mock.MagicMock(), mock.MagicMock()] + base_client.connected_clients = clients[:] + base_client.connected_clients[0].is_asyncio_based.return_value = False + base_client.connected_clients[1].is_asyncio_based.return_value = True + + async def test(): + async_client.async_signal_handler() + + asyncio.get_event_loop().run_until_complete(test()) + clients[0].disconnect.assert_not_called() + clients[1].disconnect.assert_called_once_with() diff --git a/tests/async/test_sanic.py b/tests/async/test_sanic.py new file mode 100644 index 0000000..a1aa7de --- /dev/null +++ b/tests/async/test_sanic.py @@ -0,0 +1 @@ +from engineio.async_drivers import sanic as async_sanic # noqa: F401 diff --git a/tests/async/test_server.py b/tests/async/test_server.py new file mode 100644 index 0000000..d967577 --- /dev/null +++ b/tests/async/test_server.py @@ -0,0 +1,1164 @@ +import asyncio +import gzip +import io +import logging +import unittest +from unittest import mock +import zlib + +import pytest + +from engineio import async_server +from engineio.async_drivers import aiohttp as async_aiohttp +from engineio import exceptions +from engineio import json +from engineio import packet +from engineio import payload + + +def AsyncMock(*args, **kwargs): + """Return a mock asynchronous function.""" + m = mock.MagicMock(*args, **kwargs) + + async def mock_coro(*args, **kwargs): + return m(*args, **kwargs) + + mock_coro.mock = m + return mock_coro + + +def _run(coro): + """Run the given coroutine.""" + return asyncio.get_event_loop().run_until_complete(coro) + + +class TestAsyncServer(unittest.TestCase): + @staticmethod + def get_async_mock(environ={'REQUEST_METHOD': 'GET', 'QUERY_STRING': ''}): + if environ.get('QUERY_STRING'): + if 'EIO=' not in environ['QUERY_STRING']: + environ['QUERY_STRING'] = 'EIO=4&' + environ['QUERY_STRING'] + else: + environ['QUERY_STRING'] = 'EIO=4' + a = mock.MagicMock() + a._async = { + 'asyncio': True, + 'create_route': mock.MagicMock(), + 'translate_request': mock.MagicMock(), + 'make_response': mock.MagicMock(), + 'websocket': 'w', + } + a._async['translate_request'].return_value = environ + a._async['make_response'].return_value = 'response' + return a + + def _get_mock_socket(self): + mock_socket = mock.MagicMock() + mock_socket.connected = False + mock_socket.closed = False + mock_socket.closing = False + mock_socket.upgraded = False + mock_socket.send = AsyncMock() + mock_socket.handle_get_request = AsyncMock() + mock_socket.handle_post_request = AsyncMock() + mock_socket.check_ping_timeout = AsyncMock() + mock_socket.close = AsyncMock() + mock_socket.session = {} + return mock_socket + + @classmethod + def setUpClass(cls): + async_server.AsyncServer._default_monitor_clients = False + + @classmethod + def tearDownClass(cls): + async_server.AsyncServer._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 = async_server.AsyncServer() + assert s.is_asyncio_based() + + def test_async_modes(self): + s = async_server.AsyncServer() + assert s.async_modes() == ['aiohttp', 'sanic', 'tornado', 'asgi'] + + def test_async_mode_aiohttp(self): + s = async_server.AsyncServer(async_mode='aiohttp') + assert s.async_mode == 'aiohttp' + assert s._async['asyncio'] + assert s._async['create_route'] == async_aiohttp.create_route + assert s._async['translate_request'] == async_aiohttp.translate_request + assert s._async['make_response'] == async_aiohttp.make_response + assert s._async['websocket'].__name__ == 'WebSocket' + + @mock.patch('importlib.import_module') + def test_async_mode_auto_aiohttp(self, import_module): + import_module.side_effect = [self.get_async_mock()] + s = async_server.AsyncServer() + assert s.async_mode == 'aiohttp' + + def test_async_modes_wsgi(self): + with pytest.raises(ValueError): + async_server.AsyncServer(async_mode='eventlet') + with pytest.raises(ValueError): + async_server.AsyncServer(async_mode='gevent') + with pytest.raises(ValueError): + async_server.AsyncServer(async_mode='gevent_uwsgi') + with pytest.raises(ValueError): + async_server.AsyncServer(async_mode='threading') + + @mock.patch('importlib.import_module') + def test_attach(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer() + s.attach('app', engineio_path='abc') + a._async['create_route'].assert_called_with('app', s, '/abc/') + s.attach('app', engineio_path='/def/') + a._async['create_route'].assert_called_with('app', s, '/def/') + s.attach('app', engineio_path='/ghi') + a._async['create_route'].assert_called_with('app', s, '/ghi/') + s.attach('app', engineio_path='jkl/') + a._async['create_route'].assert_called_with('app', s, '/jkl/') + + def test_session(self): + s = async_server.AsyncServer() + s.sockets['foo'] = self._get_mock_socket() + + async def _func(): + async with s.session('foo') as session: + await s.sleep(0) + session['username'] = 'bar' + assert await s.get_session('foo') == {'username': 'bar'} + + _run(_func()) + + def test_disconnect(self): + s = async_server.AsyncServer() + s.sockets['foo'] = mock_socket = self._get_mock_socket() + _run(s.disconnect('foo')) + assert mock_socket.close.mock.call_count == 1 + mock_socket.close.mock.assert_called_once_with() + assert 'foo' not in s.sockets + + def test_disconnect_all(self): + s = async_server.AsyncServer() + s.sockets['foo'] = mock_foo = self._get_mock_socket() + s.sockets['bar'] = mock_bar = self._get_mock_socket() + _run(s.disconnect()) + assert mock_foo.close.mock.call_count == 1 + assert mock_bar.close.mock.call_count == 1 + mock_foo.close.mock.assert_called_once_with() + mock_bar.close.mock.assert_called_once_with() + assert 'foo' not in s.sockets + assert 'bar' not in s.sockets + + @mock.patch('importlib.import_module') + def test_jsonp_not_supported(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'j=abc'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + response = _run(s.handle_request('request')) + assert response == 'response' + a._async['translate_request'].assert_called_once_with('request') + assert a._async['make_response'].call_count == 1 + assert a._async['make_response'].call_args[0][0] == '400 BAD REQUEST' + + @mock.patch('importlib.import_module') + def test_jsonp_index(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'j=233'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + response = _run(s.handle_request('request')) + assert response == 'response' + a._async['translate_request'].assert_called_once_with('request') + assert a._async['make_response'].call_count == 1 + assert a._async['make_response'].call_args[0][0] == '200 OK' + assert ( + a._async['make_response'] + .call_args[0][2] + .startswith(b'___eio[233]("') + ) + assert a._async['make_response'].call_args[0][2].endswith(b'");') + + @mock.patch('importlib.import_module') + def test_connect(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer() + _run(s.handle_request('request')) + assert len(s.sockets) == 1 + assert a._async['make_response'].call_count == 1 + assert a._async['make_response'].call_args[0][0] == '200 OK' + assert ('Content-Type', 'text/plain; charset=UTF-8') in a._async[ + 'make_response' + ].call_args[0][1] + packets = payload.Payload( + encoded_payload=a._async['make_response'].call_args[0][2].decode( + 'utf-8')).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 + + @mock.patch('importlib.import_module') + def test_connect_async_request_response_handlers(self, import_module): + a = self.get_async_mock() + a._async['translate_request'] = AsyncMock( + return_value=a._async['translate_request'].return_value + ) + a._async['make_response'] = AsyncMock( + return_value=a._async['make_response'].return_value + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + _run(s.handle_request('request')) + assert len(s.sockets) == 1 + assert a._async['make_response'].mock.call_count == 1 + assert a._async['make_response'].mock.call_args[0][0] == '200 OK' + assert ('Content-Type', 'text/plain; charset=UTF-8') in a._async[ + 'make_response' + ].mock.call_args[0][1] + packets = payload.Payload( + encoded_payload=a._async['make_response'].mock.call_args[0][ + 2].decode('utf-8')).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 + + @mock.patch('importlib.import_module') + def test_connect_no_upgrades(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer(allow_upgrades=False) + _run(s.handle_request('request')) + packets = payload.Payload( + encoded_payload=a._async['make_response'].call_args[0][2].decode( + 'utf-8')).packets + assert packets[0].data['upgrades'] == [] + + @mock.patch('importlib.import_module') + def test_connect_bad_eio_version(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=1'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + _run(s.handle_request('request')) + assert a._async['make_response'].call_count == 1 + assert a._async['make_response'].call_args[0][0] == '400 BAD REQUEST' + assert b'unsupported version' in \ + a._async['make_response'].call_args[0][2] + + @mock.patch('importlib.import_module') + def test_connect_custom_ping_times(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer(ping_timeout=123, ping_interval=456) + _run(s.handle_request('request')) + packets = payload.Payload( + encoded_payload=a._async['make_response'].call_args[0][2].decode( + 'utf-8')).packets + assert packets[0].data['pingTimeout'] == 123000 + assert packets[0].data['pingInterval'] == 456000 + + @mock.patch('importlib.import_module') + @mock.patch('engineio.async_server.async_socket.AsyncSocket') + def test_connect_bad_poll(self, AsyncSocket, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + AsyncSocket.return_value = self._get_mock_socket() + AsyncSocket.return_value.poll.side_effect = [exceptions.QueueEmpty] + s = async_server.AsyncServer() + _run(s.handle_request('request')) + assert a._async['make_response'].call_count == 1 + assert a._async['make_response'].call_args[0][0] == '400 BAD REQUEST' + + @mock.patch('importlib.import_module') + @mock.patch('engineio.async_server.async_socket.AsyncSocket') + def test_connect_transport_websocket(self, AsyncSocket, import_module): + a = self.get_async_mock( + { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'transport=websocket', + 'HTTP_UPGRADE': 'websocket', + } + ) + import_module.side_effect = [a] + AsyncSocket.return_value = self._get_mock_socket() + s = async_server.AsyncServer() + s.generate_id = mock.MagicMock(return_value='123') + # force socket to stay open, so that we can check it later + AsyncSocket().closed = False + _run(s.handle_request('request')) + assert ( + s.sockets['123'].send.mock.call_args[0][0].packet_type + == packet.OPEN + ) + + @mock.patch('importlib.import_module') + @mock.patch('engineio.async_server.async_socket.AsyncSocket') + def test_http_upgrade_case_insensitive(self, AsyncSocket, import_module): + a = self.get_async_mock( + { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'transport=websocket', + 'HTTP_UPGRADE': 'WebSocket', + } + ) + import_module.side_effect = [a] + AsyncSocket.return_value = self._get_mock_socket() + s = async_server.AsyncServer() + s.generate_id = mock.MagicMock(return_value='123') + # force socket to stay open, so that we can check it later + AsyncSocket().closed = False + _run(s.handle_request('request')) + assert ( + s.sockets['123'].send.mock.call_args[0][0].packet_type + == packet.OPEN + ) + + @mock.patch('importlib.import_module') + @mock.patch('engineio.async_server.async_socket.AsyncSocket') + def test_connect_transport_websocket_closed( + self, AsyncSocket, import_module): + a = self.get_async_mock( + { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'transport=websocket', + 'HTTP_UPGRADE': 'websocket' + } + ) + import_module.side_effect = [a] + AsyncSocket.return_value = self._get_mock_socket() + s = async_server.AsyncServer() + s.generate_id = mock.MagicMock(return_value='123') + + # this mock handler just closes the socket, as it would happen on a + # real websocket exchange + async def mock_handle(environ): + s.sockets['123'].closed = True + + AsyncSocket().handle_get_request = mock_handle + _run(s.handle_request('request')) + assert '123' not in s.sockets # socket should close on its own + + @mock.patch('importlib.import_module') + def test_connect_transport_invalid(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'transport=foo'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + _run(s.handle_request('request')) + assert a._async['make_response'].call_count == 1 + assert a._async['make_response'].call_args[0][0] == '400 BAD REQUEST' + + @mock.patch('importlib.import_module') + def test_connect_transport_websocket_without_upgrade(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'transport=websocket'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + _run(s.handle_request('request')) + assert a._async['make_response'].call_count == 1 + assert a._async['make_response'].call_args[0][0] == '400 BAD REQUEST' + + @mock.patch('importlib.import_module') + def test_connect_cors_headers(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer() + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + headers = a._async['make_response'].call_args[0][1] + assert ('Access-Control-Allow-Credentials', 'true') in headers + + @mock.patch('importlib.import_module') + def test_connect_cors_allowed_origin(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': '', 'HTTP_ORIGIN': 'b'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(cors_allowed_origins=['a', 'b']) + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + headers = a._async['make_response'].call_args[0][1] + assert ('Access-Control-Allow-Origin', 'b') in headers + + @mock.patch('importlib.import_module') + def test_connect_cors_allowed_origin_with_callable(self, import_module): + def cors(origin): + return origin == 'a' + + environ = { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': '', + 'HTTP_ORIGIN': 'a' + } + a = self.get_async_mock(environ) + import_module.side_effect = [a] + + s = async_server.AsyncServer(cors_allowed_origins=cors) + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + headers = a._async['make_response'].call_args[0][1] + assert ('Access-Control-Allow-Origin', 'a') in headers + + environ['HTTP_ORIGIN'] = 'b' + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '400 BAD REQUEST' + + @mock.patch('importlib.import_module') + def test_connect_cors_not_allowed_origin(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': '', 'HTTP_ORIGIN': 'c'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(cors_allowed_origins=['a', 'b']) + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '400 BAD REQUEST' + headers = a._async['make_response'].call_args[0][1] + assert ('Access-Control-Allow-Origin', 'c') not in headers + assert ('Access-Control-Allow-Origin', '*') not in headers + + @mock.patch('importlib.import_module') + def test_connect_cors_not_allowed_origin_async_response( + self, import_module + ): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': '', 'HTTP_ORIGIN': 'c'} + ) + a._async['make_response'] = AsyncMock( + return_value=a._async['make_response'].return_value + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(cors_allowed_origins=['a', 'b']) + _run(s.handle_request('request')) + assert ( + a._async['make_response'].mock.call_args[0][0] == '400 BAD REQUEST' + ) + headers = a._async['make_response'].mock.call_args[0][1] + assert ('Access-Control-Allow-Origin', 'c') not in headers + assert ('Access-Control-Allow-Origin', '*') not in headers + + @mock.patch('importlib.import_module') + def test_connect_cors_all_origins(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': '', 'HTTP_ORIGIN': 'foo'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(cors_allowed_origins='*') + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + headers = a._async['make_response'].call_args[0][1] + assert ('Access-Control-Allow-Origin', 'foo') in headers + assert ('Access-Control-Allow-Credentials', 'true') in headers + + @mock.patch('importlib.import_module') + def test_connect_cors_one_origin(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': '', 'HTTP_ORIGIN': 'a'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(cors_allowed_origins='a') + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + headers = a._async['make_response'].call_args[0][1] + assert ('Access-Control-Allow-Origin', 'a') in headers + assert ('Access-Control-Allow-Credentials', 'true') in headers + + @mock.patch('importlib.import_module') + def test_connect_cors_one_origin_not_allowed(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': '', 'HTTP_ORIGIN': 'b'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(cors_allowed_origins='a') + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '400 BAD REQUEST' + headers = a._async['make_response'].call_args[0][1] + assert ('Access-Control-Allow-Origin', 'b') not in headers + assert ('Access-Control-Allow-Origin', '*') not in headers + + @mock.patch('importlib.import_module') + def test_connect_cors_headers_default_origin(self, import_module): + a = self.get_async_mock( + { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': '', + 'wsgi.url_scheme': 'http', + 'HTTP_HOST': 'foo', + 'HTTP_ORIGIN': 'http://foo', + } + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + headers = a._async['make_response'].call_args[0][1] + assert ('Access-Control-Allow-Origin', 'http://foo') in headers + + @mock.patch('importlib.import_module') + def test_connect_cors_no_credentials(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer(cors_credentials=False) + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + headers = a._async['make_response'].call_args[0][1] + assert ('Access-Control-Allow-Credentials', 'true') not in headers + + @mock.patch('importlib.import_module') + def test_connect_cors_options(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'OPTIONS', 'QUERY_STRING': ''} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(cors_credentials=False) + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + headers = a._async['make_response'].call_args[0][1] + assert ( + 'Access-Control-Allow-Methods', + 'OPTIONS, GET, POST', + ) in headers + + @mock.patch('importlib.import_module') + def test_connect_cors_disabled(self, import_module): + a = self.get_async_mock( + { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': '', + 'HTTP_ORIGIN': 'http://foo', + } + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(cors_allowed_origins=[]) + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + headers = a._async['make_response'].call_args[0][1] + for header in headers: + assert not header[0].startswith('Access-Control-') + + @mock.patch('importlib.import_module') + def test_connect_cors_default_no_origin(self, import_module): + a = self.get_async_mock({'REQUEST_METHOD': 'GET', 'QUERY_STRING': ''}) + import_module.side_effect = [a] + s = async_server.AsyncServer(cors_allowed_origins=[]) + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + headers = a._async['make_response'].call_args[0][1] + for header in headers: + assert header[0] != 'Access-Control-Allow-Origin' + + @mock.patch('importlib.import_module') + def test_connect_cors_all_no_origin(self, import_module): + a = self.get_async_mock({'REQUEST_METHOD': 'GET', 'QUERY_STRING': ''}) + import_module.side_effect = [a] + s = async_server.AsyncServer(cors_allowed_origins='*') + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + headers = a._async['make_response'].call_args[0][1] + for header in headers: + assert header[0] != 'Access-Control-Allow-Origin' + + @mock.patch('importlib.import_module') + def test_connect_cors_disabled_no_origin(self, import_module): + a = self.get_async_mock({'REQUEST_METHOD': 'GET', 'QUERY_STRING': ''}) + import_module.side_effect = [a] + s = async_server.AsyncServer(cors_allowed_origins=[]) + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + headers = a._async['make_response'].call_args[0][1] + for header in headers: + assert header[0] != 'Access-Control-Allow-Origin' + + @mock.patch('importlib.import_module') + def test_connect_event(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer() + s.generate_id = mock.MagicMock(return_value='123') + + def mock_connect(sid, environ): + return True + + s.on('connect', handler=mock_connect) + _run(s.handle_request('request')) + assert len(s.sockets) == 1 + + @mock.patch('importlib.import_module') + def test_connect_event_rejects(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer() + s.generate_id = mock.MagicMock(return_value='123') + + def mock_connect(sid, environ): + return False + + s.on('connect')(mock_connect) + _run(s.handle_request('request')) + assert len(s.sockets) == 0 + assert a._async['make_response'].call_args[0][0] == '401 UNAUTHORIZED' + assert a._async['make_response'].call_args[0][2] == b'"Unauthorized"' + + @mock.patch('importlib.import_module') + def test_connect_event_rejects_with_message(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer() + s.generate_id = mock.MagicMock(return_value='123') + + def mock_connect(sid, environ): + return {'not': 'allowed'} + + s.on('connect')(mock_connect) + _run(s.handle_request('request')) + assert len(s.sockets) == 0 + assert a._async['make_response'].call_args[0][0] == '401 UNAUTHORIZED' + assert ( + a._async['make_response'].call_args[0][2] == b'{"not": "allowed"}' + ) + + @mock.patch('importlib.import_module') + def test_method_not_found(self, import_module): + a = self.get_async_mock({'REQUEST_METHOD': 'PUT', 'QUERY_STRING': ''}) + import_module.side_effect = [a] + s = async_server.AsyncServer() + _run(s.handle_request('request')) + assert len(s.sockets) == 0 + assert ( + a._async['make_response'].call_args[0][0] == '405 METHOD NOT FOUND' + ) + + @mock.patch('importlib.import_module') + def test_get_request_with_bad_sid(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'sid=foo'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + _run(s.handle_request('request')) + assert len(s.sockets) == 0 + assert a._async['make_response'].call_args[0][0] == '400 BAD REQUEST' + + @mock.patch('importlib.import_module') + def test_post_request_with_bad_sid(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'POST', 'QUERY_STRING': 'sid=foo'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + _run(s.handle_request('request')) + assert len(s.sockets) == 0 + assert a._async['make_response'].call_args[0][0] == '400 BAD REQUEST' + + @mock.patch('importlib.import_module') + def test_send(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer() + s.sockets['foo'] = mock_socket = self._get_mock_socket() + _run(s.send('foo', 'hello')) + assert mock_socket.send.mock.call_count == 1 + assert ( + mock_socket.send.mock.call_args[0][0].packet_type == packet.MESSAGE + ) + assert mock_socket.send.mock.call_args[0][0].data == 'hello' + + @mock.patch('importlib.import_module') + def test_send_unknown_socket(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer() + # just ensure no exceptions are raised + _run(s.send('foo', 'hello')) + + @mock.patch('importlib.import_module') + def test_get_request(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'sid=foo'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + s.sockets['foo'] = mock_socket = self._get_mock_socket() + mock_socket.handle_get_request.mock.return_value = [ + packet.Packet(packet.MESSAGE, data='hello') + ] + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + packets = payload.Payload( + encoded_payload=a._async['make_response'].call_args[0][2].decode( + 'utf-8') + ).packets + assert len(packets) == 1 + assert packets[0].packet_type == packet.MESSAGE + + @mock.patch('importlib.import_module') + def test_get_request_custom_response(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'sid=foo'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + s.sockets['foo'] = mock_socket = self._get_mock_socket() + mock_socket.handle_get_request.mock.return_value = 'resp' + r = _run(s.handle_request('request')) + assert r == 'resp' + + @mock.patch('importlib.import_module') + def test_get_request_closes_socket(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'sid=foo'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + s.sockets['foo'] = mock_socket = self._get_mock_socket() + + async def mock_get_request(*args, **kwargs): + mock_socket.closed = True + return 'resp' + + mock_socket.handle_get_request = mock_get_request + r = _run(s.handle_request('request')) + assert r == 'resp' + assert 'foo' not in s.sockets + + @mock.patch('importlib.import_module') + def test_get_request_error(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'sid=foo'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + s.sockets['foo'] = mock_socket = self._get_mock_socket() + + async def mock_get_request(*args, **kwargs): + raise exceptions.QueueEmpty() + + mock_socket.handle_get_request = mock_get_request + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '400 BAD REQUEST' + assert len(s.sockets) == 0 + + @mock.patch('importlib.import_module') + def test_post_request(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'POST', 'QUERY_STRING': 'sid=foo'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + s.sockets['foo'] = self._get_mock_socket() + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '200 OK' + + @mock.patch('importlib.import_module') + def test_post_request_error(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'POST', 'QUERY_STRING': 'sid=foo'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer() + s.sockets['foo'] = mock_socket = self._get_mock_socket() + + async def mock_post_request(*args, **kwargs): + raise exceptions.ContentTooLongError() + + mock_socket.handle_post_request = mock_post_request + _run(s.handle_request('request')) + assert a._async['make_response'].call_args[0][0] == '400 BAD REQUEST' + + @staticmethod + def _gzip_decompress(b): + bytesio = io.BytesIO(b) + with gzip.GzipFile(fileobj=bytesio, mode='r') as gz: + return gz.read() + + @mock.patch('importlib.import_module') + def test_gzip_compression(self, import_module): + a = self.get_async_mock( + { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'sid=foo', + 'HTTP_ACCEPT_ENCODING': 'gzip,deflate', + } + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(compression_threshold=0) + s.sockets['foo'] = mock_socket = self._get_mock_socket() + mock_socket.handle_get_request.mock.return_value = [ + packet.Packet(packet.MESSAGE, data='hello') + ] + _run(s.handle_request('request')) + headers = a._async['make_response'].call_args[0][1] + assert ('Content-Encoding', 'gzip') in headers + self._gzip_decompress(a._async['make_response'].call_args[0][2]) + + @mock.patch('importlib.import_module') + def test_deflate_compression(self, import_module): + a = self.get_async_mock( + { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'sid=foo', + 'HTTP_ACCEPT_ENCODING': 'deflate;q=1,gzip', + } + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(compression_threshold=0) + s.sockets['foo'] = mock_socket = self._get_mock_socket() + mock_socket.handle_get_request.mock.return_value = [ + packet.Packet(packet.MESSAGE, data='hello') + ] + _run(s.handle_request('request')) + headers = a._async['make_response'].call_args[0][1] + assert ('Content-Encoding', 'deflate') in headers + zlib.decompress(a._async['make_response'].call_args[0][2]) + + @mock.patch('importlib.import_module') + def test_gzip_compression_threshold(self, import_module): + a = self.get_async_mock( + { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'sid=foo', + 'HTTP_ACCEPT_ENCODING': 'gzip', + } + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(compression_threshold=1000) + s.sockets['foo'] = mock_socket = self._get_mock_socket() + mock_socket.handle_get_request.mock.return_value = [ + packet.Packet(packet.MESSAGE, data='hello') + ] + _run(s.handle_request('request')) + headers = a._async['make_response'].call_args[0][1] + for header, value in headers: + assert header != 'Content-Encoding' + with pytest.raises(IOError): + self._gzip_decompress(a._async['make_response'].call_args[0][2]) + + @mock.patch('importlib.import_module') + def test_compression_disabled(self, import_module): + a = self.get_async_mock( + { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'sid=foo', + 'HTTP_ACCEPT_ENCODING': 'gzip', + } + ) + import_module.side_effect = [a] + s = async_server.AsyncServer( + http_compression=False, compression_threshold=0 + ) + s.sockets['foo'] = mock_socket = self._get_mock_socket() + mock_socket.handle_get_request.mock.return_value = [ + packet.Packet(packet.MESSAGE, data='hello') + ] + _run(s.handle_request('request')) + headers = a._async['make_response'].call_args[0][1] + for header, value in headers: + assert header != 'Content-Encoding' + with pytest.raises(IOError): + self._gzip_decompress(a._async['make_response'].call_args[0][2]) + + @mock.patch('importlib.import_module') + def test_compression_unknown(self, import_module): + a = self.get_async_mock( + { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'sid=foo', + 'HTTP_ACCEPT_ENCODING': 'rar', + } + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(compression_threshold=0) + s.sockets['foo'] = mock_socket = self._get_mock_socket() + mock_socket.handle_get_request.mock.return_value = [ + packet.Packet(packet.MESSAGE, data='hello') + ] + _run(s.handle_request('request')) + headers = a._async['make_response'].call_args[0][1] + for header, value in headers: + assert header != 'Content-Encoding' + with pytest.raises(IOError): + self._gzip_decompress(a._async['make_response'].call_args[0][2]) + + @mock.patch('importlib.import_module') + def test_compression_no_encoding(self, import_module): + a = self.get_async_mock( + { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'sid=foo', + 'HTTP_ACCEPT_ENCODING': '', + } + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(compression_threshold=0) + s.sockets['foo'] = mock_socket = self._get_mock_socket() + mock_socket.handle_get_request.mock.return_value = [ + packet.Packet(packet.MESSAGE, data='hello') + ] + _run(s.handle_request('request')) + headers = a._async['make_response'].call_args[0][1] + for header, value in headers: + assert header != 'Content-Encoding' + with pytest.raises(IOError): + self._gzip_decompress(a._async['make_response'].call_args[0][2]) + + @mock.patch('importlib.import_module') + def test_cookie(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer(cookie='sid') + s.generate_id = mock.MagicMock(return_value='123') + _run(s.handle_request('request')) + headers = a._async['make_response'].call_args[0][1] + assert ('Set-Cookie', 'sid=123; path=/; SameSite=Lax') in headers + + @mock.patch('importlib.import_module') + def test_cookie_dict(self, import_module): + def get_path(): + return '/a' + + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer(cookie={ + 'name': 'test', + 'path': get_path, + 'SameSite': 'None', + 'Secure': True, + 'HttpOnly': True + }) + s.generate_id = mock.MagicMock(return_value='123') + _run(s.handle_request('request')) + headers = a._async['make_response'].call_args[0][1] + assert ('Set-Cookie', 'test=123; path=/a; SameSite=None; Secure; ' + 'HttpOnly') in headers + + @mock.patch('importlib.import_module') + def test_no_cookie(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer(cookie=None) + s.generate_id = mock.MagicMock(return_value='123') + _run(s.handle_request('request')) + headers = a._async['make_response'].call_args[0][1] + for header, value in headers: + assert header != 'Set-Cookie' + + def test_logger(self): + s = async_server.AsyncServer(logger=False) + assert s.logger.getEffectiveLevel() == logging.ERROR + s.logger.setLevel(logging.NOTSET) + s = async_server.AsyncServer(logger=True) + assert s.logger.getEffectiveLevel() == logging.INFO + s.logger.setLevel(logging.WARNING) + s = async_server.AsyncServer(logger=True) + assert s.logger.getEffectiveLevel() == logging.WARNING + s.logger.setLevel(logging.NOTSET) + my_logger = logging.Logger('foo') + s = async_server.AsyncServer(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 +++' + + async_server.AsyncServer(json=CustomJSON) + pkt = packet.Packet(packet.MESSAGE, data={'foo': 'bar'}) + assert pkt.encode() == '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): + r = [] + + async def foo(arg): + r.append(arg) + + s = async_server.AsyncServer() + s.start_background_task(foo, 'bar') + pending = asyncio.all_tasks(loop=asyncio.get_event_loop()) \ + if hasattr(asyncio, 'all_tasks') else asyncio.Task.all_tasks() + asyncio.get_event_loop().run_until_complete(asyncio.wait(pending)) + assert r == ['bar'] + + def test_sleep(self): + s = async_server.AsyncServer() + _run(s.sleep(0)) + + def test_trigger_event_function(self): + result = [] + + def foo_handler(arg): + result.append('ok') + result.append(arg) + + s = async_server.AsyncServer() + s.on('message', handler=foo_handler) + _run(s._trigger_event('message', 'bar')) + assert result == ['ok', 'bar'] + + def test_trigger_event_coroutine(self): + result = [] + + async def foo_handler(arg): + result.append('ok') + result.append(arg) + + s = async_server.AsyncServer() + s.on('message', handler=foo_handler) + _run(s._trigger_event('message', 'bar')) + assert result == ['ok', 'bar'] + + def test_trigger_event_function_error(self): + def connect_handler(arg): + return 1 / 0 + + def foo_handler(arg): + return 1 / 0 + + s = async_server.AsyncServer() + s.on('connect', handler=connect_handler) + s.on('message', handler=foo_handler) + assert not _run(s._trigger_event('connect', '123')) + assert _run(s._trigger_event('message', 'bar')) is None + + def test_trigger_event_coroutine_error(self): + async def connect_handler(arg): + return 1 / 0 + + async def foo_handler(arg): + return 1 / 0 + + s = async_server.AsyncServer() + s.on('connect', handler=connect_handler) + s.on('message', handler=foo_handler) + assert not _run(s._trigger_event('connect', '123')) + assert _run(s._trigger_event('message', 'bar')) is None + + def test_trigger_event_function_async(self): + result = [] + + def foo_handler(arg): + result.append('ok') + result.append(arg) + + s = async_server.AsyncServer() + s.on('message', handler=foo_handler) + fut = _run(s._trigger_event('message', 'bar', run_async=True)) + asyncio.get_event_loop().run_until_complete(fut) + assert result == ['ok', 'bar'] + + def test_trigger_event_coroutine_async(self): + result = [] + + async def foo_handler(arg): + result.append('ok') + result.append(arg) + + s = async_server.AsyncServer() + s.on('message', handler=foo_handler) + fut = _run(s._trigger_event('message', 'bar', run_async=True)) + asyncio.get_event_loop().run_until_complete(fut) + assert result == ['ok', 'bar'] + + def test_trigger_event_function_async_error(self): + result = [] + + def foo_handler(arg): + result.append(arg) + return 1 / 0 + + s = async_server.AsyncServer() + s.on('message', handler=foo_handler) + fut = _run(s._trigger_event('message', 'bar', run_async=True)) + asyncio.get_event_loop().run_until_complete(fut) + assert result == ['bar'] + + def test_trigger_event_coroutine_async_error(self): + result = [] + + async def foo_handler(arg): + result.append(arg) + return 1 / 0 + + s = async_server.AsyncServer() + s.on('message', handler=foo_handler) + fut = _run(s._trigger_event('message', 'bar', run_async=True)) + asyncio.get_event_loop().run_until_complete(fut) + assert result == ['bar'] + + def test_create_queue(self): + s = async_server.AsyncServer() + q = s.create_queue() + empty = s.get_queue_empty_exception() + with pytest.raises(empty): + q.get_nowait() + + def test_create_event(self): + s = async_server.AsyncServer() + e = s.create_event() + assert not e.is_set() + e.set() + assert e.is_set() + + @mock.patch('importlib.import_module') + def test_service_task_started(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer(monitor_clients=True) + s._service_task = AsyncMock() + _run(s.handle_request('request')) + s._service_task.mock.assert_called_once_with() + + @mock.patch('importlib.import_module') + def test_shutdown(self, import_module): + a = self.get_async_mock() + import_module.side_effect = [a] + s = async_server.AsyncServer(monitor_clients=True) + _run(s.handle_request('request')) + assert s.service_task_handle is not None + _run(s.shutdown()) + assert s.service_task_handle is None + + @mock.patch('importlib.import_module') + def test_transports_disallowed(self, import_module): + a = self.get_async_mock( + {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'transport=polling'} + ) + import_module.side_effect = [a] + s = async_server.AsyncServer(transports='websocket') + response = _run(s.handle_request('request')) + assert response == 'response' + a._async['translate_request'].assert_called_once_with('request') + assert a._async['make_response'].call_count == 1 + assert a._async['make_response'].call_args[0][0] == '400 BAD REQUEST' diff --git a/tests/async/test_socket.py b/tests/async/test_socket.py new file mode 100644 index 0000000..b1ea8b4 --- /dev/null +++ b/tests/async/test_socket.py @@ -0,0 +1,551 @@ +import asyncio +import time +import unittest +from unittest import mock + +import pytest + +from engineio import async_socket +from engineio import exceptions +from engineio import packet +from engineio import payload + + +def AsyncMock(*args, **kwargs): + """Return a mock asynchronous function.""" + m = mock.MagicMock(*args, **kwargs) + + async def mock_coro(*args, **kwargs): + return m(*args, **kwargs) + + mock_coro.mock = m + return mock_coro + + +def _run(coro): + """Run the given coroutine.""" + return asyncio.get_event_loop().run_until_complete(coro) + + +class TestSocket(unittest.TestCase): + def _get_read_mock_coro(self, payload): + mock_input = mock.MagicMock() + mock_input.read = AsyncMock() + mock_input.read.mock.return_value = payload + return mock_input + + def _get_mock_server(self): + mock_server = mock.Mock() + mock_server.ping_timeout = 0.2 + mock_server.ping_interval = 0.2 + mock_server.ping_interval_grace_period = 0.001 + mock_server.async_handlers = False + mock_server.max_http_buffer_size = 128 + mock_server._async = { + 'asyncio': True, + 'create_route': mock.MagicMock(), + 'translate_request': mock.MagicMock(), + 'make_response': mock.MagicMock(), + 'websocket': 'w', + } + mock_server._async['translate_request'].return_value = 'request' + mock_server._async['make_response'].return_value = 'response' + mock_server._trigger_event = AsyncMock() + + def bg_task(target, *args, **kwargs): + return asyncio.ensure_future(target(*args, **kwargs)) + + def create_queue(*args, **kwargs): + queue = asyncio.Queue(*args, **kwargs) + queue.Empty = asyncio.QueueEmpty + return queue + + mock_server.start_background_task = bg_task + mock_server.create_queue = create_queue + return mock_server + + def test_create(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + assert s.server == mock_server + assert s.sid == 'sid' + assert not s.upgraded + assert not s.closed + assert hasattr(s.queue, 'get') + assert hasattr(s.queue, 'put') + assert hasattr(s.queue, 'task_done') + assert hasattr(s.queue, 'join') + + def test_empty_poll(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + with pytest.raises(exceptions.QueueEmpty): + _run(s.poll()) + + def test_poll(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + pkt1 = packet.Packet(packet.MESSAGE, data='hello') + pkt2 = packet.Packet(packet.MESSAGE, data='bye') + _run(s.send(pkt1)) + _run(s.send(pkt2)) + assert _run(s.poll()) == [pkt1, pkt2] + + def test_poll_none(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + _run(s.queue.put(None)) + assert _run(s.poll()) == [] + + def test_poll_none_after_packet(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + pkt = packet.Packet(packet.MESSAGE, data='hello') + _run(s.send(pkt)) + _run(s.queue.put(None)) + assert _run(s.poll()) == [pkt] + assert _run(s.poll()) == [] + + def test_schedule_ping(self): + mock_server = self._get_mock_server() + mock_server.ping_interval = 0.01 + s = async_socket.AsyncSocket(mock_server, 'sid') + s.send = AsyncMock() + + async def schedule_ping(): + s.schedule_ping() + await asyncio.sleep(0.05) + + _run(schedule_ping()) + assert s.last_ping is not None + assert s.send.mock.call_args_list[0][0][0].encode() == '2' + + def test_schedule_ping_closed_socket(self): + mock_server = self._get_mock_server() + mock_server.ping_interval = 0.01 + s = async_socket.AsyncSocket(mock_server, 'sid') + s.send = AsyncMock() + s.closed = True + + async def schedule_ping(): + s.schedule_ping() + await asyncio.sleep(0.05) + + _run(schedule_ping()) + assert s.last_ping is None + s.send.mock.assert_not_called() + + def test_pong(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.schedule_ping = mock.MagicMock() + _run(s.receive(packet.Packet(packet.PONG, data='abc'))) + s.schedule_ping.assert_called_once_with() + + def test_message_sync_handler(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + _run(s.receive(packet.Packet(packet.MESSAGE, data='foo'))) + mock_server._trigger_event.mock.assert_called_once_with( + 'message', 'sid', 'foo', run_async=False + ) + + def test_message_async_handler(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + mock_server.async_handlers = True + _run(s.receive(packet.Packet(packet.MESSAGE, data='foo'))) + mock_server._trigger_event.mock.assert_called_once_with( + 'message', 'sid', 'foo', run_async=True + ) + + def test_invalid_packet(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + with pytest.raises(exceptions.UnknownPacketError): + _run(s.receive(packet.Packet(packet.OPEN))) + + def test_timeout(self): + mock_server = self._get_mock_server() + mock_server.ping_interval = 6 + mock_server.ping_interval_grace_period = 2 + s = async_socket.AsyncSocket(mock_server, 'sid') + s.last_ping = time.time() - 9 + s.close = AsyncMock() + _run(s.send('packet')) + s.close.mock.assert_called_once_with(wait=False, abort=False) + + def test_polling_read(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'foo') + pkt1 = packet.Packet(packet.MESSAGE, data='hello') + pkt2 = packet.Packet(packet.MESSAGE, data='bye') + _run(s.send(pkt1)) + _run(s.send(pkt2)) + environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'sid=foo'} + packets = _run(s.handle_get_request(environ)) + assert packets == [pkt1, pkt2] + + def test_polling_read_error(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'foo') + environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'sid=foo'} + with pytest.raises(exceptions.QueueEmpty): + _run(s.handle_get_request(environ)) + + def test_polling_write(self): + mock_server = self._get_mock_server() + mock_server.max_http_buffer_size = 1000 + pkt1 = packet.Packet(packet.MESSAGE, data='hello') + pkt2 = packet.Packet(packet.MESSAGE, data='bye') + p = payload.Payload(packets=[pkt1, pkt2]).encode().encode('utf-8') + s = async_socket.AsyncSocket(mock_server, 'foo') + s.receive = AsyncMock() + environ = { + 'REQUEST_METHOD': 'POST', + 'QUERY_STRING': 'sid=foo', + 'CONTENT_LENGTH': len(p), + 'wsgi.input': self._get_read_mock_coro(p), + } + _run(s.handle_post_request(environ)) + assert s.receive.mock.call_count == 2 + + def test_polling_write_too_large(self): + mock_server = self._get_mock_server() + pkt1 = packet.Packet(packet.MESSAGE, data='hello') + pkt2 = packet.Packet(packet.MESSAGE, data='bye') + p = payload.Payload(packets=[pkt1, pkt2]).encode().encode('utf-8') + mock_server.max_http_buffer_size = len(p) - 1 + s = async_socket.AsyncSocket(mock_server, 'foo') + s.receive = AsyncMock() + environ = { + 'REQUEST_METHOD': 'POST', + 'QUERY_STRING': 'sid=foo', + 'CONTENT_LENGTH': len(p), + 'wsgi.input': self._get_read_mock_coro(p), + } + with pytest.raises(exceptions.ContentTooLongError): + _run(s.handle_post_request(environ)) + + def test_upgrade_handshake(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'foo') + s._upgrade_websocket = AsyncMock() + environ = { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'sid=foo', + 'HTTP_CONNECTION': 'Foo,Upgrade,Bar', + 'HTTP_UPGRADE': 'websocket', + } + _run(s.handle_get_request(environ)) + s._upgrade_websocket.mock.assert_called_once_with(environ) + + def test_upgrade(self): + mock_server = self._get_mock_server() + mock_server._async['websocket'] = mock.MagicMock() + mock_ws = AsyncMock() + mock_server._async['websocket'].return_value = mock_ws + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = True + environ = "foo" + _run(s._upgrade_websocket(environ)) + mock_server._async['websocket'].assert_called_once_with( + s._websocket_handler, mock_server + ) + mock_ws.mock.assert_called_once_with(environ) + + def test_upgrade_twice(self): + mock_server = self._get_mock_server() + mock_server._async['websocket'] = mock.MagicMock() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = True + s.upgraded = True + environ = "foo" + with pytest.raises(IOError): + _run(s._upgrade_websocket(environ)) + + def test_upgrade_packet(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = True + _run(s.receive(packet.Packet(packet.UPGRADE))) + r = _run(s.poll()) + assert len(r) == 1 + assert r[0].encode() == packet.Packet(packet.NOOP).encode() + + def test_upgrade_no_probe(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = True + ws = mock.MagicMock() + ws.wait = AsyncMock() + ws.wait.mock.return_value = packet.Packet(packet.NOOP).encode() + _run(s._websocket_handler(ws)) + assert not s.upgraded + + def test_upgrade_no_upgrade_packet(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = True + s.queue.join = AsyncMock(return_value=None) + ws = mock.MagicMock() + ws.send = AsyncMock() + ws.wait = AsyncMock() + probe = 'probe' + ws.wait.mock.side_effect = [ + packet.Packet(packet.PING, data=probe).encode(), + packet.Packet(packet.NOOP).encode(), + ] + _run(s._websocket_handler(ws)) + ws.send.mock.assert_called_once_with( + packet.Packet(packet.PONG, data=probe).encode() + ) + assert _run(s.queue.get()).packet_type == packet.NOOP + assert not s.upgraded + + def test_upgrade_not_supported(self): + mock_server = self._get_mock_server() + mock_server._async['websocket'] = None + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = True + environ = "foo" + _run(s._upgrade_websocket(environ)) + mock_server._bad_request.assert_called_once_with() + + def test_close_packet(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = True + s.close = AsyncMock() + _run(s.receive(packet.Packet(packet.CLOSE))) + s.close.mock.assert_called_once_with(wait=False, abort=True) + + def test_websocket_read_write(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = False + s.queue.join = AsyncMock(return_value=None) + foo = 'foo' + bar = 'bar' + s.poll = AsyncMock( + side_effect=[[packet.Packet(packet.MESSAGE, data=bar)], None] + ) + ws = mock.MagicMock() + ws.send = AsyncMock() + ws.wait = AsyncMock() + ws.wait.mock.side_effect = [ + packet.Packet(packet.MESSAGE, data=foo).encode(), + None, + ] + ws.close = AsyncMock() + _run(s._websocket_handler(ws)) + assert s.connected + assert s.upgraded + assert mock_server._trigger_event.mock.call_count == 2 + mock_server._trigger_event.mock.assert_has_calls( + [ + mock.call('message', 'sid', 'foo', run_async=False), + mock.call('disconnect', 'sid'), + ] + ) + ws.send.mock.assert_called_with('4bar') + ws.close.mock.assert_called() + + def test_websocket_upgrade_read_write(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = True + s.queue.join = AsyncMock(return_value=None) + foo = 'foo' + bar = 'bar' + probe = 'probe' + s.poll = AsyncMock( + side_effect=[ + [packet.Packet(packet.MESSAGE, data=bar)], + exceptions.QueueEmpty, + ] + ) + ws = mock.MagicMock() + ws.send = AsyncMock() + ws.wait = AsyncMock() + ws.wait.mock.side_effect = [ + packet.Packet(packet.PING, data=probe).encode(), + packet.Packet(packet.UPGRADE).encode(), + packet.Packet(packet.MESSAGE, data=foo).encode(), + None, + ] + ws.close = AsyncMock() + _run(s._websocket_handler(ws)) + assert s.upgraded + assert mock_server._trigger_event.mock.call_count == 2 + mock_server._trigger_event.mock.assert_has_calls( + [ + mock.call('message', 'sid', 'foo', run_async=False), + mock.call('disconnect', 'sid'), + ] + ) + ws.send.mock.assert_called_with('4bar') + ws.close.mock.assert_called() + + def test_websocket_upgrade_with_payload(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = True + s.queue.join = AsyncMock(return_value=None) + probe = 'probe' + ws = mock.MagicMock() + ws.send = AsyncMock() + ws.wait = AsyncMock() + ws.wait.mock.side_effect = [ + packet.Packet(packet.PING, data=probe).encode(), + packet.Packet(packet.UPGRADE, data='2').encode(), + ] + ws.close = AsyncMock() + _run(s._websocket_handler(ws)) + assert s.upgraded + + def test_websocket_upgrade_with_backlog(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = True + s.queue.join = AsyncMock(return_value=None) + probe = 'probe' + foo = 'foo' + ws = mock.MagicMock() + ws.send = AsyncMock() + ws.wait = AsyncMock() + ws.wait.mock.side_effect = [ + packet.Packet(packet.PING, data=probe).encode(), + packet.Packet(packet.UPGRADE, data='2').encode(), + ] + ws.close = AsyncMock() + s.upgrading = True + _run(s.send(packet.Packet(packet.MESSAGE, data=foo))) + environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'sid=sid'} + packets = _run(s.handle_get_request(environ)) + assert len(packets) == 1 + assert packets[0].encode() == '6' + packets = _run(s.poll()) + assert len(packets) == 1 + assert packets[0].encode() == '4foo' + + _run(s._websocket_handler(ws)) + assert s.upgraded + assert not s.upgrading + packets = _run(s.handle_get_request(environ)) + assert len(packets) == 1 + assert packets[0].encode() == '6' + + def test_websocket_read_write_wait_fail(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = False + s.queue.join = AsyncMock(return_value=None) + foo = 'foo' + bar = 'bar' + s.poll = AsyncMock( + side_effect=[ + [packet.Packet(packet.MESSAGE, data=bar)], + [packet.Packet(packet.MESSAGE, data=bar)], + exceptions.QueueEmpty, + ] + ) + ws = mock.MagicMock() + ws.send = AsyncMock() + ws.wait = AsyncMock() + ws.wait.mock.side_effect = [ + packet.Packet(packet.MESSAGE, data=foo).encode(), + RuntimeError, + ] + ws.send.mock.side_effect = [None, RuntimeError] + ws.close = AsyncMock() + _run(s._websocket_handler(ws)) + assert s.closed + + def test_websocket_upgrade_with_large_packet(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = True + s.queue.join = AsyncMock(return_value=None) + probe = 'probe' + ws = mock.MagicMock() + ws.send = AsyncMock() + ws.wait = AsyncMock() + ws.wait.mock.side_effect = [ + packet.Packet(packet.PING, data=probe).encode(), + packet.Packet(packet.UPGRADE, data='2' * 128).encode(), + ] + with pytest.raises(ValueError): + _run(s._websocket_handler(ws)) + assert not s.upgraded + + def test_websocket_ignore_invalid_packet(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.connected = False + s.queue.join = AsyncMock(return_value=None) + foo = 'foo' + bar = 'bar' + s.poll = AsyncMock( + side_effect=[ + [packet.Packet(packet.MESSAGE, data=bar)], + exceptions.QueueEmpty, + ] + ) + ws = mock.MagicMock() + ws.send = AsyncMock() + ws.wait = AsyncMock() + ws.wait.mock.side_effect = [ + packet.Packet(packet.OPEN).encode(), + packet.Packet(packet.MESSAGE, data=foo).encode(), + None, + ] + ws.close = AsyncMock() + _run(s._websocket_handler(ws)) + assert s.connected + assert mock_server._trigger_event.mock.call_count == 2 + mock_server._trigger_event.mock.assert_has_calls( + [ + mock.call('message', 'sid', foo, run_async=False), + mock.call('disconnect', 'sid'), + ] + ) + ws.send.mock.assert_called_with('4bar') + ws.close.mock.assert_called() + + def test_send_after_close(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + _run(s.close(wait=False)) + with pytest.raises(exceptions.SocketIsClosedError): + _run(s.send(packet.Packet(packet.NOOP))) + + def test_close_after_close(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + _run(s.close(wait=False)) + assert s.closed + assert mock_server._trigger_event.mock.call_count == 1 + mock_server._trigger_event.mock.assert_called_once_with( + 'disconnect', 'sid' + ) + _run(s.close()) + assert mock_server._trigger_event.mock.call_count == 1 + + def test_close_and_wait(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.queue = mock.MagicMock() + s.queue.put = AsyncMock() + s.queue.join = AsyncMock() + _run(s.close(wait=True)) + s.queue.join.mock.assert_called_once_with() + + def test_close_without_wait(self): + mock_server = self._get_mock_server() + s = async_socket.AsyncSocket(mock_server, 'sid') + s.queue = mock.MagicMock() + s.queue.put = AsyncMock() + s.queue.join = AsyncMock() + _run(s.close(wait=False)) + assert s.queue.join.mock.call_count == 0 diff --git a/tests/async/test_tornado.py b/tests/async/test_tornado.py new file mode 100644 index 0000000..f2ea3c0 --- /dev/null +++ b/tests/async/test_tornado.py @@ -0,0 +1,66 @@ +import asyncio +import unittest +from unittest import mock + +try: + import tornado.web +except ImportError: + pass + +from engineio.async_drivers import tornado as async_tornado + + +def _run(coro): + """Run the given coroutine.""" + return asyncio.get_event_loop().run_until_complete(coro) + + +class TornadoTests(unittest.TestCase): + def test_get_tornado_handler(self): + mock_server = mock.MagicMock() + handler = async_tornado.get_tornado_handler(mock_server) + assert issubclass(handler, tornado.websocket.WebSocketHandler) + + def test_translate_request(self): + mock_handler = mock.MagicMock() + mock_handler.request.method = 'PUT' + mock_handler.request.path = '/foo/bar' + mock_handler.request.query = 'baz=1' + mock_handler.request.version = '1.1' + mock_handler.request.headers = { + 'a': 'b', + 'c': 'd', + 'content-type': 'application/json', + 'content-length': 123, + } + mock_handler.request.body = b'hello world' + environ = async_tornado.translate_request(mock_handler) + expected_environ = { + 'REQUEST_METHOD': 'PUT', + 'PATH_INFO': '/foo/bar', + 'QUERY_STRING': 'baz=1', + 'CONTENT_TYPE': 'application/json', + 'CONTENT_LENGTH': 123, + 'HTTP_A': 'b', + 'HTTP_C': 'd', + 'RAW_URI': '/foo/bar?baz=1', + 'SERVER_PROTOCOL': 'HTTP/1.1', + # 'wsgi.input': b'hello world', + 'tornado.handler': mock_handler, + } + for k, v in expected_environ.items(): + assert v == environ[k] + payload = _run(environ['wsgi.input'].read(1)) + payload += _run(environ['wsgi.input'].read()) + assert payload == b'hello world' + + def test_make_response(self): + mock_handler = mock.MagicMock() + mock_environ = {'tornado.handler': mock_handler} + async_tornado.make_response( + '202 ACCEPTED', [('foo', 'bar')], b'payload', mock_environ + ) + mock_handler.set_status.assert_called_once_with(202) + mock_handler.set_header.assert_called_once_with('foo', 'bar') + mock_handler.write.assert_called_once_with(b'payload') + mock_handler.finish.assert_called_once_with() diff --git a/tests/common/__init__.py b/tests/common/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/tests/common/__init__.py diff --git a/tests/common/index.html b/tests/common/index.html new file mode 100644 index 0000000..18ecdcb --- /dev/null +++ b/tests/common/index.html @@ -0,0 +1 @@ +<html></html> diff --git a/tests/common/test_async_eventlet.py b/tests/common/test_async_eventlet.py new file mode 100644 index 0000000..9cd652d --- /dev/null +++ b/tests/common/test_async_eventlet.py @@ -0,0 +1,29 @@ +import logging +import unittest +from unittest import mock + +from engineio.async_drivers import eventlet as async_eventlet +import pytest + + +class TestAsyncEventlet(unittest.TestCase): + def setUp(self): + logging.getLogger('engineio').setLevel(logging.NOTSET) + + def test_bad_environ(self): + wsgi = async_eventlet.WebSocketWSGI(None, mock.MagicMock()) + environ = {'foo': 'bar'} + start_response = 'bar' + with pytest.raises(RuntimeError): + wsgi(environ, start_response) + + @mock.patch( + 'engineio.async_drivers.eventlet._WebSocketWSGI.__call__', + return_value='data', + ) + def test_wsgi_call(self, _WebSocketWSGI): + _WebSocketWSGI.__call__ = lambda e, s: 'data' + environ = {'eventlet.input': mock.MagicMock()} + start_response = 'bar' + wsgi = async_eventlet.WebSocketWSGI(None, mock.MagicMock()) + assert wsgi(environ, start_response) == 'data' diff --git a/tests/common/test_client.py b/tests/common/test_client.py new file mode 100644 index 0000000..d96067d --- /dev/null +++ b/tests/common/test_client.py @@ -0,0 +1,1727 @@ +import logging +import ssl +import time +import unittest +from unittest import mock + +import pytest +import websocket + +from engineio import base_client +from engineio import client +from engineio import exceptions +from engineio import json +from engineio import packet +from engineio import payload + + +class TestClient(unittest.TestCase): + def test_is_asyncio_based(self): + c = client.Client() + assert not c.is_asyncio_based() + + def test_create(self): + c = client.Client() + assert c.handlers == {} + for attr in [ + 'base_url', + 'transports', + 'sid', + 'upgrades', + 'ping_interval', + 'ping_timeout', + 'http', + 'ws', + 'read_loop_task', + 'write_loop_task', + 'queue', + ]: + assert getattr(c, attr) is None, attr + ' is not None' + assert c.state == 'disconnected' + + def test_custon_json(self): + client.Client() + assert packet.Packet.json == json + client.Client(json='foo') + assert packet.Packet.json == 'foo' + packet.Packet.json = json + + def test_logger(self): + c = client.Client(logger=False) + assert c.logger.getEffectiveLevel() == logging.ERROR + c.logger.setLevel(logging.NOTSET) + c = client.Client(logger=True) + assert c.logger.getEffectiveLevel() == logging.INFO + c.logger.setLevel(logging.WARNING) + c = client.Client(logger=True) + assert c.logger.getEffectiveLevel() == logging.WARNING + c.logger.setLevel(logging.NOTSET) + my_logger = logging.Logger('foo') + c = client.Client(logger=my_logger) + assert c.logger == my_logger + + def test_custom_timeout(self): + c = client.Client() + assert c.request_timeout == 5 + c = client.Client(request_timeout=27) + assert c.request_timeout == 27 + + def test_on_event(self): + c = client.Client() + + @c.on('connect') + def foo(): + pass + + c.on('disconnect', foo) + + assert c.handlers['connect'] == foo + assert c.handlers['disconnect'] == foo + + def test_on_event_invalid(self): + c = client.Client() + with pytest.raises(ValueError): + c.on('invalid') + + def test_already_connected(self): + c = client.Client() + c.state = 'connected' + with pytest.raises(ValueError): + c.connect('http://foo') + + def test_invalid_transports(self): + c = client.Client() + with pytest.raises(ValueError): + c.connect('http://foo', transports=['foo', 'bar']) + + def test_some_invalid_transports(self): + c = client.Client() + c._connect_websocket = mock.MagicMock() + c.connect('http://foo', transports=['foo', 'websocket', 'bar']) + assert c.transports == ['websocket'] + + def test_connect_polling(self): + c = client.Client() + c._connect_polling = mock.MagicMock(return_value='foo') + assert c.connect('http://foo') == 'foo' + c._connect_polling.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + + c = client.Client() + c._connect_polling = mock.MagicMock(return_value='foo') + assert c.connect('http://foo', transports=['polling']) == 'foo' + c._connect_polling.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + + c = client.Client() + c._connect_polling = mock.MagicMock(return_value='foo') + assert ( + c.connect('http://foo', transports=['polling', 'websocket']) + == 'foo' + ) + c._connect_polling.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + + def test_connect_websocket(self): + c = client.Client() + c._connect_websocket = mock.MagicMock(return_value='foo') + assert c.connect('http://foo', transports=['websocket']) == 'foo' + c._connect_websocket.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + + c = client.Client() + c._connect_websocket = mock.MagicMock(return_value='foo') + assert c.connect('http://foo', transports='websocket') == 'foo' + c._connect_websocket.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + + def test_connect_query_string(self): + c = client.Client() + c._connect_polling = mock.MagicMock(return_value='foo') + assert c.connect('http://foo?bar=baz') == 'foo' + c._connect_polling.assert_called_once_with( + 'http://foo?bar=baz', {}, 'engine.io' + ) + + def test_connect_custom_headers(self): + c = client.Client() + c._connect_polling = mock.MagicMock(return_value='foo') + assert c.connect('http://foo', headers={'Foo': 'Bar'}) == 'foo' + c._connect_polling.assert_called_once_with( + 'http://foo', {'Foo': 'Bar'}, 'engine.io' + ) + + def test_wait(self): + c = client.Client() + c.read_loop_task = mock.MagicMock() + c.wait() + c.read_loop_task.join.assert_called_once_with() + + def test_wait_no_task(self): + c = client.Client() + c.read_loop_task = None + c.wait() # should not block + + def test_send(self): + c = client.Client() + saved_packets = [] + + def fake_send_packet(pkt): + saved_packets.append(pkt) + + c._send_packet = fake_send_packet + c.send('foo') + c.send('foo') + c.send(b'foo') + assert saved_packets[0].packet_type == packet.MESSAGE + assert saved_packets[0].data == 'foo' + assert not saved_packets[0].binary + assert saved_packets[1].packet_type == packet.MESSAGE + assert saved_packets[1].data == 'foo' + assert not saved_packets[1].binary + assert saved_packets[2].packet_type == packet.MESSAGE + assert saved_packets[2].data == b'foo' + assert saved_packets[2].binary + + def test_disconnect_not_connected(self): + c = client.Client() + c.state = 'foo' + c.sid = 'bar' + c.disconnect() + assert c.state == 'disconnected' + assert c.sid is None + + def test_disconnect_polling(self): + c = client.Client() + base_client.connected_clients.append(c) + c.state = 'connected' + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.read_loop_task = mock.MagicMock() + c.ws = mock.MagicMock() + c._trigger_event = mock.MagicMock() + c.disconnect() + c.read_loop_task.join.assert_called_once_with() + c.ws.mock.assert_not_called() + assert c not in base_client.connected_clients + c._trigger_event.assert_called_once_with('disconnect', run_async=False) + + def test_disconnect_websocket(self): + c = client.Client() + base_client.connected_clients.append(c) + c.state = 'connected' + c.current_transport = 'websocket' + c.queue = mock.MagicMock() + c.read_loop_task = mock.MagicMock() + c.ws = mock.MagicMock() + c._trigger_event = mock.MagicMock() + c.disconnect() + c.read_loop_task.join.assert_called_once_with() + c.ws.close.assert_called_once_with() + assert c not in base_client.connected_clients + c._trigger_event.assert_called_once_with('disconnect', run_async=False) + + def test_disconnect_polling_abort(self): + c = client.Client() + base_client.connected_clients.append(c) + c.state = 'connected' + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.read_loop_task = mock.MagicMock() + c.ws = mock.MagicMock() + c.disconnect(abort=True) + c.queue.join.assert_not_called() + c.read_loop_task.join.assert_not_called() + c.ws.mock.assert_not_called() + assert c not in base_client.connected_clients + + def test_disconnect_websocket_abort(self): + c = client.Client() + base_client.connected_clients.append(c) + c.state = 'connected' + c.current_transport = 'websocket' + c.queue = mock.MagicMock() + c.read_loop_task = mock.MagicMock() + c.ws = mock.MagicMock() + c.disconnect(abort=True) + c.queue.join.assert_not_called() + c.read_loop_task.join.assert_not_called() + c.ws.mock.assert_not_called() + assert c not in base_client.connected_clients + + def test_current_transport(self): + c = client.Client() + c.current_transport = 'foo' + assert c.transport() == 'foo' + + def test_background_tasks(self): + flag = {} + + def bg_task(): + flag['task'] = True + + c = client.Client() + task = c.start_background_task(bg_task) + task.join() + assert 'task' in flag + assert flag['task'] + + def test_sleep(self): + c = client.Client() + t = time.time() + c.sleep(0.1) + assert time.time() - t > 0.1 + + def test_create_queue(self): + c = client.Client() + q = c.create_queue() + with pytest.raises(q.Empty): + q.get(timeout=0.01) + + def test_create_event(self): + c = client.Client() + e = c.create_event() + assert not e.is_set() + e.set() + assert e.is_set() + + @mock.patch('engineio.client.time.time', return_value=123.456) + @mock.patch('engineio.client.Client._send_request', return_value=None) + def test_polling_connection_failed(self, _send_request, _time): + c = client.Client() + with pytest.raises(exceptions.ConnectionError): + c.connect('http://foo', headers={'Foo': 'Bar'}) + _send_request.assert_called_once_with( + 'GET', + 'http://foo/engine.io/?transport=polling&EIO=4&t=123.456', + headers={'Foo': 'Bar'}, + timeout=5, + ) + + @mock.patch('engineio.client.Client._send_request') + def test_polling_connection_404(self, _send_request): + _send_request.return_value.status_code = 404 + _send_request.return_value.json.return_value = {'foo': 'bar'} + c = client.Client() + try: + c.connect('http://foo') + except exceptions.ConnectionError as exc: + assert len(exc.args) == 2 + assert ( + exc.args[0] == 'Unexpected status code 404 in server response' + ) + assert exc.args[1] == {'foo': 'bar'} + + @mock.patch('engineio.client.Client._send_request') + def test_polling_connection_404_no_json(self, _send_request): + _send_request.return_value.status_code = 404 + _send_request.return_value.json.side_effect = json.JSONDecodeError( + 'error', '<html></html>', 0) + c = client.Client() + try: + c.connect('http://foo') + except exceptions.ConnectionError as exc: + assert len(exc.args) == 2 + assert ( + exc.args[0] == 'Unexpected status code 404 in server response' + ) + assert exc.args[1] is None + + @mock.patch('engineio.client.Client._send_request') + def test_polling_connection_invalid_packet(self, _send_request): + _send_request.return_value.status_code = 200 + _send_request.return_value.content = b'foo' + c = client.Client() + with pytest.raises(exceptions.ConnectionError): + c.connect('http://foo') + + @mock.patch('engineio.client.Client._send_request') + def test_polling_connection_no_open_packet(self, _send_request): + _send_request.return_value.status_code = 200 + _send_request.return_value.content = payload.Payload( + packets=[ + packet.Packet( + packet.CLOSE, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 10, + 'pingTimeout': 20, + }, + ) + ] + ).encode().encode('utf-8') + c = client.Client() + with pytest.raises(exceptions.ConnectionError): + c.connect('http://foo') + + @mock.patch('engineio.client.Client._send_request') + def test_polling_connection_successful(self, _send_request): + _send_request.return_value.status_code = 200 + _send_request.return_value.content = payload.Payload( + packets=[ + packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ) + ] + ).encode().encode('utf-8') + c = client.Client() + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('http://foo') + time.sleep(0.1) + + c._read_loop_polling.assert_called_once_with() + c._read_loop_websocket.assert_not_called() + c._write_loop.assert_called_once_with() + on_connect.assert_called_once_with() + assert c in base_client.connected_clients + assert ( + c.base_url + == 'http://foo/engine.io/?transport=polling&EIO=4&sid=123' + ) + assert c.sid == '123' + assert c.ping_interval == 1 + assert c.ping_timeout == 2 + assert c.upgrades == [] + assert c.transport() == 'polling' + + @mock.patch('engineio.client.Client._send_request') + def test_polling_https_noverify_connection_successful(self, _send_request): + _send_request.return_value.status_code = 200 + _send_request.return_value.content = payload.Payload( + packets=[ + packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ) + ] + ).encode().encode('utf-8') + c = client.Client(ssl_verify=False) + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('https://foo') + time.sleep(0.1) + + c._read_loop_polling.assert_called_once_with() + c._read_loop_websocket.assert_not_called() + c._write_loop.assert_called_once_with() + on_connect.assert_called_once_with() + assert c in base_client.connected_clients + assert ( + c.base_url + == 'https://foo/engine.io/?transport=polling&EIO=4&sid=123' + ) + assert c.sid == '123' + assert c.ping_interval == 1 + assert c.ping_timeout == 2 + assert c.upgrades == [] + assert c.transport() == 'polling' + + @mock.patch('engineio.client.Client._send_request') + def test_polling_connection_with_more_packets(self, _send_request): + _send_request.return_value.status_code = 200 + _send_request.return_value.content = payload.Payload( + packets=[ + packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ), + packet.Packet(packet.NOOP), + ] + ).encode().encode('utf-8') + c = client.Client() + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + c._receive_packet = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('http://foo') + time.sleep(0.1) + assert c._receive_packet.call_count == 1 + assert ( + c._receive_packet.call_args_list[0][0][0].packet_type + == packet.NOOP + ) + + @mock.patch('engineio.client.Client._send_request') + def test_polling_connection_upgraded(self, _send_request): + _send_request.return_value.status_code = 200 + _send_request.return_value.content = payload.Payload( + packets=[ + packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': ['websocket'], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ) + ] + ).encode().encode('utf-8') + c = client.Client() + c._connect_websocket = mock.MagicMock(return_value=True) + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('http://foo') + + c._connect_websocket.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + on_connect.assert_called_once_with() + assert c in base_client.connected_clients + assert ( + c.base_url + == 'http://foo/engine.io/?transport=polling&EIO=4&sid=123' + ) + assert c.sid == '123' + assert c.ping_interval == 1 + assert c.ping_timeout == 2 + assert c.upgrades == ['websocket'] + + @mock.patch('engineio.client.Client._send_request') + def test_polling_connection_not_upgraded(self, _send_request): + _send_request.return_value.status_code = 200 + _send_request.return_value.content = payload.Payload( + packets=[ + packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': ['websocket'], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ) + ] + ).encode().encode('utf-8') + c = client.Client() + c._connect_websocket = mock.MagicMock(return_value=False) + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('http://foo') + time.sleep(0.1) + + c._connect_websocket.assert_called_once_with( + 'http://foo', {}, 'engine.io' + ) + c._read_loop_polling.assert_called_once_with() + c._read_loop_websocket.assert_not_called() + c._write_loop.assert_called_once_with() + on_connect.assert_called_once_with() + assert c in base_client.connected_clients + + @mock.patch('engineio.client.time.time', return_value=123.456) + @mock.patch( + 'engineio.client.websocket.create_connection', + side_effect=[ConnectionError], + ) + def test_websocket_connection_failed(self, create_connection, _time): + c = client.Client() + with pytest.raises(exceptions.ConnectionError): + c.connect( + 'http://foo', transports=['websocket'], headers={'Foo': 'Bar'} + ) + create_connection.assert_called_once_with( + 'ws://foo/engine.io/?transport=websocket&EIO=4&t=123.456', + header={'Foo': 'Bar'}, + cookie=None, + enable_multithread=True, + timeout=5 + ) + + @mock.patch('engineio.client.time.time', return_value=123.456) + @mock.patch( + 'engineio.client.websocket.create_connection', + side_effect=[ConnectionError], + ) + def test_websocket_connection_extra(self, create_connection, _time): + c = client.Client(websocket_extra_options={'header': {'Baz': 'Qux'}, + 'timeout': 10}) + with pytest.raises(exceptions.ConnectionError): + c.connect( + 'http://foo', transports=['websocket'], headers={'Foo': 'Bar'} + ) + create_connection.assert_called_once_with( + 'ws://foo/engine.io/?transport=websocket&EIO=4&t=123.456', + header={'Foo': 'Bar', 'Baz': 'Qux'}, + cookie=None, + enable_multithread=True, + timeout=10 + ) + + @mock.patch('engineio.client.time.time', return_value=123.456) + @mock.patch( + 'engineio.client.websocket.create_connection', + side_effect=[websocket.WebSocketException], + ) + def test_websocket_connection_failed_with_websocket_error( + self, create_connection, _time + ): + c = client.Client() + with pytest.raises(exceptions.ConnectionError): + c.connect( + 'http://foo', transports=['websocket'], headers={'Foo': 'Bar'} + ) + create_connection.assert_called_once_with( + 'ws://foo/engine.io/?transport=websocket&EIO=4&t=123.456', + header={'Foo': 'Bar'}, + cookie=None, + enable_multithread=True, + timeout=5 + ) + + @mock.patch('engineio.client.time.time', return_value=123.456) + @mock.patch( + 'engineio.client.websocket.create_connection', + side_effect=[ConnectionError], + ) + def test_websocket_upgrade_failed(self, create_connection, _time): + c = client.Client() + c.ping_interval = 1 + c.ping_timeout = 2 + c.sid = '123' + assert not c.connect('http://foo', transports=['websocket']) + create_connection.assert_called_once_with( + 'ws://foo/engine.io/?transport=websocket&EIO=4&sid=123&t=123.456', + header={}, + cookie=None, + enable_multithread=True, + timeout=5 + ) + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_connection_no_open_packet(self, create_connection): + create_connection.return_value.recv.return_value = packet.Packet( + packet.CLOSE + ).encode() + c = client.Client() + with pytest.raises(exceptions.ConnectionError): + c.connect('http://foo', transports=['websocket']) + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_connection_successful(self, create_connection): + create_connection.return_value.recv.return_value = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + c = client.Client() + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('ws://foo', transports=['websocket']) + time.sleep(0.1) + + c._read_loop_polling.assert_not_called() + c._read_loop_websocket.assert_called_once_with() + c._write_loop.assert_called_once_with() + on_connect.assert_called_once_with() + assert c in base_client.connected_clients + assert c.base_url == 'ws://foo/engine.io/?transport=websocket&EIO=4' + assert c.sid == '123' + assert c.ping_interval == 1 + assert c.ping_timeout == 2 + assert c.upgrades == [] + assert c.transport() == 'websocket' + assert c.ws == create_connection.return_value + assert len(create_connection.call_args_list) == 1 + assert create_connection.call_args[1] == { + 'header': {}, + 'cookie': None, + 'enable_multithread': True, + 'timeout': 5, + } + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_https_noverify_connection_successful( + self, create_connection + ): + create_connection.return_value.recv.return_value = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + c = client.Client(ssl_verify=False) + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('wss://foo', transports=['websocket']) + time.sleep(0.1) + + c._read_loop_polling.assert_not_called() + c._read_loop_websocket.assert_called_once_with() + c._write_loop.assert_called_once_with() + on_connect.assert_called_once_with() + assert c in base_client.connected_clients + assert c.base_url == 'wss://foo/engine.io/?transport=websocket&EIO=4' + assert c.sid == '123' + assert c.ping_interval == 1 + assert c.ping_timeout == 2 + assert c.upgrades == [] + assert c.transport() == 'websocket' + assert c.ws == create_connection.return_value + assert len(create_connection.call_args_list) == 1 + assert create_connection.call_args[1] == { + 'header': {}, + 'cookie': None, + 'enable_multithread': True, + 'timeout': 5, + 'sslopt': {'cert_reqs': ssl.CERT_NONE}, + } + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_connection_with_cookies(self, create_connection): + create_connection.return_value.recv.return_value = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + http = mock.MagicMock() + http.cookies = [mock.MagicMock(), mock.MagicMock()] + http.cookies[0].name = 'key' + http.cookies[0].value = 'value' + http.cookies[1].name = 'key2' + http.cookies[1].value = 'value2' + http.auth = None + http.proxies = None + http.cert = None + c = client.Client(http_session=http) + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('ws://foo', transports=['websocket']) + time.sleep(0.1) + + assert len(create_connection.call_args_list) == 1 + assert create_connection.call_args[1] == { + 'header': {}, + 'cookie': 'key=value; key2=value2', + 'enable_multithread': True, + 'timeout': 5, + } + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_connection_with_cookie_header(self, create_connection): + create_connection.return_value.recv.return_value = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + http = mock.MagicMock() + http.cookies = [] + http.auth = None + http.proxies = None + http.cert = None + c = client.Client(http_session=http) + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect( + 'ws://foo', + headers={'Foo': 'bar', 'Cookie': 'key=value'}, + transports=['websocket'], + ) + time.sleep(0.1) + + assert len(create_connection.call_args_list) == 1 + assert create_connection.call_args[1] == { + 'header': {'Foo': 'bar'}, + 'cookie': 'key=value', + 'enable_multithread': True, + 'timeout': 5, + } + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_connection_with_cookies_and_headers( + self, create_connection + ): + create_connection.return_value.recv.return_value = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + http = mock.MagicMock() + http.cookies = [mock.MagicMock(), mock.MagicMock()] + http.cookies[0].name = 'key' + http.cookies[0].value = 'value' + http.cookies[1].name = 'key2' + http.cookies[1].value = 'value2' + http.auth = None + http.proxies = None + http.cert = None + c = client.Client(http_session=http) + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect( + 'ws://foo', + headers={'Cookie': 'key3=value3'}, + transports=['websocket'], + ) + time.sleep(0.1) + + assert len(create_connection.call_args_list) == 1 + assert create_connection.call_args[1] == { + 'header': {}, + 'enable_multithread': True, + 'timeout': 5, + 'cookie': 'key=value; key2=value2; key3=value3', + } + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_connection_with_auth(self, create_connection): + create_connection.return_value.recv.return_value = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + http = mock.MagicMock() + http.cookies = [] + http.auth = ('foo', 'bar') + http.proxies = None + http.cert = None + c = client.Client(http_session=http) + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('ws://foo', transports=['websocket']) + + assert len(create_connection.call_args_list) == 1 + assert create_connection.call_args[1] == { + 'header': {'Authorization': 'Basic Zm9vOmJhcg=='}, + 'cookie': '', + 'enable_multithread': True, + 'timeout': 5, + } + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_connection_with_cert(self, create_connection): + create_connection.return_value.recv.return_value = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + http = mock.MagicMock() + http.cookies = [] + http.auth = None + http.proxies = None + http.cert = 'foo.crt' + c = client.Client(http_session=http, ssl_verify=False) + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('ws://foo', transports=['websocket']) + + assert len(create_connection.call_args_list) == 1 + assert create_connection.call_args[1] == { + 'sslopt': {'cert_reqs': ssl.CERT_NONE, 'certfile': 'foo.crt'}, + 'header': {}, + 'cookie': '', + 'enable_multithread': True, + 'timeout': 5, + } + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_connection_with_cert_and_key(self, create_connection): + create_connection.return_value.recv.return_value = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + http = mock.MagicMock() + http.cookies = [] + http.auth = None + http.proxies = None + http.cert = ('foo.crt', 'key.pem') + c = client.Client(http_session=http) + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('ws://foo', transports=['websocket']) + + assert len(create_connection.call_args_list) == 1 + assert create_connection.call_args[1] == { + 'sslopt': {'certfile': 'foo.crt', 'keyfile': 'key.pem'}, + 'header': {}, + 'cookie': '', + 'enable_multithread': True, + 'timeout': 5, + } + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_connection_verify_with_cert_and_key( + self, create_connection + ): + create_connection.return_value.recv.return_value = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + http = mock.MagicMock() + http.cookies = [] + http.auth = None + http.proxies = None + http.cert = ('foo.crt', 'key.pem') + http.verify = 'ca-bundle.crt' + c = client.Client(http_session=http) + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('ws://foo', transports=['websocket']) + + assert len(create_connection.call_args_list) == 1 + assert create_connection.call_args[1] == { + 'sslopt': { + 'certfile': 'foo.crt', + 'keyfile': 'key.pem', + 'ca_certs': 'ca-bundle.crt' + }, + 'header': {}, + 'cookie': '', + 'enable_multithread': True, + 'timeout': 5, + } + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_connection_with_proxies(self, create_connection): + all_urls = [ + 'ws://foo', + 'ws://foo', + 'ws://foo', + 'ws://foo', + 'ws://foo', + 'wss://foo', + 'wss://foo', + ] + all_proxies = [ + {'http': 'foo.com:1234'}, + {'https': 'foo.com:1234'}, + {'http': 'foo.com:1234', 'ws': 'bar.com:4321'}, + {}, + {'http': 'user:pass@foo.com:1234'}, + {'https': 'foo.com:1234'}, + {'https': 'foo.com:1234', 'wss': 'bar.com:4321'}, + ] + all_results = [ + ('foo.com', 1234, None), + None, + ('bar.com', 4321, None), + None, + ('foo.com', 1234, ('user', 'pass')), + ('foo.com', 1234, None), + ('bar.com', 4321, None), + ] + for url, proxies, results in zip(all_urls, all_proxies, all_results): + create_connection.reset_mock() + create_connection.return_value.recv.return_value = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + http = mock.MagicMock() + http.cookies = [] + http.auth = None + http.proxies = proxies + http.cert = None + c = client.Client(http_session=http) + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect(url, transports=['websocket']) + + assert len(create_connection.call_args_list) == 1 + expected_results = { + 'header': {}, + 'cookie': '', + 'enable_multithread': True, + 'timeout': 5, + } + if results: + expected_results.update({ + 'http_proxy_host': results[0], + 'http_proxy_port': results[1], + 'http_proxy_auth': results[2]}) + assert create_connection.call_args[1] == expected_results + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_connection_without_verify(self, create_connection): + create_connection.return_value.recv.return_value = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + http = mock.MagicMock() + http.cookies = [] + http.auth = None + http.proxies = None + http.cert = None + http.verify = False + c = client.Client(http_session=http) + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('ws://foo', transports=['websocket']) + + assert len(create_connection.call_args_list) == 1 + assert create_connection.call_args[1] == { + 'sslopt': {"cert_reqs": ssl.CERT_NONE}, + 'header': {}, + 'cookie': '', + 'enable_multithread': True, + 'timeout': 5, + } + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_connection_with_verify(self, create_connection): + create_connection.return_value.recv.return_value = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + http = mock.MagicMock() + http.cookies = [] + http.auth = None + http.proxies = None + http.cert = None + http.verify = 'ca-bundle.crt' + c = client.Client(http_session=http) + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + c.connect('ws://foo', transports=['websocket']) + + assert len(create_connection.call_args_list) == 1 + assert create_connection.call_args[1] == { + 'sslopt': {'ca_certs': 'ca-bundle.crt'}, + 'header': {}, + 'cookie': '', + 'enable_multithread': True, + 'timeout': 5, + } + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_upgrade_no_pong(self, create_connection): + create_connection.return_value.recv.return_value = packet.Packet( + packet.OPEN, + { + 'sid': '123', + 'upgrades': [], + 'pingInterval': 1000, + 'pingTimeout': 2000, + }, + ).encode() + c = client.Client() + c.sid = '123' + c.current_transport = 'polling' + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + assert not c.connect('ws://foo', transports=['websocket']) + + c._read_loop_polling.assert_not_called() + c._read_loop_websocket.assert_not_called() + c._write_loop.assert_not_called() + on_connect.assert_not_called() + assert c.transport() == 'polling' + + @mock.patch('engineio.client.websocket.create_connection') + def test_websocket_upgrade_successful(self, create_connection): + create_connection.return_value.recv.return_value = packet.Packet( + packet.PONG, 'probe').encode() + c = client.Client() + c.ping_interval = 1 + c.ping_timeout = 2 + c.sid = '123' + c.base_url = 'http://foo' + c.current_transport = 'polling' + c._read_loop_polling = mock.MagicMock() + c._read_loop_websocket = mock.MagicMock() + c._write_loop = mock.MagicMock() + on_connect = mock.MagicMock() + c.on('connect', on_connect) + assert c.connect('ws://foo', transports=['websocket']) + time.sleep(0.1) + + c._read_loop_polling.assert_not_called() + c._read_loop_websocket.assert_called_once_with() + c._write_loop.assert_called_once_with() + on_connect.assert_not_called() # was called by polling + assert c not in base_client.connected_clients # was added by polling + assert c.base_url == 'http://foo' # not changed + assert c.sid == '123' # not changed + assert c.transport() == 'websocket' + assert c.ws == create_connection.return_value + assert create_connection.return_value.send.call_args_list[0] == ( + (packet.Packet(packet.PING, 'probe').encode(),), + ) # ping + assert create_connection.return_value.send.call_args_list[1] == ( + (packet.Packet(packet.UPGRADE).encode(),), + ) # upgrade + + def test_receive_unknown_packet(self): + c = client.Client() + c._receive_packet(packet.Packet(encoded_packet='9')) + # should be ignored + + def test_receive_noop_packet(self): + c = client.Client() + c._receive_packet(packet.Packet(packet.NOOP)) + # should be ignored + + def test_receive_ping_packet(self): + c = client.Client() + c._send_packet = mock.MagicMock() + c._receive_packet(packet.Packet(packet.PING)) + assert c._send_packet.call_args_list[0][0][0].encode() == '3' # PONG + + def test_receive_message_packet(self): + c = client.Client() + c._trigger_event = mock.MagicMock() + c._receive_packet(packet.Packet(packet.MESSAGE, {'foo': 'bar'})) + c._trigger_event.assert_called_once_with( + 'message', {'foo': 'bar'}, run_async=True + ) + + def test_receive_close_packet(self): + c = client.Client() + c.disconnect = mock.MagicMock() + c._receive_packet(packet.Packet(packet.CLOSE)) + c.disconnect.assert_called_once_with(abort=True) + + def test_send_packet_disconnected(self): + c = client.Client() + c.queue = c.create_queue() + c.state = 'disconnected' + c._send_packet(packet.Packet(packet.NOOP)) + assert c.queue.empty() + + def test_send_packet(self): + c = client.Client() + c.queue = c.create_queue() + c.state = 'connected' + c._send_packet(packet.Packet(packet.NOOP)) + assert not c.queue.empty() + pkt = c.queue.get() + assert pkt.packet_type == packet.NOOP + + def test_trigger_event(self): + c = client.Client() + f = {} + + @c.on('connect') + def foo(): + return 'foo' + + @c.on('message') + def bar(data): + f['bar'] = data + return 'bar' + + r = c._trigger_event('connect', run_async=False) + assert r == 'foo' + r = c._trigger_event('message', 123, run_async=True) + r.join() + assert f['bar'] == 123 + r = c._trigger_event('message', 321) + assert r == 'bar' + + def test_trigger_unknown_event(self): + c = client.Client() + c._trigger_event('connect', run_async=False) + c._trigger_event('message', 123, run_async=True) + # should do nothing + + def test_trigger_event_error(self): + c = client.Client() + + @c.on('connect') + def foo(): + return 1 / 0 + + @c.on('message') + def bar(data): + return 1 / 0 + + r = c._trigger_event('connect', run_async=False) + assert r is None + r = c._trigger_event('message', 123, run_async=False) + assert r is None + + def test_engineio_url(self): + c = client.Client() + assert ( + c._get_engineio_url('http://foo', 'bar', 'polling') + == 'http://foo/bar/?transport=polling&EIO=4' + ) + assert ( + c._get_engineio_url('http://foo', 'bar', 'websocket') + == 'ws://foo/bar/?transport=websocket&EIO=4' + ) + assert ( + c._get_engineio_url('ws://foo', 'bar', 'polling') + == 'http://foo/bar/?transport=polling&EIO=4' + ) + assert ( + c._get_engineio_url('ws://foo', 'bar', 'websocket') + == 'ws://foo/bar/?transport=websocket&EIO=4' + ) + assert ( + c._get_engineio_url('https://foo', 'bar', 'polling') + == 'https://foo/bar/?transport=polling&EIO=4' + ) + assert ( + c._get_engineio_url('https://foo', 'bar', 'websocket') + == 'wss://foo/bar/?transport=websocket&EIO=4' + ) + assert ( + c._get_engineio_url('http://foo?baz=1', 'bar', 'polling') + == 'http://foo/bar/?baz=1&transport=polling&EIO=4' + ) + assert ( + c._get_engineio_url('http://foo#baz', 'bar', 'polling') + == 'http://foo/bar/?transport=polling&EIO=4' + ) + + def test_read_loop_polling_disconnected(self): + c = client.Client() + c.state = 'disconnected' + c._trigger_event = mock.MagicMock() + c.write_loop_task = mock.MagicMock() + c._read_loop_polling() + c.write_loop_task.join.assert_called_once_with() + c._trigger_event.assert_not_called() + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_read_loop_polling_no_response(self, _time): + c = client.Client() + c.ping_interval = 25 + c.ping_timeout = 5 + c.state = 'connected' + c.base_url = 'http://foo' + c.queue = mock.MagicMock() + c._send_request = mock.MagicMock(return_value=None) + c._trigger_event = mock.MagicMock() + c.write_loop_task = mock.MagicMock() + c._read_loop_polling() + assert c.state == 'disconnected' + c.queue.put.assert_called_once_with(None) + c.write_loop_task.join.assert_called_once_with() + c._send_request.assert_called_once_with( + 'GET', 'http://foo&t=123.456', timeout=30 + ) + c._trigger_event.assert_called_once_with('disconnect', run_async=False) + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_read_loop_polling_bad_status(self, _time): + c = client.Client() + c.ping_interval = 25 + c.ping_timeout = 5 + c.state = 'connected' + c.base_url = 'http://foo' + c.queue = mock.MagicMock() + c._send_request = mock.MagicMock() + c._send_request.return_value.status_code = 400 + c.write_loop_task = mock.MagicMock() + c._read_loop_polling() + assert c.state == 'disconnected' + c.queue.put.assert_called_once_with(None) + c.write_loop_task.join.assert_called_once_with() + c._send_request.assert_called_once_with( + 'GET', 'http://foo&t=123.456', timeout=30 + ) + + @mock.patch('engineio.client.time.time', return_value=123.456) + def test_read_loop_polling_bad_packet(self, _time): + c = client.Client() + c.ping_interval = 25 + c.ping_timeout = 60 + c.state = 'connected' + c.base_url = 'http://foo' + c.queue = mock.MagicMock() + c._send_request = mock.MagicMock() + c._send_request.return_value.status_code = 200 + c._send_request.return_value.content = b'foo' + c.write_loop_task = mock.MagicMock() + c._read_loop_polling() + assert c.state == 'disconnected' + c.queue.put.assert_called_once_with(None) + c.write_loop_task.join.assert_called_once_with() + c._send_request.assert_called_once_with( + 'GET', 'http://foo&t=123.456', timeout=65 + ) + + def test_read_loop_polling(self): + c = client.Client() + c.ping_interval = 25 + c.ping_timeout = 5 + c.state = 'connected' + c.base_url = 'http://foo' + c.queue = mock.MagicMock() + c._send_request = mock.MagicMock() + c._send_request.side_effect = [ + mock.MagicMock( + status_code=200, + content=payload.Payload( + packets=[ + packet.Packet(packet.PING), + packet.Packet(packet.NOOP), + ] + ).encode().encode('utf-8'), + ), + None, + ] + c.write_loop_task = mock.MagicMock() + c._receive_packet = mock.MagicMock() + c._read_loop_polling() + assert c.state == 'disconnected' + c.queue.put.assert_called_once_with(None) + assert c._send_request.call_count == 2 + assert c._receive_packet.call_count == 2 + assert c._receive_packet.call_args_list[0][0][0].encode() == '2' + assert c._receive_packet.call_args_list[1][0][0].encode() == '6' + + def test_read_loop_websocket_disconnected(self): + c = client.Client() + c.state = 'disconnected' + c.write_loop_task = mock.MagicMock() + c._read_loop_websocket() + c.write_loop_task.join.assert_called_once_with() + + def test_read_loop_websocket_timeout(self): + c = client.Client() + c.state = 'connected' + c.queue = mock.MagicMock() + c.ws = mock.MagicMock() + c.ws.recv.side_effect = websocket.WebSocketTimeoutException + c.write_loop_task = mock.MagicMock() + c._read_loop_websocket() + assert c.state == 'disconnected' + c.queue.put.assert_called_once_with(None) + c.write_loop_task.join.assert_called_once_with() + + def test_read_loop_websocket_no_response(self): + c = client.Client() + c.state = 'connected' + c.queue = mock.MagicMock() + c.ws = mock.MagicMock() + c.ws.recv.side_effect = websocket.WebSocketConnectionClosedException + c.write_loop_task = mock.MagicMock() + c._read_loop_websocket() + assert c.state == 'disconnected' + c.queue.put.assert_called_once_with(None) + c.write_loop_task.join.assert_called_once_with() + + def test_read_loop_websocket_unexpected_error(self): + c = client.Client() + c.state = 'connected' + c.queue = mock.MagicMock() + c.ws = mock.MagicMock() + c.ws.recv.side_effect = ValueError + c.write_loop_task = mock.MagicMock() + c._read_loop_websocket() + assert c.state == 'disconnected' + c.queue.put.assert_called_once_with(None) + c.write_loop_task.join.assert_called_once_with() + + def test_read_loop_websocket(self): + c = client.Client() + c.ping_interval = 1 + c.ping_timeout = 2 + c.state = 'connected' + c.queue = mock.MagicMock() + c.ws = mock.MagicMock() + c.ws.recv.side_effect = [ + packet.Packet(packet.PING).encode(), + ValueError, + ] + c.write_loop_task = mock.MagicMock() + c._receive_packet = mock.MagicMock() + c._read_loop_websocket() + assert c.state == 'disconnected' + c.queue.put.assert_called_once_with(None) + c.write_loop_task.join.assert_called_once_with() + assert c._receive_packet.call_args_list[0][0][0].encode() == '2' + + def test_write_loop_disconnected(self): + c = client.Client() + c.state = 'disconnected' + c._write_loop() + # should not block + + def test_write_loop_no_packets(self): + c = client.Client() + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.queue = mock.MagicMock() + c.queue.get.return_value = None + c._write_loop() + c.queue.task_done.assert_called_once_with() + c.queue.get.assert_called_once_with(timeout=7) + + def test_write_loop_empty_queue(self): + c = client.Client() + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get.side_effect = RuntimeError + c._write_loop() + c.queue.get.assert_called_once_with(timeout=7) + + def test_write_loop_polling_one_packet(self): + c = client.Client() + c.base_url = 'http://foo' + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get.side_effect = [ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + RuntimeError, + RuntimeError, + ] + c._send_request = mock.MagicMock() + c._send_request.return_value.status_code = 200 + c._write_loop() + assert c.queue.task_done.call_count == 1 + p = payload.Payload( + packets=[packet.Packet(packet.MESSAGE, {'foo': 'bar'})] + ) + c._send_request.assert_called_once_with( + 'POST', + 'http://foo', + body=p.encode(), + headers={'Content-Type': 'text/plain'}, + timeout=5, + ) + + def test_write_loop_polling_three_packets(self): + c = client.Client() + c.base_url = 'http://foo' + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get.side_effect = [ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + packet.Packet(packet.PING), + packet.Packet(packet.NOOP), + RuntimeError, + RuntimeError, + ] + c._send_request = mock.MagicMock() + c._send_request.return_value.status_code = 200 + c._write_loop() + assert c.queue.task_done.call_count == 3 + p = payload.Payload( + packets=[ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + packet.Packet(packet.PING), + packet.Packet(packet.NOOP), + ] + ) + c._send_request.assert_called_once_with( + 'POST', + 'http://foo', + body=p.encode(), + headers={'Content-Type': 'text/plain'}, + timeout=5, + ) + + def test_write_loop_polling_two_packets_done(self): + c = client.Client() + c.base_url = 'http://foo' + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get.side_effect = [ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + packet.Packet(packet.PING), + None, + RuntimeError, + ] + c._send_request = mock.MagicMock() + c._send_request.return_value.status_code = 200 + c._write_loop() + assert c.queue.task_done.call_count == 3 + p = payload.Payload( + packets=[ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + packet.Packet(packet.PING), + ] + ) + c._send_request.assert_called_once_with( + 'POST', + 'http://foo', + body=p.encode(), + headers={'Content-Type': 'text/plain'}, + timeout=5, + ) + assert c.state == 'connected' + + def test_write_loop_polling_bad_connection(self): + c = client.Client() + c.base_url = 'http://foo' + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get.side_effect = [ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + RuntimeError, + ] + c._send_request = mock.MagicMock() + c._send_request.return_value = None + c._write_loop() + assert c.queue.task_done.call_count == 1 + p = payload.Payload( + packets=[packet.Packet(packet.MESSAGE, {'foo': 'bar'})] + ) + c._send_request.assert_called_once_with( + 'POST', + 'http://foo', + body=p.encode(), + headers={'Content-Type': 'text/plain'}, + timeout=5, + ) + assert c.state == 'connected' + + def test_write_loop_polling_bad_status(self): + c = client.Client() + c.base_url = 'http://foo' + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'polling' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get.side_effect = [ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + RuntimeError, + ] + c._send_request = mock.MagicMock() + c._send_request.return_value.status_code = 500 + c._write_loop() + assert c.queue.task_done.call_count == 1 + p = payload.Payload( + packets=[packet.Packet(packet.MESSAGE, {'foo': 'bar'})] + ) + c._send_request.assert_called_once_with( + 'POST', + 'http://foo', + body=p.encode(), + headers={'Content-Type': 'text/plain'}, + timeout=5, + ) + assert c.state == 'connected' + assert c.write_loop_task is None + + def test_write_loop_websocket_one_packet(self): + c = client.Client() + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'websocket' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get.side_effect = [ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + RuntimeError, + RuntimeError, + ] + c.ws = mock.MagicMock() + c._write_loop() + assert c.queue.task_done.call_count == 1 + assert c.ws.send.call_count == 1 + assert c.ws.send_binary.call_count == 0 + c.ws.send.assert_called_once_with('4{"foo":"bar"}') + + def test_write_loop_websocket_three_packets(self): + c = client.Client() + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'websocket' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get.side_effect = [ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + packet.Packet(packet.PING), + packet.Packet(packet.NOOP), + RuntimeError, + RuntimeError, + ] + c.ws = mock.MagicMock() + c._write_loop() + assert c.queue.task_done.call_count == 3 + assert c.ws.send.call_count == 3 + assert c.ws.send_binary.call_count == 0 + assert c.ws.send.call_args_list[0][0][0] == '4{"foo":"bar"}' + assert c.ws.send.call_args_list[1][0][0] == '2' + assert c.ws.send.call_args_list[2][0][0] == '6' + + def test_write_loop_websocket_one_packet_binary(self): + c = client.Client() + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'websocket' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get.side_effect = [ + packet.Packet(packet.MESSAGE, b'foo'), + RuntimeError, + RuntimeError, + ] + c.ws = mock.MagicMock() + c._write_loop() + assert c.queue.task_done.call_count == 1 + assert c.ws.send.call_count == 0 + assert c.ws.send_binary.call_count == 1 + c.ws.send_binary.assert_called_once_with(b'foo') + + def test_write_loop_websocket_bad_connection(self): + c = client.Client() + c.state = 'connected' + c.ping_interval = 1 + c.ping_timeout = 2 + c.current_transport = 'websocket' + c.queue = mock.MagicMock() + c.queue.Empty = RuntimeError + c.queue.get.side_effect = [ + packet.Packet(packet.MESSAGE, {'foo': 'bar'}), + RuntimeError, + RuntimeError, + ] + c.ws = mock.MagicMock() + c.ws.send.side_effect = websocket.WebSocketConnectionClosedException + c._write_loop() + assert c.state == 'connected' + + @mock.patch('engineio.base_client.original_signal_handler') + def test_signal_handler(self, original_handler): + clients = [mock.MagicMock(), mock.MagicMock()] + base_client.connected_clients = clients[:] + base_client.connected_clients[0].is_asyncio_based.return_value = False + base_client.connected_clients[1].is_asyncio_based.return_value = True + base_client.signal_handler('sig', 'frame') + clients[0].disconnect.assert_called_once_with() + clients[1].disconnect.assert_not_called() diff --git a/tests/common/test_middleware.py b/tests/common/test_middleware.py new file mode 100644 index 0000000..436b19a --- /dev/null +++ b/tests/common/test_middleware.py @@ -0,0 +1,197 @@ +import os +import unittest +from unittest import mock + +import engineio + + +class TestWSGIApp(unittest.TestCase): + def test_wsgi_routing(self): + mock_wsgi_app = mock.MagicMock() + mock_eio_app = 'foo' + m = engineio.WSGIApp(mock_eio_app, mock_wsgi_app) + environ = {'PATH_INFO': '/foo'} + start_response = "foo" + m(environ, start_response) + mock_wsgi_app.assert_called_once_with(environ, start_response) + + def test_eio_routing(self): + mock_wsgi_app = 'foo' + mock_eio_app = mock.Mock() + mock_eio_app.handle_request = mock.MagicMock() + m = engineio.WSGIApp(mock_eio_app, mock_wsgi_app) + environ = {'PATH_INFO': '/engine.io/'} + start_response = "foo" + m(environ, start_response) + mock_eio_app.handle_request.assert_called_once_with( + environ, start_response + ) + + def test_static_files(self): + root_dir = os.path.dirname(__file__) + m = engineio.WSGIApp( + 'foo', + None, + static_files={ + '/': root_dir + '/index.html', + '/foo': { + 'content_type': 'text/plain', + 'filename': root_dir + '/index.html', + }, + '/static': root_dir, + '/static/test/': root_dir + '/', + '/static2/test/': {'filename': root_dir + '/', + 'content_type': 'image/gif'}, + }, + ) + + def check_path(path, status_code, content_type, body): + environ = {'PATH_INFO': path} + start_response = mock.MagicMock() + r = m(environ, start_response) + assert r == [body.encode('utf-8')] + start_response.assert_called_once_with( + status_code, [('Content-Type', content_type)] + ) + + check_path('/', '200 OK', 'text/html', '<html></html>\n') + check_path('/foo', '200 OK', 'text/plain', '<html></html>\n') + check_path('/foo/bar', '404 Not Found', 'text/plain', 'Not Found') + check_path( + '/static/index.html', '200 OK', 'text/html', '<html></html>\n' + ) + check_path( + '/static/foo.bar', '404 Not Found', 'text/plain', 'Not Found' + ) + check_path( + '/static/test/index.html', '200 OK', 'text/html', '<html></html>\n' + ) + check_path('/static/test/', '200 OK', 'text/html', '<html></html>\n') + check_path('/static/test/index.html', '200 OK', 'text/html', + '<html></html>\n') + check_path('/static/test/files/', '200 OK', 'text/html', + '<html>file</html>\n') + check_path('/static/test/files/file.txt', '200 OK', 'text/plain', + 'file\n') + check_path('/static/test/files/x.html', '404 Not Found', 'text/plain', + 'Not Found') + check_path('/static2/test/', '200 OK', 'image/gif', '<html></html>\n') + check_path('/static2/test/index.html', '200 OK', 'image/gif', + '<html></html>\n') + check_path('/static2/test/files/', '200 OK', 'image/gif', + '<html>file</html>\n') + check_path('/static2/test/files/file.txt', '200 OK', 'image/gif', + 'file\n') + check_path('/static2/test/files/x.html', '404 Not Found', 'text/plain', + 'Not Found') + check_path('/bar/foo', '404 Not Found', 'text/plain', 'Not Found') + check_path('', '404 Not Found', 'text/plain', 'Not Found') + + m.static_files[''] = 'index.html' + check_path('/static/test/', '200 OK', 'text/html', '<html></html>\n') + + m.static_files[''] = {'filename': 'index.html'} + check_path('/static/test/', '200 OK', 'text/html', '<html></html>\n') + + m.static_files[''] = { + 'filename': 'index.html', + 'content_type': 'image/gif', + } + check_path('/static/test/', '200 OK', 'image/gif', '<html></html>\n') + + m.static_files[''] = {'filename': 'test.gif'} + check_path('/static/test/', '404 Not Found', 'text/plain', 'Not Found') + + m.static_files = {} + check_path( + '/static/test/index.html', + '404 Not Found', + 'text/plain', + 'Not Found', + ) + + def test_404(self): + mock_wsgi_app = None + mock_eio_app = mock.Mock() + m = engineio.WSGIApp(mock_eio_app, mock_wsgi_app) + environ = {'PATH_INFO': '/foo/bar'} + start_response = mock.MagicMock() + r = m(environ, start_response) + assert r == [b'Not Found'] + start_response.assert_called_once_with( + "404 Not Found", [('Content-Type', 'text/plain')] + ) + + def test_custom_eio_path(self): + mock_wsgi_app = None + mock_eio_app = mock.Mock() + mock_eio_app.handle_request = mock.MagicMock() + m = engineio.WSGIApp(mock_eio_app, mock_wsgi_app, engineio_path='foo') + environ = {'PATH_INFO': '/engine.io/'} + start_response = mock.MagicMock() + r = m(environ, start_response) + assert r == [b'Not Found'] + start_response.assert_called_once_with( + "404 Not Found", [('Content-Type', 'text/plain')] + ) + + environ = {'PATH_INFO': '/foo/'} + m(environ, start_response) + mock_eio_app.handle_request.assert_called_once_with( + environ, start_response + ) + + def test_custom_eio_path_slashes(self): + mock_wsgi_app = None + mock_eio_app = mock.Mock() + mock_eio_app.handle_request = mock.MagicMock() + m = engineio.WSGIApp( + mock_eio_app, mock_wsgi_app, engineio_path='/foo/' + ) + environ = {'PATH_INFO': '/foo/'} + start_response = mock.MagicMock() + m(environ, start_response) + mock_eio_app.handle_request.assert_called_once_with( + environ, start_response + ) + + def test_custom_eio_path_leading_slash(self): + mock_wsgi_app = None + mock_eio_app = mock.Mock() + mock_eio_app.handle_request = mock.MagicMock() + m = engineio.WSGIApp(mock_eio_app, mock_wsgi_app, engineio_path='/foo') + environ = {'PATH_INFO': '/foo/'} + start_response = mock.MagicMock() + m(environ, start_response) + mock_eio_app.handle_request.assert_called_once_with( + environ, start_response + ) + + def test_custom_eio_path_trailing_slash(self): + mock_wsgi_app = None + mock_eio_app = mock.Mock() + mock_eio_app.handle_request = mock.MagicMock() + m = engineio.WSGIApp(mock_eio_app, mock_wsgi_app, engineio_path='foo/') + environ = {'PATH_INFO': '/foo/'} + start_response = mock.MagicMock() + m(environ, start_response) + mock_eio_app.handle_request.assert_called_once_with( + environ, start_response + ) + + def test_gunicorn_socket(self): + mock_wsgi_app = None + mock_eio_app = mock.Mock() + m = engineio.WSGIApp(mock_eio_app, mock_wsgi_app) + environ = {'gunicorn.socket': 123, 'PATH_INFO': '/foo/bar'} + start_response = mock.MagicMock() + m(environ, start_response) + assert 'eventlet.input' in environ + assert environ['eventlet.input'].get_socket() == 123 + + def test_legacy_middleware_class(self): + m = engineio.Middleware('eio', 'wsgi', 'eio_path') + assert m.engineio_app == 'eio' + assert m.wsgi_app == 'wsgi' + assert m.static_files == {} + assert m.engineio_path == '/eio_path/' diff --git a/tests/common/test_packet.py b/tests/common/test_packet.py new file mode 100644 index 0000000..e81a30d --- /dev/null +++ b/tests/common/test_packet.py @@ -0,0 +1,144 @@ +import unittest + +import pytest + +from engineio import packet + + +class TestPacket(unittest.TestCase): + def test_encode_default_packet(self): + pkt = packet.Packet() + assert pkt.packet_type == packet.NOOP + assert pkt.data is None + assert not pkt.binary + assert pkt.encode() == '6' + + def test_decode_default_packet(self): + pkt = packet.Packet(encoded_packet='6') + assert pkt.encode() == '6' + + def test_encode_text_packet(self): + data = 'text' + pkt = packet.Packet(packet.MESSAGE, data=data) + assert pkt.packet_type == packet.MESSAGE + assert pkt.data == data + assert not pkt.binary + assert pkt.encode() == '4text' + + def test_decode_text_packet(self): + pkt = packet.Packet(encoded_packet=b'4text') + assert pkt.encode() == b'4text' + + def test_encode_empty_text_packet(self): + data = '' + pkt = packet.Packet(packet.MESSAGE, data=data) + assert pkt.packet_type == packet.MESSAGE + assert pkt.data == data + assert not pkt.binary + assert pkt.encode() == '4' + + def test_decode_empty_text_packet(self): + pkt = packet.Packet(encoded_packet=b'4') + assert pkt.encode() == b'4' + + def test_encode_binary_packet(self): + pkt = packet.Packet(packet.MESSAGE, data=b'\x01\x02\x03') + assert pkt.packet_type == packet.MESSAGE + assert pkt.data == b'\x01\x02\x03' + assert pkt.binary + assert pkt.encode() == b'\x01\x02\x03' + + def test_encode_binary_bytearray_packet(self): + pkt = packet.Packet(packet.MESSAGE, data=bytearray(b'\x01\x02\x03')) + assert pkt.packet_type == packet.MESSAGE + assert pkt.data == b'\x01\x02\x03' + assert pkt.binary + assert pkt.encode() == b'\x01\x02\x03' + + def test_encode_binary_b64_packet(self): + pkt = packet.Packet(packet.MESSAGE, data=b'\x01\x02\x03\x04') + assert pkt.packet_type == packet.MESSAGE + assert pkt.data == b'\x01\x02\x03\x04' + assert pkt.binary + assert pkt.encode(b64=True) == 'bAQIDBA==' + + def test_encode_empty_binary_packet(self): + pkt = packet.Packet(packet.MESSAGE, data=b'') + assert pkt.packet_type == packet.MESSAGE + assert pkt.data == b'' + assert pkt.binary + assert pkt.encode() == b'' + + def test_decode_binary_packet(self): + pkt = packet.Packet(encoded_packet=b'\x04\x01\x02\x03') + assert pkt.encode() == b'\x04\x01\x02\x03' + + def test_decode_binary_bytearray_packet(self): + pkt = packet.Packet(encoded_packet=bytearray(b'\x04\x01\x02\x03')) + assert pkt.encode() == b'\x04\x01\x02\x03' + + def test_decode_binary_b64_packet(self): + pkt = packet.Packet(encoded_packet='bBAECAw==') + assert pkt.encode() == b'\x04\x01\x02\x03' + + def test_decode_empty_binary_packet(self): + pkt = packet.Packet(encoded_packet=b'') + assert pkt.encode() == b'' + + def test_encode_json_packet(self): + pkt = packet.Packet(packet.MESSAGE, data={'a': 123, 'b': '456'}) + assert pkt.packet_type == packet.MESSAGE + assert pkt.data == {'a': 123, 'b': '456'} + assert not pkt.binary + assert pkt.encode() in [ + '4{"a":123,"b":"456"}', + '4{"b":"456","a":123}', + ] + + def test_decode_json_packet(self): + pkt = packet.Packet(encoded_packet='4{"a":123,"b":"456"}') + assert pkt.encode() in [ + '4{"a":123,"b":"456"}', + '4{"b":"456","a":123}', + ] + + def test_decode_json_packet_long_int(self): + pkt = packet.Packet(encoded_packet='4{"a":' + '1' * 100 + '}') + assert pkt.packet_type == packet.MESSAGE + assert pkt.data == {'a': int('1' * 100)} + pkt = packet.Packet(encoded_packet='4{"a":' + '1' * 101 + '}') + assert pkt.packet_type == packet.MESSAGE + assert pkt.data == '{"a":' + '1' * 101 + '}' + + def test_encode_number_packet(self): + pkt = packet.Packet(packet.MESSAGE, data=123) + assert pkt.packet_type == packet.MESSAGE + assert pkt.data == 123 + assert not pkt.binary + assert pkt.encode() == '4123' + + def test_decode_number_packet(self): + pkt = packet.Packet(encoded_packet='4123') + assert pkt.packet_type == packet.MESSAGE + # integer payloads are parsed as strings, see + # https://github.com/miguelgrinberg/python-engineio/issues/75 + # for background on this decision + assert pkt.data == '123' + assert not pkt.binary + assert pkt.encode() == '4123' + + def test_binary_non_message_packet(self): + with pytest.raises(ValueError): + packet.Packet(packet.NOOP, b'\x01\x02\x03') + + def test_decode_invalid_empty_text_packet(self): + with pytest.raises(ValueError): + packet.Packet(encoded_packet='') + + def test_encode_cache(self): + pkt = packet.Packet(packet.MESSAGE, data=123) + assert pkt.encode() == '4123' + pkt.data = 456 + assert pkt.encode() == '4123' + pkt.encode_cache = None + assert pkt.encode() == '4456' diff --git a/tests/common/test_payload.py b/tests/common/test_payload.py new file mode 100644 index 0000000..06c3289 --- /dev/null +++ b/tests/common/test_payload.py @@ -0,0 +1,68 @@ +import unittest + +import pytest + +from engineio import packet +from engineio import payload + + +class TestPayload(unittest.TestCase): + def test_encode_empty_payload(self): + p = payload.Payload() + assert p.packets == [] + assert p.encode() == '' + + def test_decode_empty_payload(self): + p = payload.Payload(encoded_payload='') + assert p.encode() == '' + + def test_encode_payload_text(self): + pkt = packet.Packet(packet.MESSAGE, data='abc') + p = payload.Payload([pkt]) + assert p.packets == [pkt] + assert p.encode() == '4abc' + + def test_encode_payload_text_multiple(self): + pkt = packet.Packet(packet.MESSAGE, data='abc') + pkt2 = packet.Packet(packet.MESSAGE, data='def') + p = payload.Payload([pkt, pkt2]) + assert p.packets == [pkt, pkt2] + assert p.encode() == '4abc\x1e4def' + + def test_encode_payload_binary(self): + pkt = packet.Packet(packet.MESSAGE, data=b'\x00\x01\x02') + p = payload.Payload([pkt]) + assert p.packets == [pkt] + assert p.encode() == 'bAAEC' + + def test_encode_payload_binary_multiple(self): + pkt = packet.Packet(packet.MESSAGE, data=b'\x00\x01\x02') + pkt2 = packet.Packet(packet.MESSAGE, data=b'\x03\x04\x05\x06') + p = payload.Payload([pkt, pkt2]) + assert p.packets == [pkt, pkt2] + assert p.encode() == 'bAAEC\x1ebAwQFBg==' + + def test_encode_payload_text_binary_multiple(self): + pkt = packet.Packet(packet.MESSAGE, data='abc') + pkt2 = packet.Packet(packet.MESSAGE, data=b'\x03\x04\x05\x06') + p = payload.Payload([pkt, pkt2, pkt2, pkt]) + assert p.packets == [pkt, pkt2, pkt2, pkt] + assert p.encode() == '4abc\x1ebAwQFBg==\x1ebAwQFBg==\x1e4abc' + + def test_encode_jsonp_payload(self): + pkt = packet.Packet(packet.MESSAGE, data='abc') + p = payload.Payload([pkt]) + assert p.packets == [pkt] + assert p.encode(jsonp_index=233) == '___eio[233]("4abc");' + + def test_decode_jsonp_payload(self): + p = payload.Payload(encoded_payload='d=4abc') + assert p.encode() == '4abc' + + def test_decode_invalid_payload(self): + with pytest.raises(ValueError): + payload.Payload(encoded_payload='bad payload') + + def test_decode_multi_payload_with_too_many_packets(self): + with pytest.raises(ValueError): + payload.Payload(encoded_payload='4abc\x1e4def\x1e' * 9 + '6') diff --git a/tests/common/test_server.py b/tests/common/test_server.py new file mode 100644 index 0000000..adddea1 --- /dev/null +++ b/tests/common/test_server.py @@ -0,0 +1,1159 @@ +import gzip +import importlib +import io +import logging +import sys +import time +import unittest +from unittest import mock +import zlib + +import pytest + +from engineio import exceptions +from engineio import json +from engineio import packet +from engineio import payload +from engineio import server + +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 kwargs.keys(): + assert getattr(s, arg) == kwargs[arg] + assert s.ping_interval_grace_period == 0 + + 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): + sys.modules['simple_websocket'] = mock.MagicMock() + s = server.Server(async_mode='threading') + assert s.async_mode == 'threading' + + from engineio.async_drivers import threading as async_threading + import queue + + assert s._async['thread'] == async_threading.DaemonThread + assert s._async['queue'] == queue.Queue + assert s._async['websocket'] == async_threading.SimpleWebSocketWSGI + del sys.modules['simple_websocket'] + del sys.modules['engineio.async_drivers.threading'] + + def test_async_mode_eventlet(self): + s = server.Server(async_mode='eventlet') + assert s.async_mode == 'eventlet' + + from eventlet import queue + from engineio.async_drivers import eventlet as async_eventlet + + assert s._async['thread'] == async_eventlet.EventletThread + 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_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(async_mode='threading') + 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 mock_sockets.values(): + 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=4&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=4&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=4'} + 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', + 'text/plain; charset=UTF-8', + ) in start_response.call_args[0][1] + assert len(r) == 1 + packets = payload.Payload(encoded_payload=r[0].decode('utf-8')).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=4'} + start_response = mock.MagicMock() + r = s.handle_request(environ, start_response) + packets = payload.Payload(encoded_payload=r[0].decode('utf-8')).packets + assert packets[0].data['upgrades'] == [] + + def test_connect_bad_eio_version(self): + s = server.Server() + environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=1'} + 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_custom_ping_times(self): + s = server.Server(ping_timeout=123, ping_interval=456) + environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=4'} + start_response = mock.MagicMock() + r = s.handle_request(environ, start_response) + packets = payload.Payload(encoded_payload=r[0].decode('utf-8')).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=4'} + 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=4&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_http_upgrade_case_insensitive(self, Socket): + s = server.Server() + s.generate_id = mock.MagicMock(return_value='123') + environ = { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'EIO=4&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=4&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=4&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=4&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=4'} + 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=4', + '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_allowed_origin_with_callable(self): + def cors(origin): + return origin == 'a' + + s = server.Server(cors_allowed_origins=cors) + environ = { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'EIO=4', + '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 + + environ['HTTP_ORIGIN'] = 'b' + s.handle_request(environ, start_response) + assert start_response.call_args[0][0] == '400 BAD REQUEST' + + def test_connect_cors_not_allowed_origin(self): + s = server.Server(cors_allowed_origins=['a', 'b']) + environ = { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'EIO=4', + '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=4', + '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=4', + '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=4', + '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=4', + '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=4', + 'wsgi.url_scheme': 'http', + 'HTTP_HOST': 'foo', + 'HTTP_ORIGIN': 'https://foo', + 'HTTP_X_FORWARDED_PROTO': 'https, ftp', + } + 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://foo') in headers + + def test_connect_cors_headers_default_origin_proxy_server2(self): + s = server.Server() + environ = { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'EIO=4', + '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=4'} + 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=4'} + 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=4', + '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=4', + '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=4'} + 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=4'} + 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=4'} + 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=4'} + 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=4'} + 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=4'} + 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=4'} + 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=4&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=4&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=4&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].decode('utf-8')).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=4&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=4&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=4&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=4&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=4&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 = io.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=4&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=4&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=4&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=4&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=4&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=4&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=4'} + 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=4'} + 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=4'} + 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() == '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(async_mode='threading') + 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(async_mode='threading', monitor_clients=True) + s._service_task = mock.MagicMock() + environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=4'} + start_response = mock.MagicMock() + s.handle_request(environ, start_response) + for _ in range(3): + if s._service_task.call_count > 0: + break + time.sleep(0.05) + s._service_task.assert_called_once_with() + + def test_shutdown(self): + s = server.Server(async_mode='threading', monitor_clients=True) + environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'EIO=4'} + start_response = mock.MagicMock() + s.handle_request(environ, start_response) + assert s.service_task_handle is not None + s.shutdown() + assert s.service_task_handle is None + + def test_transports_invalid(self): + with pytest.raises(ValueError): + server.Server(transports='invalid') + with pytest.raises(ValueError): + server.Server(transports=['invalid', 'foo']) + + def test_transports_disallowed(self): + s = server.Server(transports='websocket') + environ = { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'transport=polling', + } + start_response = mock.MagicMock() + s.handle_request(environ, start_response) + assert start_response.call_args[0][0] == '400 BAD REQUEST' diff --git a/tests/common/test_socket.py b/tests/common/test_socket.py new file mode 100644 index 0000000..41b7424 --- /dev/null +++ b/tests/common/test_socket.py @@ -0,0 +1,526 @@ +import io +import time +import unittest +from unittest import mock + +import pytest + +from engineio import exceptions +from engineio import packet +from engineio import payload +from engineio import socket + + +class TestSocket(unittest.TestCase): + def setUp(self): + self.bg_tasks = [] + + def _get_mock_server(self): + mock_server = mock.Mock() + mock_server.ping_timeout = 0.2 + mock_server.ping_interval = 0.2 + mock_server.ping_interval_grace_period = 0.001 + mock_server.async_handlers = True + mock_server.max_http_buffer_size = 128 + + try: + import queue + except ImportError: + import Queue as queue + import threading + + mock_server._async = { + 'threading': threading.Thread, + 'queue': queue.Queue, + 'websocket': None, + } + + def bg_task(target, *args, **kwargs): + th = threading.Thread(target=target, args=args, kwargs=kwargs) + self.bg_tasks.append(th) + th.start() + return th + + def create_queue(*args, **kwargs): + return queue.Queue(*args, **kwargs) + + mock_server.start_background_task = bg_task + mock_server.create_queue = create_queue + mock_server.get_queue_empty_exception.return_value = queue.Empty + return mock_server + + def _join_bg_tasks(self): + for task in self.bg_tasks: + task.join() + + def test_create(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + assert s.server == mock_server + assert s.sid == 'sid' + assert not s.upgraded + assert not s.closed + assert hasattr(s.queue, 'get') + assert hasattr(s.queue, 'put') + assert hasattr(s.queue, 'task_done') + assert hasattr(s.queue, 'join') + + def test_empty_poll(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + with pytest.raises(exceptions.QueueEmpty): + s.poll() + + def test_poll(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + pkt1 = packet.Packet(packet.MESSAGE, data='hello') + pkt2 = packet.Packet(packet.MESSAGE, data='bye') + s.send(pkt1) + s.send(pkt2) + assert s.poll() == [pkt1, pkt2] + + def test_poll_none(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.queue.put(None) + assert s.poll() == [] + + def test_poll_none_after_packet(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + pkt = packet.Packet(packet.MESSAGE, data='hello') + s.send(pkt) + s.queue.put(None) + assert s.poll() == [pkt] + assert s.poll() == [] + + def test_schedule_ping(self): + mock_server = self._get_mock_server() + mock_server.ping_interval = 0.01 + s = socket.Socket(mock_server, 'sid') + s.send = mock.MagicMock() + s.schedule_ping() + time.sleep(0.05) + assert s.last_ping is not None + assert s.send.call_args_list[0][0][0].encode() == '2' + + def test_schedule_ping_closed_socket(self): + mock_server = self._get_mock_server() + mock_server.ping_interval = 0.01 + s = socket.Socket(mock_server, 'sid') + s.send = mock.MagicMock() + s.closed = True + s.schedule_ping() + time.sleep(0.05) + assert s.last_ping is None + s.send.assert_not_called() + + def test_pong(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.schedule_ping = mock.MagicMock() + s.receive(packet.Packet(packet.PONG)) + s.schedule_ping.assert_called_once_with() + + def test_message_async_handler(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.receive(packet.Packet(packet.MESSAGE, data='foo')) + mock_server._trigger_event.assert_called_once_with( + 'message', 'sid', 'foo', run_async=True + ) + + def test_message_sync_handler(self): + mock_server = self._get_mock_server() + mock_server.async_handlers = False + s = socket.Socket(mock_server, 'sid') + s.receive(packet.Packet(packet.MESSAGE, data='foo')) + mock_server._trigger_event.assert_called_once_with( + 'message', 'sid', 'foo', run_async=False + ) + + def test_invalid_packet(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + with pytest.raises(exceptions.UnknownPacketError): + s.receive(packet.Packet(packet.OPEN)) + + def test_timeout(self): + mock_server = self._get_mock_server() + mock_server.ping_interval = 6 + mock_server.ping_interval_grace_period = 2 + s = socket.Socket(mock_server, 'sid') + s.last_ping = time.time() - 9 + s.close = mock.MagicMock() + s.send('packet') + s.close.assert_called_once_with(wait=False, abort=False) + + def test_polling_read(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'foo') + pkt1 = packet.Packet(packet.MESSAGE, data='hello') + pkt2 = packet.Packet(packet.MESSAGE, data='bye') + s.send(pkt1) + s.send(pkt2) + environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'sid=foo'} + start_response = mock.MagicMock() + packets = s.handle_get_request(environ, start_response) + assert packets == [pkt1, pkt2] + + def test_polling_read_error(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'foo') + environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'sid=foo'} + start_response = mock.MagicMock() + with pytest.raises(exceptions.QueueEmpty): + s.handle_get_request(environ, start_response) + + def test_polling_write(self): + mock_server = self._get_mock_server() + mock_server.max_http_buffer_size = 1000 + pkt1 = packet.Packet(packet.MESSAGE, data='hello') + pkt2 = packet.Packet(packet.MESSAGE, data='bye') + p = payload.Payload(packets=[pkt1, pkt2]).encode().encode('utf-8') + s = socket.Socket(mock_server, 'foo') + s.receive = mock.MagicMock() + environ = { + 'REQUEST_METHOD': 'POST', + 'QUERY_STRING': 'sid=foo', + 'CONTENT_LENGTH': len(p), + 'wsgi.input': io.BytesIO(p), + } + s.handle_post_request(environ) + assert s.receive.call_count == 2 + + def test_polling_write_too_large(self): + mock_server = self._get_mock_server() + pkt1 = packet.Packet(packet.MESSAGE, data='hello') + pkt2 = packet.Packet(packet.MESSAGE, data='bye') + p = payload.Payload(packets=[pkt1, pkt2]).encode().encode('utf-8') + mock_server.max_http_buffer_size = len(p) - 1 + s = socket.Socket(mock_server, 'foo') + s.receive = mock.MagicMock() + environ = { + 'REQUEST_METHOD': 'POST', + 'QUERY_STRING': 'sid=foo', + 'CONTENT_LENGTH': len(p), + 'wsgi.input': io.BytesIO(p), + } + with pytest.raises(exceptions.ContentTooLongError): + s.handle_post_request(environ) + + def test_upgrade_handshake(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'foo') + s._upgrade_websocket = mock.MagicMock() + environ = { + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': 'sid=foo', + 'HTTP_CONNECTION': 'Foo,Upgrade,Bar', + 'HTTP_UPGRADE': 'websocket', + } + start_response = mock.MagicMock() + s.handle_get_request(environ, start_response) + s._upgrade_websocket.assert_called_once_with(environ, start_response) + + def test_upgrade(self): + mock_server = self._get_mock_server() + mock_server._async['websocket'] = mock.MagicMock() + mock_ws = mock.MagicMock() + mock_server._async['websocket'].return_value = mock_ws + s = socket.Socket(mock_server, 'sid') + s.connected = True + environ = "foo" + start_response = "bar" + s._upgrade_websocket(environ, start_response) + mock_server._async['websocket'].assert_called_once_with( + s._websocket_handler, mock_server + ) + mock_ws.assert_called_once_with(environ, start_response) + + def test_upgrade_twice(self): + mock_server = self._get_mock_server() + mock_server._async['websocket'] = mock.MagicMock() + s = socket.Socket(mock_server, 'sid') + s.connected = True + s.upgraded = True + environ = "foo" + start_response = "bar" + with pytest.raises(IOError): + s._upgrade_websocket(environ, start_response) + + def test_upgrade_packet(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.connected = True + s.receive(packet.Packet(packet.UPGRADE)) + r = s.poll() + assert len(r) == 1 + assert r[0].encode() == packet.Packet(packet.NOOP).encode() + + def test_upgrade_no_probe(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.connected = True + ws = mock.MagicMock() + ws.wait.return_value = packet.Packet(packet.NOOP).encode() + s._websocket_handler(ws) + assert not s.upgraded + + def test_upgrade_no_upgrade_packet(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.connected = True + s.queue.join = mock.MagicMock(return_value=None) + ws = mock.MagicMock() + probe = 'probe' + ws.wait.side_effect = [ + packet.Packet(packet.PING, data=probe).encode(), + packet.Packet(packet.NOOP).encode(), + ] + s._websocket_handler(ws) + ws.send.assert_called_once_with( + packet.Packet(packet.PONG, data=probe).encode() + ) + assert s.queue.get().packet_type == packet.NOOP + assert not s.upgraded + + def test_close_packet(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.connected = True + s.close = mock.MagicMock() + s.receive(packet.Packet(packet.CLOSE)) + s.close.assert_called_once_with(wait=False, abort=True) + + def test_invalid_packet_type(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + pkt = packet.Packet(packet_type=99) + with pytest.raises(exceptions.UnknownPacketError): + s.receive(pkt) + + def test_upgrade_not_supported(self): + mock_server = self._get_mock_server() + mock_server._async['websocket'] = None + s = socket.Socket(mock_server, 'sid') + s.connected = True + environ = "foo" + start_response = "bar" + s._upgrade_websocket(environ, start_response) + mock_server._bad_request.assert_called_once_with() + + def test_websocket_read_write(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.connected = False + s.queue.join = mock.MagicMock(return_value=None) + foo = 'foo' + bar = 'bar' + s.poll = mock.MagicMock( + side_effect=[ + [packet.Packet(packet.MESSAGE, data=bar)], + exceptions.QueueEmpty, + ] + ) + ws = mock.MagicMock() + ws.wait.side_effect = [ + packet.Packet(packet.MESSAGE, data=foo).encode(), + None, + ] + s._websocket_handler(ws) + self._join_bg_tasks() + assert s.connected + assert s.upgraded + assert mock_server._trigger_event.call_count == 2 + mock_server._trigger_event.assert_has_calls( + [ + mock.call('message', 'sid', 'foo', run_async=True), + mock.call('disconnect', 'sid', run_async=False), + ] + ) + ws.send.assert_called_with('4bar') + + def test_websocket_upgrade_read_write(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.connected = True + s.queue.join = mock.MagicMock(return_value=None) + foo = 'foo' + bar = 'bar' + probe = 'probe' + s.poll = mock.MagicMock( + side_effect=[ + [packet.Packet(packet.MESSAGE, data=bar)], + exceptions.QueueEmpty, + ] + ) + ws = mock.MagicMock() + ws.wait.side_effect = [ + packet.Packet(packet.PING, data=probe).encode(), + packet.Packet(packet.UPGRADE).encode(), + packet.Packet(packet.MESSAGE, data=foo).encode(), + None, + ] + s._websocket_handler(ws) + self._join_bg_tasks() + assert s.upgraded + assert mock_server._trigger_event.call_count == 2 + mock_server._trigger_event.assert_has_calls( + [ + mock.call('message', 'sid', 'foo', run_async=True), + mock.call('disconnect', 'sid', run_async=False), + ] + ) + ws.send.assert_called_with('4bar') + + def test_websocket_upgrade_with_payload(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.connected = True + s.queue.join = mock.MagicMock(return_value=None) + probe = 'probe' + ws = mock.MagicMock() + ws.wait.side_effect = [ + packet.Packet(packet.PING, data=probe).encode(), + packet.Packet(packet.UPGRADE, data='2').encode(), + ] + s._websocket_handler(ws) + self._join_bg_tasks() + assert s.upgraded + + def test_websocket_upgrade_with_backlog(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.connected = True + s.queue.join = mock.MagicMock(return_value=None) + probe = 'probe' + foo = 'foo' + ws = mock.MagicMock() + ws.wait.side_effect = [ + packet.Packet(packet.PING, data=probe).encode(), + packet.Packet(packet.UPGRADE, data='2').encode(), + ] + s.upgrading = True + s.send(packet.Packet(packet.MESSAGE, data=foo)) + environ = {'REQUEST_METHOD': 'GET', 'QUERY_STRING': 'sid=sid'} + start_response = mock.MagicMock() + packets = s.handle_get_request(environ, start_response) + assert len(packets) == 1 + assert packets[0].encode() == '6' + packets = s.poll() + assert len(packets) == 1 + assert packets[0].encode() == '4foo' + + s._websocket_handler(ws) + self._join_bg_tasks() + assert s.upgraded + assert not s.upgrading + packets = s.handle_get_request(environ, start_response) + assert len(packets) == 1 + assert packets[0].encode() == '6' + + def test_websocket_read_write_wait_fail(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.connected = False + s.queue.join = mock.MagicMock(return_value=None) + foo = 'foo' + bar = 'bar' + s.poll = mock.MagicMock( + side_effect=[ + [packet.Packet(packet.MESSAGE, data=bar)], + [packet.Packet(packet.MESSAGE, data=bar)], + exceptions.QueueEmpty, + ] + ) + ws = mock.MagicMock() + ws.wait.side_effect = [ + packet.Packet(packet.MESSAGE, data=foo).encode(), + RuntimeError, + ] + ws.send.side_effect = [None, RuntimeError] + s._websocket_handler(ws) + self._join_bg_tasks() + assert s.closed + + def test_websocket_upgrade_with_large_packet(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.connected = True + s.queue.join = mock.MagicMock(return_value=None) + probe = 'probe' + ws = mock.MagicMock() + ws.wait.side_effect = [ + packet.Packet(packet.PING, data=probe).encode(), + packet.Packet(packet.UPGRADE, data='2' * 128).encode(), + ] + with pytest.raises(ValueError): + s._websocket_handler(ws) + self._join_bg_tasks() + assert not s.upgraded + + def test_websocket_ignore_invalid_packet(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.connected = False + s.queue.join = mock.MagicMock(return_value=None) + foo = 'foo' + bar = 'bar' + s.poll = mock.MagicMock( + side_effect=[ + [packet.Packet(packet.MESSAGE, data=bar)], + exceptions.QueueEmpty, + ] + ) + ws = mock.MagicMock() + ws.wait.side_effect = [ + packet.Packet(packet.OPEN).encode(), + packet.Packet(packet.MESSAGE, data=foo).encode(), + None, + ] + s._websocket_handler(ws) + self._join_bg_tasks() + assert s.connected + assert mock_server._trigger_event.call_count == 2 + mock_server._trigger_event.assert_has_calls( + [ + mock.call('message', 'sid', foo, run_async=True), + mock.call('disconnect', 'sid', run_async=False), + ] + ) + ws.send.assert_called_with('4bar') + + def test_send_after_close(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.close(wait=False) + with pytest.raises(exceptions.SocketIsClosedError): + s.send(packet.Packet(packet.NOOP)) + + def test_close_after_close(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.close(wait=False) + assert s.closed + assert mock_server._trigger_event.call_count == 1 + mock_server._trigger_event.assert_called_once_with( + 'disconnect', 'sid', run_async=False + ) + s.close() + assert mock_server._trigger_event.call_count == 1 + + def test_close_and_wait(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.queue = mock.MagicMock() + s.close(wait=True) + s.queue.join.assert_called_once_with() + + def test_close_without_wait(self): + mock_server = self._get_mock_server() + s = socket.Socket(mock_server, 'sid') + s.queue = mock.MagicMock() + s.close(wait=False) + assert s.queue.join.call_count == 0 diff --git a/tests/performance/README.md b/tests/performance/README.md new file mode 100644 index 0000000..afce173 --- /dev/null +++ b/tests/performance/README.md @@ -0,0 +1,5 @@ +Performance +=========== + +This directory contains several scripts and tools to test the performance of +the project. diff --git a/tests/performance/binary_b64_packet.py b/tests/performance/binary_b64_packet.py new file mode 100644 index 0000000..64e2158 --- /dev/null +++ b/tests/performance/binary_b64_packet.py @@ -0,0 +1,19 @@ +import time +from engineio import packet + + +def test(): + p = packet.Packet(packet.MESSAGE, b'hello world') + start = time.time() + count = 0 + while True: + p = packet.Packet(encoded_packet=p.encode(b64=True)) + count += 1 + if time.time() - start >= 5: + break + return count + + +if __name__ == '__main__': + count = test() + print('binary_b64_packet:', count, 'packets processed.') diff --git a/tests/performance/binary_packet.py b/tests/performance/binary_packet.py new file mode 100644 index 0000000..0bda74c --- /dev/null +++ b/tests/performance/binary_packet.py @@ -0,0 +1,19 @@ +import time +from engineio import packet + + +def test(): + p = packet.Packet(packet.MESSAGE, b'hello world') + start = time.time() + count = 0 + while True: + p = packet.Packet(encoded_packet=p.encode()) + count += 1 + if time.time() - start >= 5: + break + return count + + +if __name__ == '__main__': + count = test() + print('binary_packet:', count, 'packets processed.') diff --git a/tests/performance/json_packet.py b/tests/performance/json_packet.py new file mode 100644 index 0000000..075c52d --- /dev/null +++ b/tests/performance/json_packet.py @@ -0,0 +1,19 @@ +import time +from engineio import packet + + +def test(): + p = packet.Packet(packet.MESSAGE, {'hello': 'world'}) + start = time.time() + count = 0 + while True: + p = packet.Packet(encoded_packet=p.encode()) + count += 1 + if time.time() - start >= 5: + break + return count + + +if __name__ == '__main__': + count = test() + print('json_packet:', count, 'packets processed.') diff --git a/tests/performance/payload.py b/tests/performance/payload.py new file mode 100644 index 0000000..cee1415 --- /dev/null +++ b/tests/performance/payload.py @@ -0,0 +1,20 @@ +import time +from engineio import packet, payload + + +def test(): + p = payload.Payload( + packets=[packet.Packet(packet.MESSAGE, b'hello world')] * 10) + start = time.time() + count = 0 + while True: + p = payload.Payload(encoded_payload=p.encode()) + count += 1 + if time.time() - start >= 5: + break + return count + + +if __name__ == '__main__': + count = test() + print('payload:', count, 'payloads processed.') diff --git a/tests/performance/run.sh b/tests/performance/run.sh new file mode 100755 index 0000000..e2f1dac --- /dev/null +++ b/tests/performance/run.sh @@ -0,0 +1,7 @@ +#!/bin/bash +python text_packet.py +python binary_packet.py +python binary_b64_packet.py +python json_packet.py +python payload.py +python server_receive.py diff --git a/tests/performance/server_receive.py b/tests/performance/server_receive.py new file mode 100644 index 0000000..91028c4 --- /dev/null +++ b/tests/performance/server_receive.py @@ -0,0 +1,37 @@ +import io +import sys +import time +import engineio + + +def test(eio_version, payload): + s = engineio.Server() + start = time.time() + count = 0 + s.handle_request({ + 'REQUEST_METHOD': 'GET', + 'QUERY_STRING': eio_version, + }, lambda s, h: None) + sid = list(s.sockets.keys())[0] + while True: + environ = { + 'REQUEST_METHOD': 'POST', + 'QUERY_STRING': eio_version + '&sid=' + sid, + 'CONTENT_LENGTH': '6', + 'wsgi.input': io.BytesIO(payload) + } + s.handle_request(environ, lambda s, h: None) + count += 1 + if time.time() - start >= 5: + break + return count + + +if __name__ == '__main__': + eio_version = 'EIO=4' + payload = b'4hello' + if len(sys.argv) > 1 and sys.argv[1] == '3': + eio_version = 'EIO=3' + payload = b'\x00\x06\xff4hello' + count = test(eio_version, payload) + print('server_receive:', count, 'packets received.') diff --git a/tests/performance/text_packet.py b/tests/performance/text_packet.py new file mode 100644 index 0000000..98e7021 --- /dev/null +++ b/tests/performance/text_packet.py @@ -0,0 +1,19 @@ +import time +from engineio import packet + + +def test(): + p = packet.Packet(packet.MESSAGE, 'hello world') + start = time.time() + count = 0 + while True: + p = packet.Packet(encoded_packet=p.encode()) + count += 1 + if time.time() - start >= 5: + break + return count + + +if __name__ == '__main__': + count = test() + print('text_packet:', count, 'packets processed.') |
