You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
142 lines
3.8 KiB
Python
142 lines
3.8 KiB
Python
from fastapi import APIRouter, HTTPException
|
|
from fastapi.testclient import TestClient
|
|
from starlette.responses import PlainTextResponse
|
|
|
|
from iti import create_app
|
|
from iti.config import BaseConfig
|
|
from iti.exceptions import BizError
|
|
from iti.limiter import limit
|
|
from iti.responses import ok, raw_response
|
|
|
|
|
|
class RoutesModule:
|
|
name = "routes"
|
|
|
|
def register_routes(self, app):
|
|
router = APIRouter()
|
|
|
|
@router.get("/demo")
|
|
def demo():
|
|
return ok({"value": 1})
|
|
|
|
@router.get("/auto")
|
|
def auto():
|
|
return {"value": 2}
|
|
|
|
@router.get("/raw")
|
|
@raw_response
|
|
def raw():
|
|
return {"value": 3}
|
|
|
|
@router.get("/text")
|
|
def text():
|
|
return PlainTextResponse("ok")
|
|
|
|
@router.get("/boom")
|
|
def boom():
|
|
raise BizError("业务失败", code=400)
|
|
|
|
@router.get("/http-error")
|
|
def http_error():
|
|
raise HTTPException(status_code=418, detail={"message": "茶壶错误"})
|
|
|
|
@router.get("/limited", dependencies=[limit("1 per minute")])
|
|
def limited():
|
|
return ok()
|
|
|
|
app.include_router(router)
|
|
|
|
|
|
def make_app(**config_values):
|
|
config = BaseConfig(
|
|
database_url="sqlite+pysqlite:///:memory:",
|
|
testing=True,
|
|
**config_values,
|
|
)
|
|
return create_app(modules=[RoutesModule()], config_mapping=config)
|
|
|
|
|
|
def test_framework_health_routes():
|
|
client = TestClient(make_app())
|
|
|
|
assert client.get("/health").json() == {"status": "ok"}
|
|
assert client.get("/ready").json() == {"status": "ok"}
|
|
|
|
|
|
def test_envelope_and_error_handlers():
|
|
client = TestClient(make_app())
|
|
|
|
assert client.get("/demo").json() == {
|
|
"data": {"value": 1},
|
|
"code": 200,
|
|
"message": "成功",
|
|
}
|
|
response = client.get("/boom")
|
|
assert response.status_code == 200
|
|
assert response.json()["message"] == "业务失败"
|
|
|
|
|
|
def test_http_errors_use_envelope():
|
|
client = TestClient(make_app())
|
|
|
|
not_found = client.get("/")
|
|
assert not_found.status_code == 200
|
|
assert not_found.json() == {
|
|
"data": None,
|
|
"code": 404,
|
|
"message": "Not Found",
|
|
}
|
|
|
|
method_not_allowed = client.post("/demo")
|
|
assert method_not_allowed.status_code == 200
|
|
assert method_not_allowed.json()["code"] == 405
|
|
assert method_not_allowed.json()["message"] == "Method Not Allowed"
|
|
|
|
http_error = client.get("/http-error")
|
|
assert http_error.status_code == 200
|
|
assert http_error.json() == {
|
|
"data": {"message": "茶壶错误"},
|
|
"code": 418,
|
|
"message": "茶壶错误",
|
|
}
|
|
|
|
|
|
def test_auto_envelope_wraps_plain_json_and_raw_can_skip():
|
|
client = TestClient(make_app(raw_response_paths=["/health", "/ready", "/raw-path"]))
|
|
|
|
assert client.get("/auto").json() == {
|
|
"data": {"value": 2},
|
|
"code": 200,
|
|
"message": "成功",
|
|
}
|
|
assert client.get("/raw").json() == {"value": 3}
|
|
assert client.get("/text").text == "ok"
|
|
|
|
|
|
def test_rate_limit_dependency():
|
|
client = TestClient(make_app(ratelimit_enabled=True))
|
|
|
|
assert client.get("/limited").status_code == 200
|
|
response = client.get("/limited")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["code"] == 429
|
|
|
|
|
|
def test_request_log_includes_business_code(monkeypatch):
|
|
records = []
|
|
|
|
def capture(message, *args, **kwargs):
|
|
records.append((message, args, kwargs))
|
|
|
|
monkeypatch.setattr("iti.app.logger.info", capture)
|
|
client = TestClient(make_app())
|
|
|
|
assert client.get("/auto").json()["code"] == 200
|
|
assert records[-1][1][3] == 200
|
|
assert records[-1][2]["extra"]["response_code"] == 200
|
|
|
|
assert client.get("/boom").json()["code"] == 400
|
|
assert records[-1][1][3] == 400
|
|
assert records[-1][2]["extra"]["response_code"] == 400
|