/ justapi / crates / justapi-py / python / justapi / test_body_schema_parsed.py
swadhinbiswas p1: stabilize route overriding, app.graphql, add_middleware, and PyO3 native handlers · 7/24/2026
f85aaeb
69 lines · 2.4 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
"""Regression test for P1.1 / BUG-3 — `body_schema` routes must receive the
parsed/validated body as a Python object, not raw bytes.

When a route is registered with a `justapi.Schema` body_schema, the dispatch
layer parses the JSON body on the fast path and attaches it to the `Request`
object, so `request.json()` and `request["body"]` return the parsed dict
(already validated). Legacy callable schemas keep the original behaviour.
"""
import json
import sys

import pytest

from justapi import JustAPIApp, Schema, Request
from justapi.testing import AsyncTestClient


class CreateProduct(Schema):
    name: str
    price: float
    stock: int = 0


def captured():
    state = {}

    async def create(req: Request):
        # `request.json()` should already be the validated/parsed dict, not raw
        # bytes. It must equal the posted payload (minus server-side mutation).
        body = req.json()
        state["json_type"] = type(body).__name__
        state["json_value"] = body
        # `request["body"]` should also return the parsed dict for schema routes.
        state["getitem_type"] = type(req["body"]).__name__
        state["validated_body"] = req.validated_body
        return {"ok": True, "name": body["name"], "price": body["price"]}

    return create, state


@pytest.mark.asyncio
async def test_body_schema_delivers_parsed_dict():
    app = JustAPIApp()
    handler, state = captured()
    app.post("/products", handler, body_schema=CreateProduct)

    async with AsyncTestClient(app) as client:
        payload = {"name": "Widget", "price": 12.5, "stock": 3}
        resp = await client.post("/products", json.dumps(payload).encode())
        assert resp["status"] == 200, resp.get("status")
        data = json.loads(resp["body"])
        assert data["name"] == "Widget"
        assert data["price"] == 12.5

    # The handler must have received a parsed dict, not raw bytes.
    assert state["json_type"] == "dict", state["json_type"]
    assert state["getitem_type"] == "dict", state["getitem_type"]
    assert state["validated_body"] == payload or state["validated_body"] == {
        "name": "Widget",
        "price": 12.5,
        "stock": 3,
    }
    assert state["json_value"]["name"] == "Widget"
    print("PASS: body_schema route received parsed dict (not raw bytes)")


if __name__ == "__main__":
    sys.exit(0 if __import__("asyncio").run(test_body_schema_delivers_parsed_dict()) is None else 1)