testdata/tests/test_testdata.py

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_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': '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': '100KB',
    'max-data': '100KB',
    'buffer-size': '12MiB',
})], indirect=['_server'])
def test_request_max_data_used(_server):
    response = requests.get(f'{PROTOCOL}://{HOST}:{PORT}/zeros?api_key=one&size=100KB', timeout=TIMEOUT)
    assert response.status_code == 200

    response = requests.get(f'{PROTOCOL}://{HOST}:{PORT}/zeros?api_key=one&size=1', timeout=TIMEOUT)
    assert response.status_code == 500


@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):
    import importlib.metadata
    from datetime import datetime

    database = _server

    with open(database, 'r', encoding='utf-8') as file:
        file.seek(0)
        today = datetime.today()
        assert json.load(file) == {
            'version': importlib.metadata.version('testdata'),
            'data-used': {
                f'{today.year}-{today.month:02}': 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) == {
            'version': importlib.metadata.version('testdata'),
            'data-used': {
                f'{today.year}-{today.month:02}': 100
            }
        }