test_websocket.py 2.39 KB
Newer Older
Luna's avatar
Luna committed
1 2 3 4 5 6 7 8 9 10 11 12 13
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)


14 15 16 17 18
async def _json_send(conn, data):
    frame = json.dumps(data)
    await conn.send(frame)


Luna's avatar
Luna committed
19 20 21 22 23 24 25 26
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):
27
    """Start a websocket connection"""
Luna's avatar
Luna committed
28
    gw_url = await get_gw(test_cli)
29
    return await websockets.connect(gw_url)
Luna's avatar
Luna committed
30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45


@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')
46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69


@pytest.mark.asyncio
async def test_ready(test_cli):
    token = await login('normal', test_cli)
    conn = await gw_start(test_cli)

    # get the hello frame but ignore it
    await _json(conn)

    await _json_send(conn, {
        'op': OP.IDENTIFY,
        'd': {
            'token': token,
        }
    })

    # try to get a ready
    try:
        await _json(conn)
        assert True
        await conn.close(1000, 'test end')
    except (Exception, websockets.ConnectionClosed):
        assert False
70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105


@pytest.mark.asyncio
async def test_ready_fields(test_cli):
    token = await login('normal', test_cli)
    conn = await gw_start(test_cli)

    # get the hello frame but ignore it
    await _json(conn)

    await _json_send(conn, {
        'op': OP.IDENTIFY,
        'd': {
            'token': token,
        }
    })

    ready = await _json(conn)
    assert isinstance(ready, dict)
    assert ready['op'] == OP.DISPATCH
    assert ready['t'] == 'READY'

    data = ready['d']
    assert isinstance(data, dict)

    # NOTE: change if default gateway changes
    assert data['v'] == 6

    # make sure other fields exist and are with
    # proper types.
    assert isinstance(data['user'], dict)
    assert isinstance(data['private_channels'], list)
    assert isinstance(data['guilds'], list)
    assert isinstance(data['session_id'], str)

    await conn.close(1000, 'test end')