PythonでJSONを返す簡易モックサーバーの構築

start_response(status: str, response_headers: list[tuple[str, str]], exc_info=None)

from wsgiref.simple_server import make_server
import json
import time

PORT = 8081

# モックデータの定義
# PATH_INFO (リクエストパス) に応じて返すJSONデータを設定
MOCK_DATA_SETTINGS = [
    {
        "PATH": "/api/1",
        "VALUE": {"items": [{"item1": "test1"}, {"item2": "test2"}]},
    },
    {
        "PATH": "/api/2",
        "VALUE": {"items2": [{"itemA": "testA"}, {"itemB": "testB"}]},
    },
]

ALLOWED_METHODS = "GET, POST, OPTIONS"
ALLOWED_HEADERS = "Content-Type"


def cors_headers():
    return [
        ("Access-Control-Allow-Origin", "*"),
        ("Access-Control-Allow-Methods", ALLOWED_METHODS),
        ("Access-Control-Allow-Headers", ALLOWED_HEADERS),
        ("Access-Control-Max-Age", "86400"),
    ]


def application(environ, start_response):
    """
    WSGIアプリケーションのエントリポイント。
    リクエストに応じて適切なJSONデータ・エラー・CORSヘッダーを返す。
    """
    method = environ.get("REQUEST_METHOD", "GET")
    path = environ.get("PATH_INFO", "/")

    # --- CORSプリフライト ---
    # ブラウザは Content-Type: application/json 付きの POST/PUT/DELETE などを
    # "simple request" とみなさないため、実リクエストの前に OPTIONS で
    # プリフライトを送る。ここで Access-Control-Allow-* を返さないと
    # 実リクエストは送信されずブラウザ側でブロックされる。
    if method == "OPTIONS":
        start_response("204 No Content", cors_headers())
        return [b""]

    # --- エラー系エンドポイント（意図的に 500 を返す） ---
    if path.startswith("/api/error"):
        headers = [("Content-type", "text/plain; charset=utf-8")] + cors_headers()
        start_response("500 Internal Server Error", headers)
        return [b"500 Internal Server Error"]

    # --- 疑似的に遅いエンドポイント（同時実行性デモ用） ---
    if path.startswith("/api/slow"):
        time.sleep(0.2)
        headers = [
            ("Content-type", "application/json; charset=utf-8"),
        ] + cors_headers()
        start_response("200 OK", headers)
        return [json.dumps({"slept": 0.2}).encode("utf-8")]

    # --- 通常のモックデータ ---
    for setting in MOCK_DATA_SETTINGS:
        if path.startswith(setting["PATH"]):
            headers = [
                ("Content-type", "application/json; charset=utf-8"),
            ] + cors_headers()
            start_response("200 OK", headers)
            return [json.dumps(setting["VALUE"]).encode("utf-8")]

    # --- どのルールにもマッチしない場合は 404 ---
    headers = [("Content-type", "text/plain; charset=utf-8")] + cors_headers()
    start_response("404 Not Found", headers)
    return [b"404 Not Found"]


if __name__ == "__main__":
    httpd = make_server("", PORT, application)
    print(f"Serving on port {PORT}...")
    httpd.serve_forever()

import subprocess, sys, time, requests

proc = subprocess.Popen([sys.executable, "mock_server.py"], ...)
try:
    r = requests.get("http://localhost:8081/api/1")
    print(r.status_code, r.text)

    r = requests.get("http://localhost:8081/api/2")
    print(r.status_code, r.text)

    r = requests.get("http://localhost:8081/api/nonexistent")
    print(r.status_code, r.text)

    r = requests.get("http://localhost:8081/api/error")
    print(r.status_code, r.text)
finally:
    proc.terminate()  # デモ終了後は必ずプロセスを後始末する
    proc.wait(timeout=5)

=== GET /api/1 (200 OK) ===
status: 200
headers: Content-Type=application/json; charset=utf-8
body: {"items": [{"item1": "test1"}, {"item2": "test2"}]}

=== GET /api/2 (200 OK) ===
status: 200
body: {"items2": [{"itemA": "testA"}, {"itemB": "testB"}]}

=== GET /api/nonexistent (404 Not Found) ===
status: 404
body: 404 Not Found

=== GET /api/error (500 Internal Server Error) ===
status: 500
body: 500 Internal Server Error

r = requests.options(
    "http://localhost:8081/api/1",
    headers={
        "Origin": "https://example.com",
        "Access-Control-Request-Method": "POST",
        "Access-Control-Request-Headers": "Content-Type",
    },
)

=== OPTIONS /api/1 (CORS preflight) ===
status: 204
Access-Control-Allow-Origin: *
Access-Control-Allow-Methods: GET, POST, OPTIONS
Access-Control-Allow-Headers: Content-Type
Access-Control-Max-Age: 86400
body length: 0

=== Actual request following preflight: GET /api/1 with Origin header ===
status: 200
Access-Control-Allow-Origin: *
body: {"items": [{"item1": "test1"}, {"item2": "test2"}]}

from socketserver import ThreadingMixIn
from wsgiref.simple_server import WSGIServer, make_server
import time

PORT = 8082


def application(environ, start_response):
    time.sleep(0.2)  # 疑似的なI/O待ち
    start_response("200 OK", [("Content-type", "text/plain; charset=utf-8")])
    return [b"ok"]


class ThreadingWSGIServer(ThreadingMixIn, WSGIServer):
    daemon_threads = True  # プロセス終了時にリクエスト処理スレッドを道連れにする


def main(mode):
    if mode == "threaded":
        httpd = make_server("", PORT, application, server_class=ThreadingWSGIServer)
    else:
        httpd = make_server("", PORT, application)  # デフォルトはシングルスレッド
    httpd.serve_forever()

from concurrent.futures import ThreadPoolExecutor
import time, requests

def run_one_request(_):
    t0 = time.perf_counter()
    requests.get("http://localhost:8082/api/slow")
    return time.perf_counter() - t0

t0 = time.perf_counter()
with ThreadPoolExecutor(max_workers=10) as ex:
    latencies = list(ex.map(run_one_request, range(10)))
total = time.perf_counter() - t0