httpbin-like server with extra capabilities like delayed errors

server.py

#!/usr/bin/env python3

import json
import sys
import time
from http.server import BaseHTTPRequestHandler, HTTPServer, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse

sessions = {}


class S(BaseHTTPRequestHandler):
    def do_GET(self):
        parsed_url = urlparse(self.path)
        qs = parse_qs(parsed_url.query)

        if "wait_ms" in qs:
            try:
                wait_ms = int(qs["wait_ms"][0])
                if wait_ms > 0:
                    time.sleep(wait_ms / 1_000)
            except ValueError:
                pass

        status_code = 200
        if "status_code" in qs:
            try:
                status_code = int(qs["status_code"][0])
            except ValueError:
                pass

        # Look up existing session.
        if "session_id" in qs:
            global sessions
            session_id = qs["session_id"][0]
            session = sessions.get(session_id)
            if session is not None:
                # Existing session.
                if time.time() > session["session_end_time"]:
                    del sessions[session_id]
                else:
                    if (
                        "status_code_after" in session
                        and "status_code_after_time" in session
                    ):
                        if time.time() > session["status_code_after_time"]:
                            status_code = session["status_code_after"]
            else:
                # New session.
                session = {"session_end_time": time.time() + 30.0}

                if "session_ttl_ms" in qs:
                    session["session_end_time"] = (
                        time.time() + int(qs["session_ttl_ms"][0]) / 1_000
                    )
                if "status_code_after_ms" in qs:
                    session["status_code_after_time"] = (
                        time.time() + int(qs["status_code_after_ms"][0]) / 1_000
                    )
                if "status_code_after" in qs:
                    session["status_code_after"] = int(qs["status_code_after"][0])

                # Store the session.
                sessions[session_id] = session

        self.send_response(status_code)

        send_headers = False
        if status_code < 400:
            send_headers = True

        elif 400 < status_code < 500 and "headers_if_4xx" in qs:
            send_headers = True

        elif 500 < status_code and "headers_if_5xx" in qs:
            send_headers = True

        if send_headers and "header" in qs:
            for header in qs["header"]:
                key, val = header.split("=", 1)
                self.send_header(key, val)

        self.end_headers()

        body = b"Hello, World!"
        if "body" in qs:
            body = qs["body"][0].encode("utf-8")

        self.wfile.write(body)


def run(server_class=HTTPServer, handler_class=S, addr="localhost", port=8080):
    server_address = (addr, port)
    httpd = server_class(server_address, handler_class)

    print(f"Starting httpd server on {addr}:{port}")
    httpd.serve_forever()


if __name__ == "__main__":
    run(
        server_class=ThreadingHTTPServer,
        addr=sys.argv[1],
        port=int(sys.argv[2]),
    )