Commit 40f7efd1 authored by Luna's avatar Luna 😻

tests: add test_ratelimits, test_gateway

 - tests: use event_loop fixture instead of get_event_loop
 - tests: move custom fixtures to conftest
parent 77a72f6a
......@@ -64,8 +64,6 @@ def make_app():
def set_blueprints(app_):
"""Set the blueprints for a given app instance"""
log.debug('blueprint setup')
bps = {
gateway: None,
auth: '/auth',
......@@ -94,8 +92,6 @@ def set_blueprints(app_):
suffix = suffix or ''
app_.register_blueprint(bp, url_prefix=f'/api/v6{suffix}')
log.debug('blueprint setup: OK')
app = make_app()
set_blueprints(app)
......@@ -197,7 +193,6 @@ async def app_before_serving():
ws_future = websockets.serve(_wrapper, host, port)
await ws_future
log.info('started')
@app.after_serving
......
from .credentials import CREDS
async def login(acc_name: str, test_cli):
creds = CREDS[acc_name]
resp = await test_cli.post('/api/v6/auth/login', json={
'email': creds['email'],
'password': creds['password']
})
if resp.status_code != 200:
raise RuntimeError(f'non-200 on login: {resp.status_code}')
rjson = await resp.json
return rjson['token']
import asyncio
import sys
import os
import pytest
# this is very hacky.
sys.path.append(os.getcwd())
from run import app as main_app, set_blueprints
@pytest.fixture(name='app')
def _test_app(unused_tcp_port, event_loop):
set_blueprints(main_app)
# reassign an unused tcp port for websockets
# since the config might give a used one.
main_app.config['WS_PORT'] = unused_tcp_port
# make sure we're calling the before_serving hooks
event_loop.run_until_complete(main_app.startup())
return main_app
@pytest.fixture(name='test_cli')
def _test_cli(app):
"""Give a test client."""
return app.test_client()
import sys
import os
sys.path.append(os.getcwd())
import pytest
from tests.common import login
@pytest.mark.asyncio
async def test_gw(test_cli):
"""Test if the gateway route is sane."""
resp = await test_cli.get('/api/v6/gateway')
assert resp.status_code == 200
rjson = await resp.json
assert isinstance(rjson, dict)
assert 'url' in rjson
assert isinstance(rjson['url'], str)
@pytest.mark.asyncio
async def test_gw_bot(test_cli):
"""Test the Get Bot Gateway route"""
token = await login('normal', test_cli)
resp = await test_cli.get('/api/v6/gateway/bot', headers={
'Authorization': token
})
assert resp.status_code == 200
rjson = await resp.json
assert isinstance(rjson, dict)
assert isinstance(rjson['url'], str)
assert isinstance(rjson['shards'], int)
assert 'session_start_limit' in rjson
ssl = rjson['session_start_limit']
assert isinstance(ssl['total'], int)
assert isinstance(ssl['remaining'], int)
assert isinstance(ssl['reset_after'], int)
import asyncio
import sys
import os
import pytest
# this is very hacky.
sys.path.append(os.getcwd())
from run import app as main_app, set_blueprints
@pytest.fixture(name='app')
def _test_app(unused_tcp_port):
set_blueprints(main_app)
loop = asyncio.get_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
# make sure we're calling the before_serving hooks
loop.run_until_complete(main_app.startup())
return main_app
@pytest.fixture(name='test_cli')
def _test_cli(app):
"""Give a test client."""
return app.test_client()
@pytest.mark.asyncio
async def test_index(test_cli):
......@@ -38,14 +7,3 @@ async def test_index(test_cli):
resp = await test_cli.get('/')
assert resp.status_code == 200
assert (await resp.get_data()).decode() == 'hewwo'
@pytest.mark.asyncio
async def test_gw(test_cli):
"""Test if the gateway route is sane."""
resp = await test_cli.get('/api/v6/gateway')
assert resp.status_code == 200
rjson = await resp.json
assert isinstance(rjson, dict)
assert 'url' in rjson
assert isinstance(rjson['url'], str)
import sys
import os
sys.path.append(os.getcwd())
import pytest
from litecord.ratelimits.bucket import Ratelimit
def test_ratelimit():
"""Test basic ratelimiting"""
r = Ratelimit(0, 10)
bucket = r.get_bucket(0)
retry_after = bucket.update_rate_limit()
assert isinstance(retry_after, float)
assert retry_after <= 10
@pytest.mark.asyncio
async def test_ratelimit_headers(test_cli):
"""Test if the basic ratelimit headers are sent."""
resp = await test_cli.get('/api/v6/gateway')
assert resp.status_code == 200
hdrs = resp.headers
assert 'X-RateLimit-Limit' in hdrs
assert 'X-RateLimit-Remaining' in hdrs
assert 'X-RateLimit-Reset' in hdrs
assert 'X-RateLimit-Global' in hdrs
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment