117 lines
3.4 KiB
Python
117 lines
3.4 KiB
Python
import json
|
|
import time
|
|
import tempfile
|
|
import asyncio
|
|
from multiprocessing import Process
|
|
from typing import Generator
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
import testdata
|
|
|
|
|
|
PROTOCOL = 'http'
|
|
HOST = 'localhost'
|
|
PORT = 1234
|
|
TIMEOUT = 1 # seconds
|
|
|
|
|
|
@pytest.fixture(scope='function')
|
|
def _server(request) -> Generator[str, None, None]:
|
|
|
|
with tempfile.NamedTemporaryFile() as tmpfile:
|
|
request.param['database'] = tmpfile.name
|
|
config = testdata.Testdata.Config.model_validate_json(json.dumps(request.param))
|
|
server = testdata.Testdata(config)
|
|
|
|
def run_server():
|
|
asyncio.run(server.run(HOST, PORT))
|
|
|
|
process = Process(target=run_server)
|
|
process.start()
|
|
|
|
# Wait until webserver becomes available
|
|
start = time.time()
|
|
while (time.time() - start) < TIMEOUT:
|
|
try:
|
|
requests.get(f'{PROTOCOL}://{HOST}:{PORT}', timeout=TIMEOUT)
|
|
break
|
|
except requests.exceptions.ConnectionError:
|
|
pass
|
|
|
|
yield tmpfile.name
|
|
|
|
process.terminate()
|
|
|
|
# Wait until webserver is completely shut down
|
|
start = time.time()
|
|
while (time.time() - start) < TIMEOUT:
|
|
try:
|
|
requests.get(f'{PROTOCOL}://{HOST}:{PORT}', timeout=TIMEOUT)
|
|
except requests.exceptions.ConnectionError:
|
|
break
|
|
|
|
|
|
@pytest.mark.parametrize('_server', [({
|
|
'keys': ['one', 'two', 'three'],
|
|
'max-size': '100',
|
|
'max-data': 1234,
|
|
'buffer-size': '12MiB',
|
|
})], indirect=['_server'])
|
|
def test_request_size_lower_bound(_server):
|
|
response = requests.get(f'{PROTOCOL}://{HOST}:{PORT}/zeros?api_key=one&size=-1', timeout=TIMEOUT)
|
|
assert response.status_code == 416
|
|
|
|
response = requests.get(f'{PROTOCOL}://{HOST}:{PORT}/zeros?api_key=one&size=0', timeout=TIMEOUT)
|
|
assert response.status_code == 200
|
|
assert response.content == b''
|
|
|
|
|
|
@pytest.mark.parametrize('_server', [({
|
|
'keys': ['one', 'two', 'three'],
|
|
'max-size': '100',
|
|
'max-data': 1234,
|
|
'buffer-size': '12MiB',
|
|
})], indirect=['_server'])
|
|
def test_request_size_upper_bound(_server):
|
|
response = requests.get(f'{PROTOCOL}://{HOST}:{PORT}/zeros?api_key=one&size=100', timeout=TIMEOUT)
|
|
assert response.status_code == 200
|
|
assert response.content == b'\0' * 100
|
|
|
|
response = requests.get(f'{PROTOCOL}://{HOST}:{PORT}/zeros?api_key=one&size=101', timeout=TIMEOUT)
|
|
assert response.status_code == 416
|
|
|
|
|
|
@pytest.mark.parametrize('_server', [({
|
|
'keys': ['one', 'two', 'three'],
|
|
'max-size': '100',
|
|
'max-data': 1234,
|
|
'buffer-size': '12MiB',
|
|
})], indirect=['_server'])
|
|
def test_invalid_api_key(_server):
|
|
response = requests.get(f'{PROTOCOL}://{HOST}:{PORT}/zeros?api_key=four&size=100', timeout=TIMEOUT)
|
|
assert response.status_code == 401
|
|
|
|
|
|
@pytest.mark.parametrize('_server', [({
|
|
'keys': ['one', 'two', 'three'],
|
|
'max-size': '1KB',
|
|
'max-data': '1KB',
|
|
'buffer-size': '12MiB',
|
|
'database-update-interval': 0.1
|
|
})], indirect=['_server'])
|
|
def test_check_database_update(_server):
|
|
database = _server
|
|
|
|
with open(database, 'r', encoding='utf-8') as file:
|
|
file.seek(0)
|
|
assert json.load(file) == {'data-used': 0}
|
|
|
|
response = requests.get(f'{PROTOCOL}://{HOST}:{PORT}/zeros?api_key=one&size=100', timeout=TIMEOUT)
|
|
assert response.status_code == 200
|
|
|
|
time.sleep(0.1)
|
|
file.seek(0)
|
|
assert json.load(file) == {'data-used': 100}
|