updated

* Project can now be started as a package
* Added unit-tests

.gitignore

@@ -1,25 +1,14 @@
-# Nix #
 /result
 
-# Direnv #
 /.direnv/
 
-# Python #
+# config and database
-# Virtual Environment
+*.json
-/.venv/
 
-# Cache
+# python
 __pycache__/
+.pytest_cache/
+.mypy_cache/
 
-# Build
+# FastHTML
-/dist/
+/.sesskey
-*.egg-info/
-
-# Tools
-/.pytest_cache/
-/.mypy_cache/
-
-# Project specific files #
-config.json
-db.json
-log.jsonl

Dockerfile

@@ -1,28 +0,0 @@
-# Nix builder
-FROM nixos/nix:latest AS builder
-
-# Copy our source and setup our working dir.
-COPY . /tmp/build
-WORKDIR /tmp/build
-
-# Build our Nix environment
-RUN nix \
-    --extra-experimental-features "nix-command flakes" \
-    --option filter-syscalls false \
-    build
-
-# Copy the Nix store closure into a directory. The Nix store closure is the
-# entire set of Nix store values that we need for our build.
-RUN mkdir /tmp/nix-store-closure
-RUN cp -r $(nix-store -qR result/) /tmp/nix-store-closure
-
-# Final image is based on scratch. We copy a bunch of Nix dependencies
-# but they're fully self-contained so we don't need Nix anymore.
-FROM scratch
-
-WORKDIR /app
-
-# Copy /nix/store
-COPY --from=builder /tmp/nix-store-closure /nix/store
-COPY --from=builder /tmp/build/result /app
-CMD ["/app/bin/testdata"]

README.md

@@ -1,15 +1,19 @@
-# Simple Testdata Generator
+# Simple Test Data Generator
 
 ## Example Config
 
 ```json
 {
-    "keys": ["TESTKEY1", "TESTKEY2", "TESTKEY3"],
+    "binds": [
-    "max-size": "1GiB",
+        "127.0.0.1:9250"
-    "max-data": "1TiB",
+    ],
-    "buffer-size": "12MiB",
+    "log": "-",
-    "database": "./db.json",
+    "buffer-size": "4KiB",
-    "database-update-interval": 5.0,
+    "max-size": "2GB",
-    "log": "./log.jsonl"
+    "api-keys": [
+        "TESTKEY"
+    ],
+    "max-data": "10GB",
+    "database": "database.json"
 }
 ```

docker-compose.yaml

@@ -1,15 +0,0 @@
-services:
-  testdata:
-    image: result/latest
-
-    build:
-      dockerfile: ./Dockerfile
-
-    environment:
-          TESTDATA_HOST: 0.0.0.0
-          TESTDATA_PORT: 1234
-          TESTDATA_CONFIG: ./config.json
-    volumes:
-      - ./config.json:/app/config.json
-      - ./db.json:/app/db.json
-      - ./log.jsonl:/app/log.jsonl

flake.lock

@@ -5,11 +5,29 @@
         "systems": "systems"
       },
       "locked": {
-        "lastModified": 1731533236,
+        "lastModified": 1705309234,
-        "narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=",
+        "narHash": "sha256-uNRRNRKmJyCRC/8y1RqBkqWBLM034y4qN7EprSdmgyA=",
         "owner": "numtide",
         "repo": "flake-utils",
-        "rev": "11707dc2f618dd54ca8739b309ec4fc024de578b",
+        "rev": "1ef2e671c3b0c19053962c07dbda38332dcebf26",
+        "type": "github"
+      },
+      "original": {
+        "owner": "numtide",
+        "repo": "flake-utils",
+        "type": "github"
+      }
+    },
+    "flake-utils_2": {
+      "inputs": {
+        "systems": "systems_2"
+      },
+      "locked": {
+        "lastModified": 1710146030,
+        "narHash": "sha256-SZ5L6eA7HJ/nmkzGG7/ISclqe6oZdOZTNoesiInkXPQ=",
+        "owner": "numtide",
+        "repo": "flake-utils",
+        "rev": "b1d9ab70662946ef0850d488da1c9019f3a9752a",
         "type": "github"
       },
       "original": {
@@ -27,11 +45,11 @@
         "spectrum": "spectrum"
       },
       "locked": {
-        "lastModified": 1748464257,
+        "lastModified": 1720034501,
-        "narHash": "sha256-PdnQSE2vPfql9WEjunj2qQnDpuuvk7HH+4djgXJSwFs=",
+        "narHash": "sha256-fzZpuVnhw5uOtA4OuXw3a+Otpy8C+QV0Uu5XfhGEPSg=",
         "owner": "astro",
         "repo": "microvm.nix",
-        "rev": "e238645b6f0447a2eb1d538d300d5049d4006f9f",
+        "rev": "a808af7775f508a2afedd1e4940a382fe1194f21",
         "type": "github"
       },
       "original": {
@@ -40,36 +58,82 @@
         "type": "github"
       }
     },
-    "nixpkgs": {
+    "nix-github-actions": {
+      "inputs": {
+        "nixpkgs": [
+          "poetry2nix-lib",
+          "nixpkgs"
+        ]
+      },
       "locked": {
-        "lastModified": 1748370509,
+        "lastModified": 1703863825,
-        "narHash": "sha256-QlL8slIgc16W5UaI3w7xHQEP+Qmv/6vSNTpoZrrSlbk=",
+        "narHash": "sha256-rXwqjtwiGKJheXB43ybM8NwWB8rO2dSRrEqes0S7F5Y=",
-        "owner": "NixOS",
+        "owner": "nix-community",
-        "repo": "nixpkgs",
+        "repo": "nix-github-actions",
-        "rev": "4faa5f5321320e49a78ae7848582f684d64783e9",
+        "rev": "5163432afc817cf8bd1f031418d1869e4c9d5547",
         "type": "github"
       },
       "original": {
-        "owner": "NixOS",
+        "owner": "nix-community",
+        "repo": "nix-github-actions",
+        "type": "github"
+      }
+    },
+    "nixpkgs": {
+      "locked": {
+        "lastModified": 1720031269,
+        "narHash": "sha256-rwz8NJZV+387rnWpTYcXaRNvzUSnnF9aHONoJIYmiUQ=",
+        "owner": "nixos",
+        "repo": "nixpkgs",
+        "rev": "9f4128e00b0ae8ec65918efeba59db998750ead6",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nixos",
         "ref": "nixos-unstable",
         "repo": "nixpkgs",
         "type": "github"
       }
     },
+    "poetry2nix-lib": {
+      "inputs": {
+        "flake-utils": "flake-utils_2",
+        "nix-github-actions": "nix-github-actions",
+        "nixpkgs": [
+          "nixpkgs"
+        ],
+        "systems": "systems_3",
+        "treefmt-nix": "treefmt-nix"
+      },
+      "locked": {
+        "lastModified": 1719850884,
+        "narHash": "sha256-UU/lVTHFx0GpEkihoLJrMuM9DcuhZmNe3db45vshSyI=",
+        "owner": "nix-community",
+        "repo": "poetry2nix",
+        "rev": "42262f382c68afab1113ebd1911d0c93822d756e",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nix-community",
+        "repo": "poetry2nix",
+        "type": "github"
+      }
+    },
     "root": {
       "inputs": {
         "microvm": "microvm",
-        "nixpkgs": "nixpkgs"
+        "nixpkgs": "nixpkgs",
+        "poetry2nix-lib": "poetry2nix-lib"
       }
     },
     "spectrum": {
       "flake": false,
       "locked": {
-        "lastModified": 1746869549,
+        "lastModified": 1708358594,
-        "narHash": "sha256-BKZ/yZO/qeLKh9YqVkKB6wJiDQJAZNN5rk5NsMImsWs=",
+        "narHash": "sha256-e71YOotu2FYA67HoC/voJDTFsiPpZNRwmiQb4f94OxQ=",
         "ref": "refs/heads/main",
-        "rev": "d927e78530892ec8ed389e8fae5f38abee00ad87",
+        "rev": "6d0e73864d28794cdbd26ab7b37259ab0e1e044c",
-        "revCount": 862,
+        "revCount": 614,
         "type": "git",
         "url": "https://spectrum-os.org/git/spectrum"
       },
@@ -92,6 +156,56 @@
         "repo": "default",
         "type": "github"
       }
+    },
+    "systems_2": {
+      "locked": {
+        "lastModified": 1681028828,
+        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
+        "owner": "nix-systems",
+        "repo": "default",
+        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nix-systems",
+        "repo": "default",
+        "type": "github"
+      }
+    },
+    "systems_3": {
+      "locked": {
+        "lastModified": 1681028828,
+        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
+        "owner": "nix-systems",
+        "repo": "default",
+        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
+        "type": "github"
+      },
+      "original": {
+        "id": "systems",
+        "type": "indirect"
+      }
+    },
+    "treefmt-nix": {
+      "inputs": {
+        "nixpkgs": [
+          "poetry2nix-lib",
+          "nixpkgs"
+        ]
+      },
+      "locked": {
+        "lastModified": 1719749022,
+        "narHash": "sha256-ddPKHcqaKCIFSFc/cvxS14goUhCOAwsM1PbMr0ZtHMg=",
+        "owner": "numtide",
+        "repo": "treefmt-nix",
+        "rev": "8df5ff62195d4e67e2264df0b7f5e8c9995fd0bd",
+        "type": "github"
+      },
+      "original": {
+        "owner": "numtide",
+        "repo": "treefmt-nix",
+        "type": "github"
+      }
     }
   },
   "root": "root",

flake.nix

@@ -1,123 +1,89 @@
 {
-  description = "A webserver to create files for testing purposes";
+  description = "A webserver to create files for tetsing purposes";
 
   inputs = {
-    nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
+    nixpkgs.url = "github:nixos/nixpkgs/nixos-unstable";
-
+    poetry2nix-lib = {
+      url = "github:nix-community/poetry2nix";
+      inputs.nixpkgs.follows = "nixpkgs";
+    };
     microvm = {
       url = "github:astro/microvm.nix";
       inputs.nixpkgs.follows = "nixpkgs";
     };
   };
 
-  outputs =
+  outputs = {self, ...} @ inputs: let
-    {
+    supportedSystems = ["x86_64-linux" "x86_64-darwin" "aarch64-linux" "aarch64-darwin"];
-      self,
+    forAllSystems = inputs.nixpkgs.lib.genAttrs supportedSystems;
-      nixpkgs,
+    pkgs = forAllSystems (system: inputs.nixpkgs.legacyPackages.${system});
-      ...
+    poetry2nix = forAllSystems (system: inputs.poetry2nix-lib.lib.mkPoetry2Nix {pkgs = pkgs.${system};});
-    }@inputs:
-    let
-      supportedSystems = [ "x86_64-linux" ];
-      forAllSystems = nixpkgs.lib.genAttrs supportedSystems;
-      pkgs = forAllSystems (system: nixpkgs.legacyPackages.${system});
-    in
-    {
-      # `nix build`
-      packages = forAllSystems (system: rec {
-        default = testdata;
-        testdata = pkgs.${system}.callPackage ./nix/package.nix { src = ./.; };
-        vm = self.nixosConfigurations.vm.config.microvm.declaredRunner;
-      });
 
-      # `nix develop`
+    addSetuptools = final: prev: list:
-      devShells = forAllSystems (system: rec {
+      builtins.listToAttrs (builtins.map (package: {
-        default = venv;
+          name = "${package}";
-
+          value = prev."${package}".overridePythonAttrs (old: {
-        venv = pkgs.${system}.mkShell {
+            nativeBuildInputs = (old.nativeBuildInputs or []) ++ [final.setuptools];
-          shellHook = ''
+          });
-            if [ ! -d .venv/ ]; then
+        })
-              echo "Creating Virtual Environment..."
+        list);
-              ${pkgs.${system}.python3}/bin/python3 -m venv .venv
+  in {
-            fi
+    # `nix build`
-
+    packages = forAllSystems (system: let
-            alias activate='source .venv/bin/activate'
+      mkPackage = {debug ? false}:
-
+        poetry2nix.${system}.mkPoetryApplication {
-            echo "Entering Virtual Environment..."
+          projectDir = ./.;
-            source .venv/bin/activate
+          checkPhase =
-          '';
+            if debug
+            then "pyright --warnings testdata && pytest"
+            else "";
+          # doCheck = debug;
+          preferWheels = false;
+          nativeBuildInputs = with pkgs.${system}; [pyright];
+          overrides =
+            poetry2nix.${system}.overrides.withDefaults (final: prev:
+              addSetuptools final prev ["sqlite-minutils" "fastlite" "python-fasthtml"]);
         };
-      });
+    in {
+      default = mkPackage {debug = false;};
+      debug = mkPackage {debug = true;};
+      vm = self.nixosConfigurations.vm.config.microvm.declaredRunner;
+    });
 
-      # NixOS Module
+    # `nix run`
-      nixosModules.default = import ./nix/module.nix inputs;
+    apps = forAllSystems (system: {
-
+      default = {
-      # NixOS definition for a microvm to test nixosModules
+        program = "${self.packages.${system}.default}/bin/testdata";
-      nixosConfigurations."vm" = nixpkgs.lib.nixosSystem {
+        type = "app";
-        system = "x86_64-linux";
-
-        modules = [
-          inputs.microvm.nixosModules.microvm
-          (
-            { config, ... }:
-            {
-              services.getty.autologinUser = "root";
-
-              microvm = {
-                hypervisor = "qemu";
-
-                shares = [
-                  {
-                    # Host's /nix/store will be picked up so that no squashfs/erofs will be built for it.
-                    tag = "ro-store";
-                    source = "/nix/store";
-                    mountPoint = "/nix/.ro-store";
-                  }
-                ];
-
-                interfaces = [
-                  {
-                    type = "user";
-                    id = "qemu";
-                    # Locally administered have one of 2/6/A/E in the second nibble.
-                    mac = "02:00:00:01:01:01";
-                  }
-                ];
-
-                forwardPorts = [
-                  {
-                    host.port = config.services.testdata.port;
-                    guest.port = config.services.testdata.port;
-                  }
-                ];
-              };
-            }
-          )
-
-          self.nixosModules.default
-          rec {
-            networking.firewall.allowedTCPPorts = [ services.testdata.port ];
-            services.testdata = {
-              enable = true;
-
-              host = "0.0.0.0";
-              port = 1234;
-
-              settings = {
-                keys = [
-                  "one"
-                  "two"
-                  "three"
-                ];
-                max-size = "1GB";
-                max-data = "100GB";
-                buffer-size = "12MiB";
-                database = "/root/testdata-state.json";
-                database-update-interval = 5.0;
-                log = "/root/log.jsonl";
-              };
-            };
-          }
-        ];
       };
-    };
+    });
+
+    # `nix fmt`
+    formatter = forAllSystems (system: pkgs.${system}.alejandra);
+
+    # `nix develop`
+    devShells = forAllSystems (system: {
+      # Shell for app dependencies.
+      #
+      #     nix develop
+      #
+      # Use this shell for developing your app.
+      default = pkgs.${system}.mkShellNoCC {
+        inputsFrom = [self.packages.${system}.default];
+        packages = [self.packages.${system}.default];
+      };
+
+      # Shell for poetry.
+      #
+      #     nix develop .#poetry
+      #
+      # Use this shell for changes to pyproject.toml and poetry.lock.
+      poetry = pkgs.${system}.mkShellNoCC {
+        packages = [pkgs.${system}.poetry];
+      };
+    });
+
+    # NixOS Module
+    nixosModules.default = import ./nix/module.nix inputs;
+  };
 }

nix/module.nix

@@ -1,68 +1,50 @@
-inputs:
+inputs: {
-{
   config,
   lib,
   pkgs,
+  system,
   ...
-}:
+}: let
-let
+  cfg = config.testdata;
-  cfg = config.services.testdata;
   package = inputs.self.packages.${pkgs.stdenv.hostPlatform.system}.default;
-  inherit (lib)
+  inherit (lib) mkIf mkEnableOption mkOption types;
-    mkIf
-    mkEnableOption
-    mkOption
-    types
-    ;
 
-  format = pkgs.formats.json { };
+  format = pkgs.formats.json {};
   configFile = format.generate "config.json" cfg.settings;
-in
+in {
-{
+  options.testdata = {
-  options.services.testdata = {
     enable = mkEnableOption "testdata";
 
     settings = mkOption {
-      type =
+      type = with types; let
-        with types;
+        valueType = nullOr (oneOf [
-        let
+          # TODO: restrict type to actual config file structure
-          valueType = nullOr (oneOf [
+          bool
-            bool
+          int
-            int
+          float
-            float
+          str
-            str
+          path
-            path
+          (attrsOf valueType)
-            (attrsOf valueType)
+          (listOf valueType)
-            (listOf valueType)
+        ]);
-          ]);
+      in
-        in
         valueType;
-      default = throw "Please specify services.testdata.settings";
+      default = throw "Please specify testdata.settings";
-    };
-
-    host = mkOption {
-      type = types.str;
-      default = throw "Please specify a services.testdata.port";
-    };
-
-    port = mkOption {
-      type = types.int;
-      default = throw "Please specify a services.testdata.port";
     };
   };
 
   config = mkIf cfg.enable {
-    environment.systemPackages = [ package ];
+    environment.systemPackages = [package];
 
     systemd.services.testdata = {
       enable = true;
 
       serviceConfig = {
         Type = "simple";
-        ExecStart = "${package}/bin/testdata --config ${configFile} --listen ${cfg.host} --port ${builtins.toString cfg.port}";
+        ExecStart = "${package}/bin/testdata --config ${configFile}";
       };
 
-      wantedBy = [ "multi-user.target" ];
+      wantedBy = ["multi-user.target"];
     };
   };
 }

nix/package.nix

@@ -1,46 +0,0 @@
-{
-  src,
-  python3Packages,
-}:
-let
-  inherit (python3Packages)
-    setuptools
-    fastapi
-    uvicorn
-    pydantic
-    pytest
-    requests
-    mypy
-    pylint
-    ;
-
-  project = (builtins.fromTOML (builtins.readFile "${src}/pyproject.toml")).project;
-  pname = project.name;
-  version = project.version;
-in
-python3Packages.buildPythonPackage {
-  inherit pname version src;
-
-  pyproject = true;
-
-  build-system = [ setuptools ];
-
-  dependencies = [
-    fastapi
-    uvicorn
-    pydantic
-  ];
-
-  nativeCheckInputs = [
-    pytest
-    requests
-    mypy
-    pylint
-  ];
-
-  checkPhase = ''
-    pytest tests
-    mypy src
-    pylint src
-  '';
-}

pyproject.toml

@@ -1,40 +1,27 @@
-[project]
+[tool.poetry]
 name = "testdata"
-version = "1.2.2"
+version = "0.1.0"
-requires-python = "~=3.12, <4"
+description = ""
-dependencies = ["fastapi~=0.115", "uvicorn~=0.32", "pydantic~=2.9"]
+authors = ["Kristian Krsnik <git@krsnik.at>"]
+readme = "README.md"
+packages = [{ include = "testdata" }]
 
-[project.optional-dependencies]
+[tool.poetry.dependencies]
-dev = [
+python = "^3.11"
-    "pytest~=8.3",
+pydantic = "^2.6.4"
-    "mypy~=1.13",
+ipaddress = "^1.0.23"
-    "pylint~=3.3",
+uvicorn = "^0.30.3"
-    "requests~=2.32",
+python-fasthtml = "^0.2.1"
-    "types-requests~=2.32",
-]
 
-[project.scripts]
+[tool.poetry.scripts]
 testdata = "testdata.main:main"
 
+[tool.poetry.group.dev.dependencies]
+pytest = "^8.3.2"
+requests = "^2.32.3"
+types-requests = "^2.32.0.20240712"
+pylint = "^3.2.6"
+
 [build-system]
-requires = ["setuptools~=78.1"]
+requires = ["poetry-core"]
-build-backend = "setuptools.build_meta"
+build-backend = "poetry.core.masonry.api"
-
-[tool.setuptools.packages.find]
-where = ["src"]
-
-[tool.setuptools.package-data]
-testdata = ["py.typed"]
-
-[tool.autopep8]
-max_line_length = 150
-
-[tool.pylint.'MESSAGES CONTROL']
-disable = [
-    "line-too-long",
-    "missing-module-docstring",
-    "missing-class-docstring",
-    "missing-function-docstring",
-    "too-few-public-methods",
-    "broad-exception-caught",
-]

src/testdata/__init__.py

@@ -1,3 +0,0 @@
-from .testdata import Testdata
-from .utils import convert_to_bytes
-from .main import run

src/testdata/__main__.py

@@ -1,3 +0,0 @@
-from .main import main
-
-main()

src/testdata/logger/__init__.py

@@ -1,3 +0,0 @@
-from logging import getLogger, Logger
-
-from .logger import setup_logging

src/testdata/logger/logger.py

@@ -1,173 +0,0 @@
-import sys
-import json
-import logging
-import logging.handlers
-import atexit
-from datetime import datetime, timezone
-from typing_extensions import override
-
-
-
-LOG_RECORD_BUILTIN_ATTRS = {
-    "args",
-    "asctime",
-    "created",
-    "exc_info",
-    "exc_text",
-    "filename",
-    "funcName",
-    "levelname",
-    "levelno",
-    "lineno",
-    "module",
-    "msecs",
-    "message",
-    "msg",
-    "name",
-    "pathname",
-    "process",
-    "processName",
-    "relativeCreated",
-    "stack_info",
-    "thread",
-    "threadName",
-    "taskName",
-}
-
-
-class JSONFormatter(logging.Formatter):
-
-    def __init__(self, *, fmt_keys: dict[str, str] | None = None):
-        super().__init__()
-        self.fmt_keys = fmt_keys if fmt_keys is not None else {}
-
-    @override
-    def format(self, record: logging.LogRecord) -> str:
-        message = self._prepare_log_dict(record)
-        return json.dumps(message, default=str)
-
-    def _prepare_log_dict(self, record: logging.LogRecord) -> dict:
-        always_fields = {
-            'message': record.getMessage(),
-            'timestamp': datetime.fromtimestamp(
-                record.created, tz=timezone.utc
-            ).isoformat()
-        }
-
-        if record.exc_info is not None:
-            always_fields['exc_info'] = self.formatException(record.exc_info)
-
-        if record.stack_info is not None:
-            always_fields['stack_info'] = self.formatStack(record.stack_info)
-
-        message = {
-            key: msg_value
-            if (msg_value := always_fields.pop(value, None)) is not None
-            else getattr(record, value)
-            for key, value in self.fmt_keys.items()
-        }
-
-        message.update(always_fields)
-
-        for key, value in record.__dict__.items():
-            if key not in LOG_RECORD_BUILTIN_ATTRS:
-                message[key] = value
-
-        return message
-
-
-class NonErrorFilter(logging.Filter):
-    @override
-    def filter(self, record: logging.LogRecord) -> bool | logging.LogRecord:
-        return record.levelno <= logging.INFO
-
-
-def generate_log_config(log_path: str | None = None) -> dict:
-    logger_config: dict = {
-        'version': 1,
-        'disable_existing_loggers': False,
-        'filters': {
-            'no_errors': {
-                "()": NonErrorFilter
-            }
-        },
-        'formatters': {
-            'simple': {
-                'format': '[%(asctime)s][%(levelname)s] %(message)s',
-                'datefmt': '%Y-%m-%d %H:%M:%S'
-            },
-            'detailed': {
-                'format': '[%(asctime)s][%(levelname)s] %(message)s',
-                'datefmt': '%Y-%m-%dT%H:%M:%S%z'  # ISO-8601 Timestamp
-            },
-            'json': {
-                '()': JSONFormatter,
-                'fmt_keys': {
-                    'timestamp': 'timestamp',
-                    'level': 'levelname',
-                    'message': 'message',
-                    'logger': 'name',
-                    'module': 'module',
-                    'function': 'funcName',
-                    'line': 'lineno',
-                    'thread_name': 'threadName'
-                },
-            }
-        },
-        'handlers': {
-            'stdout': {
-                'class': logging.StreamHandler,
-                'level': 'INFO',
-                'filters': ['no_errors'],
-                'formatter': 'simple',
-                'stream': 'ext://sys.stdout'
-            },
-            'stderr': {
-                'class': logging.StreamHandler,
-                'level': 'WARNING',
-                'formatter': 'simple',
-                'stream': 'ext://sys.stderr'
-            }
-        } | ({'file': {
-                'class': logging.handlers.RotatingFileHandler,
-                'level': 'DEBUG',
-                'formatter': 'json',
-                'filename': log_path,
-                'maxBytes': 1024 * 1024 * 10,  # 10 MiB
-                'backupCount': 3
-        }} if log_path is not None else {}),
-        'loggers': {
-            'root': {
-                'level': 'DEBUG',
-                'handlers': [
-                    'stdout',
-                    'stderr'
-                ] + (['file'] if log_path is not None else []),
-            }
-        }
-    }
-
-    if sys.version_info >= (3, 12):  # Python 3.12+
-        logger_config['handlers']['queue_handler'] = {
-            'class': logging.handlers.QueueHandler,
-            'respect_handler_level': True,
-            'handlers': [
-                'stdout',
-                'stderr'
-            ] + (['file'] if log_path is not None else []),
-        }
-
-        logger_config['loggers']['root']['handlers'] = ['queue_handler']
-
-    return logger_config
-
-
-def setup_logging(log_path: str | None = None) -> None:
-    log_config = generate_log_config(log_path if log_path != '-' else None)
-    logging.config.dictConfig(log_config)
-
-    if sys.version_info >= (3, 12):  # Python 3.12+
-        queue_handler = logging.getHandlerByName('queue_handler')
-        if queue_handler is not None:
-            queue_handler.listener.start()  # type: ignore
-            atexit.register(queue_handler.listener.stop)  # type: ignore

src/testdata/main.py

@@ -1,46 +0,0 @@
-import os
-import sys
-import argparse
-import asyncio
-import shutil
-
-from .testdata import Testdata
-
-def parse_args(args: list[str]):
-    def formatter(prog):
-        return argparse.ArgumentDefaultsHelpFormatter(prog, max_help_position=shutil.get_terminal_size().columns)
-
-    parser = argparse.ArgumentParser(formatter_class=formatter)
-
-    parser.add_argument(
-        '-c', '--config', type=argparse.FileType('r'),
-        default=os.environ['TESTDATA_CONFIG'] if 'TESTDATA_CONFIG' in os.environ else './config.json',
-        help='Path to config file in JSON format.'
-    )
-    parser.add_argument(
-        '-l', '--listen', type=str,
-        default=os.environ['TESTDATA_HOST'] if 'TESTDATA_HOST' in os.environ else '0.0.0.0',
-        help='IP on which to listen.'
-    )
-    parser.add_argument(
-        '-p', '--port', type=int,
-        default=os.environ['TESTDATA_PORT'] if 'TESTDATA_PORT' in os.environ else 8080,
-        help='Port on which to serve the webserver.'
-    )
-
-    return parser.parse_args(args)
-
-
-def run(argv: list[str]) -> None:
-    # Parse command-line parameters
-    args = parse_args(argv)
-
-    # Load Config
-    config = Testdata.Config.model_validate_json(args.config.read())
-
-    # Run webserver
-    asyncio.run(Testdata(config).run(args.listen, args.port))
-
-
-def main() -> None:
-    run(sys.argv[1:])

src/testdata/testdata.py

@@ -1,240 +0,0 @@
-import os
-import json
-import asyncio
-import inspect
-import functools
-import random
-import importlib.metadata
-from datetime import datetime
-
-import uvicorn
-from typing_extensions import Annotated
-from fastapi import FastAPI, Request, Security, status, HTTPException
-from fastapi.security import APIKeyHeader, APIKeyQuery
-from fastapi.responses import StreamingResponse
-from pydantic import BaseModel, ConfigDict, Field, BeforeValidator, ValidationError
-
-from . import logger
-from .utils import convert_to_bytes, generate_data
-
-class MaxSizePerRequestError(Exception):
-    pass
-
-
-class MinSizePerRequestError(Exception):
-    pass
-
-
-class Testdata:
-
-    class Config(BaseModel):
-
-        model_config = ConfigDict(extra='forbid')
-
-        @staticmethod
-        def to_bytes(value: int | str) -> int:
-            try:
-                return convert_to_bytes(value)
-            except Exception as err:
-                raise ValidationError from err
-
-        @staticmethod
-        def is_authorized_keys(value: set[str] | str) -> set[str]:
-            if isinstance(value, str):
-                with open(value, encoding='utf-8') as file:
-                    return set(filter(lambda x: x.strip() != '', file.read().splitlines()))
-            return value
-
-        authorized_keys: Annotated[set[str], BeforeValidator(is_authorized_keys)] = Field(alias='keys')
-        max_size: Annotated[int, BeforeValidator(to_bytes)] = Field(alias='max-size')
-        max_data: Annotated[int, BeforeValidator(to_bytes)] = Field(alias='max-data')
-        buffer_size: Annotated[int, BeforeValidator(to_bytes)] = Field(alias='buffer-size')
-        database: str | None = None
-        log: str | None = Field(alias='log', default=None)
-        database_update_interval: float = Field(alias='database-update-interval', default=5)
-
-    _config: Config
-    _api: FastAPI
-    _state: dict
-    _logger: logger.Logger
-
-    def __init__(self, config: Config):
-        self._config = config
-        self._logger = logger.getLogger('testdata')
-        self._api = self._setup_api()
-
-        # Store internal state
-        self._state = {
-            'version': importlib.metadata.version('testdata'),  # For future compatibility
-            'data-used': {f'{(today := datetime.today()).year}-{today.month:02}': 0}  # math each months data usage
-        }
-
-    def _setup_api(self) -> FastAPI:
-        api = FastAPI(docs_url='/', redoc_url=None)
-
-        # Security
-        def get_api_key(
-            api_key_query: str = Security(APIKeyQuery(name="api_key", auto_error=False)),
-            api_key_header: str = Security(APIKeyHeader(name="x-api-key", auto_error=False))
-        ) -> str:
-            # https://joshdimella.com/blog/adding-api-key-auth-to-fast-api
-
-            if api_key_query in self._config.authorized_keys:
-                return api_key_query
-            if api_key_header in self._config.authorized_keys:
-                return api_key_header
-
-            raise HTTPException(
-                status_code=status.HTTP_401_UNAUTHORIZED,
-                detail='Invalid or missing API Key'
-            )
-
-        # A wrapper to set the function signature to accept the api key dependency
-        def secure(func):
-            # Get old signature
-            positional_only, positional_or_keyword, variadic_positional, keyword_only, variadic_keyword = [], [], [], [], []
-            for value in inspect.signature(func).parameters.values():
-                if value.kind == inspect.Parameter.POSITIONAL_ONLY:
-                    positional_only.append(value)
-                elif value.kind == inspect.Parameter.POSITIONAL_OR_KEYWORD:
-                    positional_or_keyword.append(value)
-                elif value.kind == inspect.Parameter.VAR_POSITIONAL:
-                    variadic_positional.append(value)
-                elif value.kind == inspect.Parameter.KEYWORD_ONLY:
-                    keyword_only.append(value)
-                elif value.kind == inspect.Parameter.VAR_KEYWORD:
-                    variadic_keyword.append(value)
-
-            # Avoid passing an unrecognized keyword
-            if inspect.iscoroutinefunction(func):
-                async def wrapper(*args, **kwargs):
-                    if len(variadic_keyword) == 0:
-                        if 'api_key' in kwargs:
-                            del kwargs['api_key']
-
-                    return await func(*args, **kwargs)
-            else:
-                def wrapper(*args, **kwargs):
-                    if len(variadic_keyword) == 0:
-                        if 'api_key' in kwargs:
-                            del kwargs['api_key']
-
-                    return func(*args, **kwargs)
-
-            # Override signature
-            wrapper.__signature__ = inspect.signature(func).replace(
-                parameters=(
-                    *positional_only,
-                    *positional_or_keyword,
-                    *variadic_positional,
-                    *keyword_only,
-                    inspect.Parameter('api_key', inspect.Parameter.POSITIONAL_OR_KEYWORD, default=Security(get_api_key)),
-                    *variadic_keyword
-                )
-            )
-
-            return functools.wraps(func)(wrapper)
-
-        # Routes
-        api.get('/zeros')(secure(self._zeros))
-
-        return api
-
-    async def _zeros(self, size: int | str, request: Request, filename: str = 'zeros.bin') -> StreamingResponse:
-        try:
-            extra = {'id': f'{random.randint(0, 2 ** 32 - 1):08X}'}
-            self._logger.debug(
-                'Initiated request.',
-                extra=extra | {
-                    'ip': request.client.host if request.client is not None else None,
-                    'query-params': dict(request.query_params),
-                    'headers': dict(request.headers)
-                }
-            )
-
-            try:
-                size = convert_to_bytes(size)
-            except ValueError as err:
-                self._logger.warning('Invalid format for size.', extra=extra)
-                raise HTTPException(
-                    status_code=status.HTTP_400_BAD_REQUEST,
-                    detail='Invalid format for size.'
-                ) from err
-
-            if size < 0:
-                raise MinSizePerRequestError
-            if self._config.max_size < size:
-                raise MaxSizePerRequestError
-
-            # update internal state
-            current_date = f'{(today := datetime.today()).year}-{today.month:02}'
-            if current_date not in self._state['data-used']:
-                self._state['data-used'][current_date] = 0
-            if self._config.max_data < self._state['data-used'][current_date] + size:
-                raise HTTPException(
-                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-                    detail='Service not available.'
-                )
-            self._state['data-used'][current_date] += size
-
-            self._logger.debug('Successfully processed request.', extra=extra)
-            return StreamingResponse(
-                status_code=status.HTTP_200_OK,
-                content=generate_data(size, self._config.buffer_size),
-                media_type='application/octet-stream',
-                headers={
-                    'Content-Length': str(size),
-                    'Content-Disposition': f'attachment; filename="{filename}"'
-                }
-            )
-
-        except MinSizePerRequestError as err:
-            self._logger.warning('Size if negative.', extra=extra)
-            raise HTTPException(
-                status_code=status.HTTP_416_REQUESTED_RANGE_NOT_SATISFIABLE,
-                detail='Size has to be non-negative.'
-            ) from err
-        except MaxSizePerRequestError as err:
-            self._logger.warning('Exceeded max size per request.', extra=extra)
-            raise HTTPException(
-                status_code=status.HTTP_416_REQUESTED_RANGE_NOT_SATISFIABLE,
-                detail=f'Exceeded max size per request of {self._config.max_size} Bytes.'
-            ) from err
-        except Exception as err:
-            self._logger.exception(err)
-            raise err
-
-    async def _update_state(self) -> None:
-        assert self._config.database is not None
-
-        mode = 'r+' if os.path.exists(self._config.database) else 'w+'
-
-        with open(self._config.database, mode, encoding='utf-8') as file:
-            try:
-                self._state = json.load(file)
-            except json.JSONDecodeError:
-                pass
-
-            while True:
-                file.seek(0)
-                json.dump(self._state, file, indent=2)
-                file.truncate()
-                await asyncio.sleep(self._config.database_update_interval)
-
-    async def run(self, host: str, port: int) -> None:
-        try:
-            if self._config.log is not None:
-                logger.setup_logging(self._config.log)
-
-            self._logger = logger.getLogger('testdata')
-            self._logger.info('Server started.')
-
-            coroutines = [uvicorn.Server(uvicorn.Config(self._api, host, port)).serve()]
-            if self._config.database is not None:
-                coroutines.append(self._update_state())
-
-            await asyncio.gather(*coroutines)
-        except asyncio.exceptions.CancelledError:
-            self._logger.info('Server stopped.')
-        except Exception as err:
-            self._logger.exception(err)

src/testdata/utils.py

@@ -1,40 +0,0 @@
-import asyncio
-from typing import AsyncGenerator
-
-def convert_to_bytes(size: int | str) -> int:
-    if isinstance(size, int):
-        return size
-    if isinstance(size, str):
-        try:
-            return int(size)
-        except ValueError as err:
-            units = {
-                'TB': 1000 ** 4, 'TiB': 1024 ** 4,
-                'GB': 1000 ** 3, 'GiB': 1024 ** 3,
-                'MB': 1000 ** 2, 'MiB': 1024 ** 2,
-                'KB': 1000, 'KiB': 1024,
-                'B': 1
-            }
-
-            for unit, value in units.items():
-                if size.endswith(unit):
-                    return int(float(size.removesuffix(unit)) * value)
-
-            raise ValueError from err
-
-    else:
-        raise ValueError
-
-
-async def generate_data(size: int, buffer_size: int = 4 * 1024) -> AsyncGenerator[bytes, None]:
-    # https://github.com/tiangolo/fastapi/issues/5183
-    # https://github.com/encode/starlette/discussions/1776#discussioncomment-3207518
-
-    size_left = size
-    while size_left > buffer_size:
-        size_left -= buffer_size
-        yield b'\0' * buffer_size
-        await asyncio.sleep(0)
-
-    yield b'\0' * size_left
-    await asyncio.sleep(0)

test/test_imports.py

@@ -0,0 +1,3 @@
+def test_import():
+    import testdata
+    from testdata import run

test/test_run.py

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

test/test_utils.py

@@ -0,0 +1,23 @@
+from pytest import raises
+
+from testdata.utils import convert_to_bytes
+
+
+def test_convert_to_bytes():
+    test_values = {
+        ('0', 0),
+        ('32', 32),
+        ('10_000', 10000),
+        ('999999999999999999', 999999999999999999),
+    }
+
+    for input, expected_output in test_values:
+        assert convert_to_bytes(input) == expected_output
+
+    test_exceptions = {
+        ('9E3', ValueError)
+    }
+
+    for input, exception in test_exceptions:
+        with raises(exception):
+            convert_to_bytes(input)

testdata/__init__.py

@@ -0,0 +1 @@
+from .main import run

testdata/__main__.py

@@ -0,0 +1,3 @@
+from testdata.main import main
+
+main()

testdata/app.py

@@ -0,0 +1,124 @@
+# 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

@@ -0,0 +1,41 @@
+from pydantic import BaseModel, Field, ConfigDict, field_validator
+
+from .utils import convert_to_bytes
+
+
+class Config(BaseModel):
+    host: str
+    port: int
+    buffer_size: int = Field(alias='buffer-size', default=4 * 1024)  # 4KB
+    max_size: int = Field(alias='max-size', default=2 * 1024 ** 3)  # 2GB
+    max_data: int = Field(alias='max-data', default=0)  # unlimited
+    api_keys: set[str] = Field(alias='api-keys')
+    database: str
+    log: str = '-'
+
+    model_config = ConfigDict(extra='forbid')
+
+    @field_validator('buffer_size', 'max_size', 'max_data', mode='before')
+    @classmethod
+    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
+    size: int
+
+    model_config = ConfigDict(extra='forbid')
+
+    @field_validator('size', mode='before')
+    @classmethod
+    def convert_size(cls, value: str) -> int:
+        return convert_to_bytes(value)

testdata/main.py

@@ -0,0 +1,38 @@
+import sys
+import argparse
+import os
+from os.path import exists
+
+import uvicorn
+
+from .utils import save_database
+from .custom_types import Config
+
+
+# Setup Parser
+def parse_cli_arguments(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        '-c',
+        '--config',
+        type = argparse.FileType('r'),
+        default = './config.json',
+        help = 'Path to config file in JSON format.'
+    )
+
+    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):
+    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)
+
+    uvicorn.run("testdata.app:app", host=host, port=port, reload=reload)
+
+
+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)

testdata/static/style.css

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

testdata/utils.py

@@ -0,0 +1,55 @@
+import json
+import asyncio
+from typing import AsyncGenerator
+
+
+def convert_to_bytes(size: int | str) -> int:
+    if isinstance(size, int):
+        return size
+
+    try:
+        return int(size)
+    except ValueError as err:
+        units = {
+            'TB': 1000 ** 4, 'TiB': 1024 ** 4,
+            'GB': 1000 ** 3, 'GiB': 1024 ** 3,
+            'MB': 1000 ** 2, 'MiB': 1024 ** 2,
+            'KB': 1000, 'KiB': 1024,
+            'B': 1
+        }
+
+        for unit, value in units.items():
+            if size.endswith(unit):
+                return int(float(size.removesuffix(unit)) * value)
+
+        raise ValueError(
+            'Invalid format. Expected integer or float ending with a data unit (B, KB, MiB,...).'
+        ) from err
+
+
+async def generate_data(size: int, buffer_size: int = 4 * 1024) -> AsyncGenerator[bytes, None]:
+    size_left = size
+
+    # https://github.com/tiangolo/fastapi/issues/5183
+    # https://github.com/encode/starlette/discussions/1776#discussioncomment-3207518
+
+    try:
+        while size_left > buffer_size:
+            size_left -= buffer_size
+            yield b'\0' * buffer_size
+            await asyncio.sleep(0)
+
+        yield b'\0' * size_left
+        await asyncio.sleep(0)
+    except asyncio.CancelledError as err:
+        raise GeneratorExit from err
+
+
+def load_database(path: str) -> dict:
+    with open(path, 'r', encoding='utf-8') as file:
+        return json.load(file)
+
+
+def save_database(path: str, database: dict) -> None:
+    with open(path, 'w', encoding='utf-8') as file:
+        json.dump(database, file, indent=2)

tests/__main__.py

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

tests/test_imports.py

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

tests/test_testdata.py

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