Rewrite

tests/__main__.py

@@ -0,0 +1,5 @@
+import sys
+
+import pytest
+
+retcode = pytest.main(sys.argv[1:])

tests/test_imports.py

@@ -0,0 +1,2 @@
+def test_import_testdata():
+    import testdata  # pylint: disable=unused-import,import-outside-toplevel

tests/test_testdata.py

@@ -0,0 +1,116 @@
+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}