feat(openapi): AppListQuery — Pydantic validation for /apps

Replaces ad-hoc int(request.args.get(...)) parsing in AppListApi.get
with a typed Pydantic query model. Bad inputs (page=abc, limit=-1,
limit=500, mode=invalid, missing workspace_id) raise ValidationError
which the handler converts to 422 with field-level error detail
instead of 500 / silent empty page. Closes the mode whitelist via
AppMode enum.

Verified via direct unit tests on AppListQuery (no HTTP integration
tests required since the model carries the validation contract).
This commit is contained in:
GareArc 2026-05-05 20:02:47 -07:00
parent 87620050d7
commit 25034612b8
No known key found for this signature in database
2 changed files with 128 additions and 10 deletions

View File

@ -15,7 +15,8 @@ from typing import Any, cast
import sqlalchemy as sa
from flask import g, request
from flask_restx import Resource
from werkzeug.exceptions import BadRequest, NotFound
from pydantic import BaseModel, Field, ValidationError
from werkzeug.exceptions import NotFound, UnprocessableEntity
from controllers.common.fields import Parameters
from controllers.openapi import openapi_ns
@ -149,6 +150,22 @@ class AppDescribeApi(AppReadResource):
return AppDescribeResponse(info=info, parameters=parameters).model_dump(mode="json"), 200
class AppListQuery(BaseModel):
"""Query-param validator for `GET /openapi/v1/apps`.
`mode` is a closed set (AppMode) invalid values surface as 422
instead of returning silently-empty data. `workspace_id` is required;
page / limit have numeric bounds; name / tag have length caps.
"""
workspace_id: str
page: int = Field(1, ge=1)
limit: int = Field(20, ge=1, le=MAX_PAGE_LIMIT)
mode: AppMode | None = None
name: str | None = Field(None, max_length=200)
tag: str | None = Field(None, max_length=100)
@openapi_ns.route("/apps")
class AppListApi(Resource):
method_decorators = _APPS_READ_DECORATORS
@ -160,18 +177,19 @@ class AppListApi(Resource):
# (dfoe_ lacks apps:read). Defensive guard for future scope shifts.
return PaginationEnvelope[AppListRow].build(page=1, limit=0, total=0, items=[]).model_dump(mode="json"), 200
workspace_id = request.args.get("workspace_id")
if not workspace_id:
raise BadRequest("workspace_id query param is required")
try:
query = AppListQuery.model_validate(dict(request.args))
except ValidationError as e:
raise UnprocessableEntity(str(e.errors()))
workspace_id = query.workspace_id
require_workspace_member(ctx, workspace_id)
# NOTE: ad-hoc int(...) parsing below — Task 3 swaps this for AppListQuery.
page = int(request.args.get("page", "1"))
limit = min(int(request.args.get("limit", "20")), MAX_PAGE_LIMIT)
mode = request.args.get("mode")
name_filter = request.args.get("name")
tag_name = request.args.get("tag")
page = query.page
limit = query.limit
mode = query.mode.value if query.mode else None
name_filter = query.name
tag_name = query.tag
filters = [
App.tenant_id == workspace_id,

View File

@ -0,0 +1,100 @@
"""Unit tests for AppListQuery — the /apps query-param validator.
Runs against the model directly, not the HTTP layer. Pins:
- defaults match the plan (page=1, limit=20).
- workspace_id is required.
- numeric bounds enforced (page >= 1, limit in [1, MAX_PAGE_LIMIT]).
- mode validates against the AppMode enum.
- name and tag have length caps.
"""
from __future__ import annotations
import pytest
from pydantic import ValidationError
def test_defaults():
from controllers.openapi.apps import AppListQuery
q = AppListQuery.model_validate({"workspace_id": "ws-1"})
assert q.workspace_id == "ws-1"
assert q.page == 1
assert q.limit == 20
assert q.mode is None
assert q.name is None
assert q.tag is None
def test_workspace_id_required():
from controllers.openapi.apps import AppListQuery
with pytest.raises(ValidationError):
AppListQuery.model_validate({})
def test_page_must_be_positive():
from controllers.openapi.apps import AppListQuery
with pytest.raises(ValidationError):
AppListQuery.model_validate({"workspace_id": "ws-1", "page": 0})
with pytest.raises(ValidationError):
AppListQuery.model_validate({"workspace_id": "ws-1", "page": -1})
def test_page_rejects_non_integer_string():
from controllers.openapi.apps import AppListQuery
with pytest.raises(ValidationError):
AppListQuery.model_validate({"workspace_id": "ws-1", "page": "abc"})
def test_limit_must_be_positive():
from controllers.openapi.apps import AppListQuery
with pytest.raises(ValidationError):
AppListQuery.model_validate({"workspace_id": "ws-1", "limit": 0})
with pytest.raises(ValidationError):
AppListQuery.model_validate({"workspace_id": "ws-1", "limit": -1})
def test_limit_caps_at_max_page_limit():
from controllers.openapi._models import MAX_PAGE_LIMIT
from controllers.openapi.apps import AppListQuery
# Boundary accepts.
q = AppListQuery.model_validate({"workspace_id": "ws-1", "limit": MAX_PAGE_LIMIT})
assert q.limit == MAX_PAGE_LIMIT
# Just over rejects.
with pytest.raises(ValidationError):
AppListQuery.model_validate({"workspace_id": "ws-1", "limit": MAX_PAGE_LIMIT + 1})
def test_mode_whitelisted_against_app_mode():
from controllers.openapi.apps import AppListQuery
# Valid mode passes.
q = AppListQuery.model_validate({"workspace_id": "ws-1", "mode": "chat"})
assert q.mode is not None
assert q.mode.value == "chat"
# Invalid mode rejects.
with pytest.raises(ValidationError):
AppListQuery.model_validate({"workspace_id": "ws-1", "mode": "not-a-mode"})
def test_name_length_capped():
from controllers.openapi.apps import AppListQuery
AppListQuery.model_validate({"workspace_id": "ws-1", "name": "x" * 200})
with pytest.raises(ValidationError):
AppListQuery.model_validate({"workspace_id": "ws-1", "name": "x" * 201})
def test_tag_length_capped():
from controllers.openapi.apps import AppListQuery
AppListQuery.model_validate({"workspace_id": "ws-1", "tag": "x" * 100})
with pytest.raises(ValidationError):
AppListQuery.model_validate({"workspace_id": "ws-1", "tag": "x" * 101})