/ justapi / crates / justapi-py / python / justapi / test_openapi_parity.py
swadhinbiswas WIP save-point: production hardening, H3 removal, fmt gate, and accumulated phase work · 7/15/2026
2080d40
143 lines · 4.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
"""Integration test for FastAPI APIRouter parity (OpenAPI metadata, head/options/trace, frontend, include_router)."""
import json
import os
import subprocess
import sys
import tempfile
import time
import urllib.request
import urllib.error

SERVER_SCRIPT = r"""
import tempfile, os
from justapi import JustAPIApp, APIRouter

app = JustAPIApp()

static_dir = tempfile.mkdtemp()
with open(os.path.join(static_dir, "index.html"), "w") as f:
    f.write("<html><body>SPA</body></html>")

async def list_items(request):
    return [{"id": 1}]

async def create_item(request):
    return {"id": 2}

async def hidden(request):
    return {"ok": True}

async def head_items(request):
    return {"ok": True}

async def options_items(request):
    return {"ok": True}

async def trace_items(request):
    return {"ok": True}

app.get(
    "/items",
    list_items,
    tags=["items"],
    summary="List items",
    description="Return all items.",
    deprecated=True,
    responses={404: {"description": "Not found"}},
    operation_id="list_items",
    openapi_extra={"x-custom": "hello"},
)
app.post("/items", create_item, tags=["items"], status_code=201)
app.get("/hidden", hidden, include_in_schema=False)
# Explicit head/options/trace are registered on paths without a GET (GET
# already auto-registers HEAD, mirroring FastAPI/Starlette behavior).
app.head("/head1", head_items)
app.options("/opts1", options_items)
app.trace("/trace1", trace_items)

admin = APIRouter(prefix="/admin", tags=["admin"])
@admin.get("/ping", tags=["health"])
async def ping(request):
    return {"pong": True}
app.include_router(admin)

app.frontend("/static", static_dir, html=True)

app.run("127.0.0.1:9871")
"""

ADDR = "127.0.0.1:9871"


def _get(path, method="GET"):
    req = urllib.request.Request(f"http://{ADDR}{path}", method=method)
    return urllib.request.urlopen(req)


def test_openapi_parity():
    proc = subprocess.Popen(
        [sys.executable, "-c", SERVER_SCRIPT],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    time.sleep(1.0)
    try:
        # OpenAPI metadata
        resp = _get("/openapi.json")
        spec = json.loads(resp.read())
        paths = spec["paths"]

        # GET /items metadata
        get_items = paths["/items"]["get"]
        assert get_items["tags"] == ["items"], get_items
        assert get_items["summary"] == "List items"
        assert get_items["description"] == "Return all items."
        assert get_items.get("deprecated") is True
        assert get_items["operationId"] == "list_items"
        assert get_items["x-custom"] == "hello", get_items
        assert "404" in get_items["responses"]
        # default success code 200
        assert "200" in get_items["responses"]

        # POST /items custom status code
        post_items = paths["/items"]["post"]
        assert "201" in post_items["responses"], post_items

        # HEAD is excluded from OpenAPI (auto-registered from GET)
        assert "head" not in paths["/items"]
        # Explicit OPTIONS / TRACE are present on their own paths
        assert "head" in paths["/head1"]
        assert "options" in paths["/opts1"]
        assert "trace" in paths["/trace1"]

        # include_in_schema=False route is excluded
        assert "/hidden" not in paths

        # include_router merged tags: /admin/ping has router tags + route tags
        ping = paths["/admin/ping"]["get"]
        assert "admin" in ping["tags"], ping["tags"]
        assert "health" in ping["tags"], ping["tags"]

        # Frontend SPA serving + fallback
        resp = _get("/static/")
        assert b"SPA" in resp.read(), "frontend index not served"
        resp = _get("/static/does-not-exist")
        assert b"SPA" in resp.read(), "frontend fallback (index.html) not served"

        # head/options respond at runtime
        assert _get("/head1", method="HEAD").status == 200
        assert _get("/opts1", method="OPTIONS").status == 200

        print("PASS: openapi parity + frontend + methods")
    finally:
        proc.terminate()
        try:
            proc.communicate(timeout=5)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.communicate()


if __name__ == "__main__":
    test_openapi_parity()