diff --git a/tests/conftest.py b/tests/conftest.py index 0b9c5da..105827b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -17,6 +17,7 @@ def _test_app(unused_tcp_port, event_loop): # reassign an unused tcp port for websockets # since the config might give a used one. main_app.config['WS_PORT'] = unused_tcp_port + main_app.config['WEBSOCKET_URL'] = f'localhost:{unused_tcp_port}' # make sure we're calling the before_serving hooks event_loop.run_until_complete(main_app.startup()) diff --git a/tests/test_websocket.py b/tests/test_websocket.py new file mode 100644 index 0000000..08d98a0 --- /dev/null +++ b/tests/test_websocket.py @@ -0,0 +1,39 @@ +import pytest +import websockets +import json + +from tests.common import login +from litecord.gateway.opcodes import OP + + +async def _json(conn): + frame = await conn.recv() + return json.loads(frame) + + +async def get_gw(test_cli) -> str: + """Get the Gateway URL.""" + gw_resp = await test_cli.get('/api/v6/gateway') + gw_json = await gw_resp.json + return gw_json['url'] + + +async def gw_start(test_cli): + gw_url = await get_gw(test_cli) + return websockets.connect(gw_url) + + +@pytest.mark.asyncio +async def test_gw(test_cli): + """Test if the gateway connects and sends a proper + HELLO payload.""" + conn = await gw_start(test_cli) + + hello = await _json(conn) + assert hello['op'] == OP.HELLO + + assert isinstance(hello['d'], dict) + assert isinstance(hello['d']['heartbeat_interval'], int) + assert isinstance(hello['d']['_trace'], list) + + await conn.close(1000, 'test end')