64 lines
1.4 KiB
Python
64 lines
1.4 KiB
Python
|
from multiprocessing import Process
|
||
|
|
import os
|
||
|
|
import json
|
||
|
|
import tempfile
|
||
|
|
import random
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
import requests
|
||
|
|
|
||
|
|
import testdata
|
||
|
|
|
||
|
|
|
||
|
|
PROTOCOL = 'http'
|
||
|
|
HOST = '127.0.0.1'
|
||
|
|
PORT = 8080
|
||
|
|
BUFFER_SIZE = 4 * 1024
|
||
|
|
MAX_SIZE = 2 * 1024 * 1024 * 1024
|
||
|
|
MAX_DATA = 10 * 1024 * 1024 * 1024
|
||
|
|
API_KEY = f'{random.randrange(16 ** 32):032x}'
|
||
|
|
API_KEYS = { API_KEY }
|
||
|
|
|
||
|
|
TIMEOUT = 5
|
||
|
|
|
||
|
|
@pytest.fixture(autouse = True)
|
||
|
|
def server():
|
||
|
|
# Create Temporary Databases File
|
||
|
|
database = tempfile.NamedTemporaryFile(delete = False).name
|
||
|
|
|
||
|
|
proc = Process(target = testdata.run, args = (HOST, PORT, API_KEYS, MAX_SIZE, MAX_DATA, database, BUFFER_SIZE))
|
||
|
|
|
||
|
|
proc.start()
|
||
|
|
|
||
|
|
# Wait until webserver becomes available
|
||
|
|
while True:
|
||
|
|
try:
|
||
|
|
requests.get(f'{PROTOCOL}://{HOST}:{PORT}', timeout = TIMEOUT)
|
||
|
|
except requests.ConnectionError:
|
||
|
|
continue
|
||
|
|
|
||
|
|
break
|
||
|
|
|
||
|
|
yield database
|
||
|
|
|
||
|
|
# Terminate webserver
|
||
|
|
proc.terminate()
|
||
|
|
proc.join()
|
||
|
|
|
||
|
|
# Delete Temporary File
|
||
|
|
os.unlink(database)
|
||
|
|
|
||
|
|
|
||
|
|
def test_get_file():
|
||
|
|
|
||
|
|
response = requests.get(f'{PROTOCOL}://{HOST}:{PORT}/?api_key={API_KEY}&size=32', timeout = TIMEOUT)
|
||
|
|
assert response.content == b'\0' * 32
|
||
|
|
|
||
|
|
|
||
|
|
def test_database_data_used(server):
|
||
|
|
|
||
|
|
requests.get(f'{PROTOCOL}://{HOST}:{PORT}/?api_key={API_KEY}&size=32', timeout = TIMEOUT)
|
||
|
|
with open(server) as file:
|
||
|
|
assert json.loads(file.read())['data-used'] == 32
|
||
|
|
|