flake.nix

@@ -19,11 +19,11 @@
     pkgs = forAllSystems (system: inputs.nixpkgs.legacyPackages.${system});
     poetry2nix = forAllSystems (system: inputs.poetry2nix-lib.lib.mkPoetry2Nix {pkgs = pkgs.${system};});
 
-    addSetuptools = final: prev: list:
+    addSetuptools = self: super: list:
       builtins.listToAttrs (builtins.map (package: {
           name = "${package}";
-          value = prev."${package}".overridePythonAttrs (old: {
+          value = super."${package}".overridePythonAttrs (old: {
-            nativeBuildInputs = (old.nativeBuildInputs or []) ++ [final.setuptools];
+            nativeBuildInputs = (old.nativeBuildInputs or []) ++ [self.setuptools];
           });
         })
         list);
@@ -32,7 +32,7 @@
     packages = forAllSystems (system: let
       mkPackage = {debug ? false}:
         poetry2nix.${system}.mkPoetryApplication {
-          projectDir = ./.;
+          projectDir = self;
           checkPhase =
             if debug
             then "pyright --warnings testdata && pytest"
@@ -41,8 +41,8 @@
           preferWheels = false;
           nativeBuildInputs = with pkgs.${system}; [pyright];
           overrides =
-            poetry2nix.${system}.overrides.withDefaults (final: prev:
+            poetry2nix.${system}.overrides.withDefaults (self: super:
-              addSetuptools final prev ["sqlite-minutils" "fastlite" "python-fasthtml"]);
+              addSetuptools self super ["sqlite-minutils" "fastlite" "python-fasthtml"]);
         };
     in {
       default = mkPackage {debug = false;};

test/test_run.py

@@ -51,22 +51,21 @@ def server():
 
 def test_get_file():
 
-    response = requests.get(
+    response = requests.get(f'{PROTOCOL}://{HOST}:{PORT}/?api_key={API_KEY}&size=32', timeout = TIMEOUT)
-        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)
+        f'{PROTOCOL}://{HOST}:{PORT}/?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)
+        f'{PROTOCOL}://{HOST}:{PORT}/?api_key={API_KEY}&size=32KB', timeout=TIMEOUT)
     assert response.status_code == 200
     assert response.content == b'\0' * 32 * 1000
 
@@ -74,7 +73,7 @@ def test_get_file_KB():
 def test_get_file_KiB():
 
     response = requests.get(
-        f'{PROTOCOL}://{HOST}:{PORT}/zeros/?api_key={API_KEY}&size=32KiB', timeout=TIMEOUT)
+        f'{PROTOCOL}://{HOST}:{PORT}/?api_key={API_KEY}&size=32KiB', timeout=TIMEOUT)
     assert response.status_code == 200
     assert response.content == b'\0' * 32 * 1024
 
@@ -82,15 +81,14 @@ def test_get_file_KiB():
 def test_get_file_invalid_format():
 
     response = requests.get(
-        f'{PROTOCOL}://{HOST}:{PORT}/zeros/?api_key={API_KEY}&size=32Invalid', timeout=TIMEOUT)
+        f'{PROTOCOL}://{HOST}:{PORT}/?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(
+    requests.get(f'{PROTOCOL}://{HOST}:{PORT}/?api_key={API_KEY}&size=32', timeout = TIMEOUT)
-        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
 

testdata/api.py

@@ -0,0 +1,74 @@
+from fasthtml.fastapp import FastHTML
+from starlette import status
+from starlette.responses import StreamingResponse
+from starlette.exceptions import HTTPException
+from pydantic import ValidationError
+
+from .utils import load_database, save_database, generate_data
+from .custom_types import TestDataBody
+
+def create_api(api_keys: set[str], max_size: int, max_data: int, database: str, buffer_size: int) -> FastHTML:
+
+    api = FastHTML()
+
+    class MaxSizePerRequestError(Exception):
+        pass
+
+
+    class MinSizePerRequestError(Exception):
+        pass
+
+    @api.get('/')  # type: ignore
+    async def test_data(api_key: str, size: str) -> StreamingResponse:
+        try:
+            body = TestDataBody(api_key=api_key, size=size)  # type: ignore
+        except ValidationError as err:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail='Invalid Format.'
+            ) from err
+
+        try:
+            if body.api_key not in api_keys:
+                raise HTTPException(
+                    status_code=status.HTTP_401_UNAUTHORIZED,
+                    detail='Invalid API Key.'
+                )
+
+            if body.size < 0:
+                raise MinSizePerRequestError
+
+            if max_size < body.size:
+                raise MaxSizePerRequestError
+
+            db = load_database(database)
+            if max_data <= db['data-used'] + body.size:
+                raise HTTPException(
+                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                    detail='Service not available.'
+                )
+            db['data-used'] += body.size
+
+            save_database(database, db)
+
+            return StreamingResponse(
+                status_code=status.HTTP_200_OK,
+                content=generate_data(body.size, buffer_size),
+                media_type='application/octet-stream',
+                headers={
+                    'Content-Length': str(body.size)
+                }
+            )
+
+        except MinSizePerRequestError as err:
+            raise HTTPException(
+                status_code=status.HTTP_416_REQUESTED_RANGE_NOT_SATISFIABLE,
+                detail='Size has to be not-negative.'
+            ) from err
+        except MaxSizePerRequestError as err:
+            raise HTTPException(
+                status_code=status.HTTP_416_REQUESTED_RANGE_NOT_SATISFIABLE,
+                detail=f'Exceeded max size per request of {max_size} Bytes.'
+            ) from err
+
+    return api

testdata/app.py

@@ -1,124 +0,0 @@
-# pylint: disable=missing-module-docstring,missing-class-docstring,missing-function-docstring
-# pylint: disable=line-too-long
-
-from pathlib import Path
-
-from fasthtml.fastapp import FastHTML, FastHTMLWithLiveReload, Request, StaticFiles, picolink
-from fasthtml.common import Form, Label, Input, Br, Button, A, Html, serve, Body, Link, P, Body, Div, Span, Main, Title # pylint: disable=no-name-in-module
-from starlette import status
-from starlette.responses import Response, StreamingResponse
-from starlette.exceptions import HTTPException
-from pydantic import ValidationError
-
-from .utils import load_database, save_database, generate_data
-from .custom_types import TestDataBody, Config
-
-
-class MaxSizePerRequestError(Exception):
-    pass
-
-
-class MinSizePerRequestError(Exception):
-    pass
-
-
-with open('./config.json') as file:  # parse_cli_arguments(sys.argv[1:])
-    config = Config.model_validate_json(file.read())
-
-
-app = FastHTMLWithLiveReload(hdrs=(
-    (Link(rel='stylesheet', type='text/css', href='static/style.css'),)
-))
-
-app.mount(
-    '/static',
-    StaticFiles(directory=Path(__file__).parent.absolute() / 'static'),
-    name='static'
-)
-
-
-@app.get('/')
-async def homepage():
-    return Main(
-        P('Testdata Download'),
-        P('Download a file of exact size. The file will only contain zeros.'),
-        Div(
-            Div(
-                Form(
-                    Div(
-                        Label('API Key'), Br(),
-                        Input(name='api_key'), Br()
-                    ),
-                    Div(
-                        Label('Size'), Br(),
-                        Input(name='size', placeholder='2MiB, 1GB, etc.'), Br()
-                    ),
-                    Button('Download'),
-                    action='/zeros',  hx_boost="true"
-                ),
-                P(f'Max Download Size: {config.max_size / (1024 * 1024 * 1024):.02f} GiB'), Br(),  # nopep8
-                cls='download'
-            ),
-            P('Contact: ', A('contact.testdata@krsnik.at', href='mailto:contact.testdata@krsnik.at'), id='contact')  # nopep8
-        )
-    )
-
-
-@ app.get('/zeros')  # type: ignore
-async def get_zeros(api_key: str, size: str, request: Request) -> StreamingResponse:
-    # AJAX Requests cannot return a file, thus send a redirect.
-    if 'HX-Request' in request.headers:
-        return Response(
-            content='<p>Hello</p>',
-            headers={'HX-Redirect': f'/zeros/?api_key={api_key}&size={size}'})
-
-    try:
-        body = TestDataBody(api_key=api_key, size=size)  # type: ignore
-    except ValidationError as err:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail='Invalid Format.'
-        ) from err
-
-    try:
-        if body.api_key not in config.api_keys:
-            raise HTTPException(
-                status_code=status.HTTP_401_UNAUTHORIZED,
-                detail='Invalid API Key.'
-            )
-
-        if body.size < 0:
-            raise MinSizePerRequestError
-
-        if config.max_size < body.size:
-            raise MaxSizePerRequestError
-
-        db = load_database(config.database)
-        if config.max_data <= db['data-used'] + body.size:
-            raise HTTPException(
-                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-                detail='Service not available.'
-            )
-
-        db['data-used'] += body.size
-        save_database(config.database, db)
-
-        return StreamingResponse(
-            status_code=status.HTTP_200_OK,
-            content=generate_data(body.size, config.buffer_size),
-            media_type='application/octet-stream',
-            headers={
-                'Content-Length': str(body.size),
-            }
-        )
-
-    except MinSizePerRequestError as err:
-        raise HTTPException(
-            status_code=status.HTTP_416_REQUESTED_RANGE_NOT_SATISFIABLE,
-            detail='Size has to be not-negative.'
-        ) from err
-    except MaxSizePerRequestError as err:
-        raise HTTPException(
-            status_code=status.HTTP_416_REQUESTED_RANGE_NOT_SATISFIABLE,
-            detail=f'Exceeded max size per request of {config.max_size} Bytes.'
-        ) from err

testdata/custom_types.py

@@ -20,14 +20,6 @@ class Config(BaseModel):
     def convert_size(cls, value: int | str) -> int:
         return convert_to_bytes(value)
 
-    @field_validator('api_keys', mode='before')
-    @classmethod
-    def check_if_key_in_file(cls, value: str | list[str]) -> set[str]:
-        if isinstance(value, str):
-            with open(value) as file:
-                return set(filter(lambda x: x.strip() != '', file.read().splitlines()))
-        return set(value)
-
 
 class TestDataBody(BaseModel):
     api_key: str

testdata/main.py

@@ -1,11 +1,13 @@
 import sys
 import argparse
+import json
 import os
 from os.path import exists
 
 import uvicorn
 
 from .utils import save_database
+from .api import create_api
 from .custom_types import Config
 
 
@@ -23,16 +25,16 @@ def parse_cli_arguments(argv: list[str]) -> argparse.Namespace:
     return parser.parse_args(argv)
 
 
-def run(host: str, port: int, api_keys: set[str], max_size: int, max_data: int, database: str, buffer_size: int, reload: bool = False):
+def run(host: str, port: int, api_keys: set[str], max_size: int, max_data: int, database: str, buffer_size: int):
     if not exists(database) or os.stat(database).st_size == 0:
         save_database(database, {'data-used': 0})
 
-    # app = create_api(api_keys, max_size, max_data, database, buffer_size)
+    api = create_api(api_keys, max_size, max_data, database, buffer_size)
 
-    uvicorn.run("testdata.app:app", host=host, port=port, reload=reload)
+    uvicorn.run(api, host = host, port = port)
 
 
 def main():
     args = parse_cli_arguments(sys.argv[1:])
     config = Config.model_validate_json(args.config.read())
-    run(**config.model_dump(exclude={'log'}), reload=True)
+    run(**config.model_dump(exclude={'log'}))

testdata/static/style.css

@@ -1,48 +0,0 @@
-@font-face {
-    font-family: roboto;
-    src: url(Roboto-Regular.ttf);
-}
-
-:root {
-    --background-color: #ffffff;
-    --form-background-color: #e2ddff;
-}
-
-body {
-    background-color: var(--background-color);
-    font-family: roboto;
-}
-
-.download {
-    background-color: var(--form-background-color);
-    border-radius: 1em;
-    text-align: center;
-    width: 70%;
-    margin: 0 auto;
-    margin-top: 5em;
-}
-
-.download > form > div {
-    width: 70%;
-    margin: 0 auto;
-}
-
-.download > form > div > label {
-    font-size: 3em;
-}
-
-.download > form > div > input {
-    width: 100%;
-    font-size: 3em;
-    text-indent: 1em;
-    background-color: var(--background-color);
-}
-
-.download > form > button {
-    margin-top: 1em;
-    font-size: 3em;
-}
-
-#contact {
-    text-align: center;
-}