testdata/test/test_run.py

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}/zeros/?api_key={API_KEY}&size=32', timeout=TIMEOUT)
    assert response.content == b'\0' * 32


def test_get_file_B():

    response = requests.get(
        f'{PROTOCOL}://{HOST}:{PORT}/zeros/?api_key={API_KEY}&size=32B', timeout=TIMEOUT)
    assert response.content == b'\0' * 32


def test_get_file_KB():

    response = requests.get(
        f'{PROTOCOL}://{HOST}:{PORT}/zeros/?api_key={API_KEY}&size=32KB', timeout=TIMEOUT)
    assert response.status_code == 200
    assert response.content == b'\0' * 32 * 1000


def test_get_file_KiB():

    response = requests.get(
        f'{PROTOCOL}://{HOST}:{PORT}/zeros/?api_key={API_KEY}&size=32KiB', timeout=TIMEOUT)
    assert response.status_code == 200
    assert response.content == b'\0' * 32 * 1024


def test_get_file_invalid_format():

    response = requests.get(
        f'{PROTOCOL}://{HOST}:{PORT}/zeros/?api_key={API_KEY}&size=32Invalid', timeout=TIMEOUT)
    assert response.status_code == 400
    assert response.text == 'Invalid Format.'


def test_database_data_used(server):

    requests.get(
        f'{PROTOCOL}://{HOST}:{PORT}/zeros/?api_key={API_KEY}&size=32', timeout=TIMEOUT)
    with open(server) as file:
        assert json.loads(file.read())['data-used'] == 32