group_fqcd_jr/tests/unit/api/test_frontend_boundaries.py

"""前端 ↔ 后端边界守卫（`W11` 批次补测，答辩前的前端边界盘点产物）。

## 为什么要单开这一组

答辩前的边界盘点发现一类**同族缺口**：前端 `widget.js` 的输入框已经用
`maxlength="8000"` 截断，但服务端请求模型当初**不设上限**（或上限与落库列宽不符）。
前端截断只是体验，绕过前端（curl / 脚本 / 改前端）直发就能把任意长度文本或超长
ID 送进入库、入模型链路；MySQL 严格模式下这时表现为 `500` 而不是 `422`——
一次本该是"参数不合格"的失败，被冒泡成了服务端故障。

本文件把三类不变量钉死：

1. **口径一致**：前端输入框的 `maxlength` 必须等于服务端 `max_length`，
   否则前端"允许输入"与服务端"允许接收"会静默分叉；
2. **入参上限 = 落库列宽**：请求模型的 `max_length` 必须 ≤ 对应列的 `String(n)`
   （见 `app/model/session.py` / `app/model/platform.py` / `app/model/conversation.py`），
   超限必须在入口变 `422`，而不是落库时才炸 `500`；
3. **路径参数同宽**：`{session_id}` / `{run_id}` / `{handover_id}` 等路径参数与列宽
   对齐——这一层用 OpenAPI 契约断言，避免单元环境连库。

另含一条端点表守卫：`api-client.js` 里 `ENDPOINTS` 的每一项都必须在 `app.openapi()`
里真实存在。历史上 `HEALTH` 写的是 `/health`，而后端只挂了 `/internal/health/live`
与 `/internal/health/ready`——前端因此恒判"后端未连接"，正是缺了这条守卫。
"""

from __future__ import annotations

import re
from pathlib import Path
from typing import Any

import pytest
from fastapi.testclient import TestClient
from pydantic import ValidationError

from app.api.dependencies.auth import build_request_context
from app.api.schemas.agent_runs import AgentRunCreateRequest
from app.api.schemas.conversations import FeedbackRequest
from app.core.contracts import RequestContext
from app.main import create_app
from app.model.conversation import ConversationFeedback
from app.model.platform import RequestIdempotency
from app.model.session import ConversationSession

ROOT = Path(__file__).resolve().parents[3]
PORTAL = ROOT / "app" / "static" / "portal"
WIDGET = PORTAL / "common" / "customer-service-widget" / "widget.js"
API_CLIENT = PORTAL / "common" / "api-client.js"

AGENT_RUNS = "/api/v1/agent-runs"

#: 一份**合法**的建运行请求，边界用例只在它上面改一个字段。
VALID_RUN: dict[str, Any] = {
    "agent_type": "customer_service",
    "message": "查询我的基金持仓",
    "session_id": "session-1",
    "idempotency_key": "1234567890123456",
}


def _column_width(model: type, column: str) -> int:
    length = model.__table__.c[column].type.length  # type: ignore[attr-defined]
    assert isinstance(length, int), f"{model.__name__}.{column} 不是定长列，无法比对"
    return length


def _rejected(**overrides: Any) -> bool:
    payload = {**VALID_RUN, **overrides}
    try:
        AgentRunCreateRequest(**payload)
    except ValidationError:
        return True
    return False


def _client(permissions: tuple[str, ...] = ("agent:run",)) -> TestClient:
    application = create_app()

    async def context() -> RequestContext:
        return RequestContext(
            user_id="1", trace_id="fe-boundary", portal="public", permissions=permissions
        )

    application.dependency_overrides[build_request_context] = context
    return TestClient(application)


def _endpoints() -> dict[str, tuple[str, str]]:
    source = API_CLIENT.read_text(encoding="utf-8")
    pairs = re.findall(r"(\w+):\s*\{\s*method:\s*'(\w+)',\s*path:\s*'([^']+)'", source)
    assert len(pairs) >= 50, "端点表解析异常，正则可能已与 api-client.js 脱节"
    return {name: (method, path) for name, method, path in pairs}


def _normalized(path: str) -> str:
    """把 `{sessionId}` / `{session_id}` 之类占位符统一成 `{}` 再比对。"""
    return re.sub(r"\{[^}]+\}", "{}", path)


# --------------------------------------------------------------------------- #
# 一、口径一致：前端输入框 ↔ 服务端校验
# --------------------------------------------------------------------------- #

def test_widget_message_maxlength_matches_backend_limit() -> None:
    """`widget.js` 的 `maxlength` 必须与 `AgentRunCreateRequest.message` 的上限逐字一致。

    这条断言是"两端同宽"的唯一防线：只改一边（比如前端产品改到 4000、或后端放宽
    到 20000）都会立刻变红，而不会拖到线上才以"用户输入被静默截断/提交 500"暴露。
    """
    source = WIDGET.read_text(encoding="utf-8")
    matched = re.search(r'name="message"[^>]*maxlength="(\d+)"', source)
    assert matched, "widget.js 里找不到 message 输入框的 maxlength"
    frontend_limit = int(matched.group(1))

    assert not _rejected(message="问" * frontend_limit), "前端允许的长度，后端不该拒绝"
    assert _rejected(message="问" * (frontend_limit + 1)), "前端截断之外的长度，后端必须拒绝"


def test_message_and_session_bounds_match_storage_widths() -> None:
    """入参上限必须 ≤ 落库列宽，否则超限会以 500 而不是 422 出现。"""
    session_width = _column_width(ConversationSession, "session_id")
    idempotency_width = _column_width(RequestIdempotency, "idempotency_key")

    assert not _rejected(session_id="s" * session_width)
    assert _rejected(session_id="s" * (session_width + 1))

    assert not _rejected(idempotency_key="k" * idempotency_width)
    assert _rejected(idempotency_key="k" * (idempotency_width + 1))


def test_feedback_type_bound_matches_storage_width() -> None:
    width = _column_width(ConversationFeedback, "feedback_type")
    assert FeedbackRequest(rating=1, feedback_type="t" * width).feedback_type is not None
    with pytest.raises(ValidationError):
        FeedbackRequest(rating=1, feedback_type="t" * (width + 1))


def test_degenerate_inputs_are_rejected_at_the_model_layer() -> None:
    assert _rejected(message="")
    assert _rejected(session_id="")
    assert _rejected(idempotency_key="123456789012345")


# --------------------------------------------------------------------------- #
# 二、HTTP 层：超限必须是 422 信封，不是 500
# --------------------------------------------------------------------------- #

@pytest.mark.parametrize(
    ("overrides", "field"),
    [
        ({"message": "问" * 8001}, "body.message"),
        ({"message": ""}, "body.message"),
        ({"session_id": "s" * 65}, "body.session_id"),
        ({"session_id": ""}, "body.session_id"),
        ({"idempotency_key": "k" * 65}, "body.idempotency_key"),
        ({"idempotency_key": "123456789012345"}, "body.idempotency_key"),
    ],
)
def test_oversized_body_fields_return_422_envelope(
    overrides: dict[str, Any], field: str
) -> None:
    with _client() as client:
        response = client.post(
            AGENT_RUNS, json={**VALID_RUN, **overrides},
            headers={"X-Trace-ID": "fe-boundary"},
        )

    assert response.status_code == 422, response.text
    body = response.json()
    assert set(body) == {"error", "meta"}
    error = body["error"]
    assert error["code"] == "AGENT_INPUT_INVALID"
    assert error["retryable"] is False
    assert field in {item["field"] for item in error["field_errors"]}
    assert body["meta"]["trace_id"] == "fe-boundary"


@pytest.mark.parametrize(
    ("method", "path"),
    [
        ("GET", "/api/v1/conversations/" + "s" * 65),
        ("POST", "/api/v1/conversations/" + "s" * 65 + "/closures"),
        ("POST", "/api/v1/conversations/" + "s" * 65 + "/handover-requests"),
        ("GET", "/api/v1/conversations/" + "s" * 65 + "/messages"),
        ("GET", "/api/v1/agent-runs/" + "r" * 65),
        ("GET", "/api/v1/agent-runs/" + "r" * 65 + "/events"),
        ("POST", "/api/v1/agent-runs/" + "r" * 65 + "/cancellations"),
        ("GET", "/api/v1/handover-requests/" + "t" * 65),
    ],
)
def test_oversized_path_params_return_422(method: str, path: str) -> None:
    """路径参数超宽必须在路由层被拦下——不能带着超长 ID 去查库/写库。"""
    with _client(permissions=("agent:run", "conversation:read", "handover:write")) as client:
        response = client.request(method, path, json={} if method == "POST" else None)

    assert response.status_code == 422, response.text
    assert response.json()["error"]["code"] == "AGENT_INPUT_INVALID"


@pytest.mark.parametrize(
    ("method", "route", "param"),
    [
        ("GET", "/api/v1/conversations/{session_id}", "session_id"),
        ("GET", "/api/v1/conversations/{session_id}/messages", "session_id"),
        ("POST", "/api/v1/conversations/{session_id}/closures", "session_id"),
        ("POST", "/api/v1/conversations/{session_id}/handover-requests", "session_id"),
        ("GET", "/api/v1/agent-runs/{run_id}", "run_id"),
        ("GET", "/api/v1/agent-runs/{run_id}/events", "run_id"),
        ("POST", "/api/v1/agent-runs/{run_id}/cancellations", "run_id"),
        ("GET", "/api/v1/handover-requests/{handover_id}", "handover_id"),
    ],
)
def test_path_params_declare_width_constraints_in_openapi(
    method: str, route: str, param: str
) -> None:
    """契约层守卫：OpenAPI 必须暴露路径参数的长度上限（前端据此生成校验/文档）。"""
    spec = create_app().openapi()["paths"][route][method.lower()]
    parameters = {item["name"]: item for item in spec.get("parameters", [])}
    assert param in parameters, f"{route} 未声明路径参数 {param}"
    schema = parameters[param]["schema"]
    assert schema.get("maxLength") == 64
    assert schema.get("pattern") == r"^[A-Za-z0-9_-]+$"


@pytest.mark.parametrize("path", ["/api/v1/conversations/有中文", "/api/v1/agent-runs/run id"])
def test_path_params_reject_illegal_characters(path: str) -> None:
    with _client(permissions=("agent:run", "conversation:read")) as client:
        response = client.get(path)
    assert response.status_code == 422, response.text


# --------------------------------------------------------------------------- #
# 三、鉴权/权限：边界收紧了，入口守卫不能跟着松
# --------------------------------------------------------------------------- #

def test_agent_run_requires_authentication() -> None:
    with TestClient(create_app()) as client:
        response = client.post(AGENT_RUNS, json=VALID_RUN)
    assert response.status_code == 401
    assert response.json()["error"]["code"] == "AUTHENTICATION_REQUIRED"


def test_missing_agent_permission_is_forbidden() -> None:
    """权限集里没有 `agent:run` 就必须 403——长度约束不得掩盖权限判定。"""
    with _client(permissions=()) as client:
        response = client.post(AGENT_RUNS, json=VALID_RUN)
    assert response.status_code == 403, response.text
    assert response.json()["error"]["code"] == "AGENT_PERMISSION_DENIED"


def test_visitor_role_can_run_but_only_with_public_scope() -> None:
    """访客**可以**提交运行（`agent:run` 是设计内），但权限集里不得有任何个人数据权限。

    ⚠️ 这一条曾经被写反。盘点时的第一直觉是「访客调 R001 必须 403」，实测打到
    真机才发现访客令牌**本来就带** `agent:run`——客服浮窗的匿名提问正是走这条路；
    真正的不变量不是「访客不能提问」，而是「访客只能拿公共档位、不得触个人数据」。
    因此这里的判据改成**权限集内容**：一旦有人往访客权限里追加
    `customer:` / `profile:` / `memory:` / `handover:` / `conversation:` 前缀的权限码，
    这条会立刻变红（`app/core/actor.py` 的注释也明确写了"不允许在这里追加权限码"）。
    """
    from app.core.actor import VISITOR_DATA_SCOPE, VISITOR_PERMISSIONS

    assert set(VISITOR_PERMISSIONS) == {"agent:run", "knowledge:query"}
    assert VISITOR_DATA_SCOPE == "public"
    personal_prefixes = ("customer:", "profile:", "memory:", "handover:", "conversation:")
    assert [
        name for name in VISITOR_PERMISSIONS if name.startswith(personal_prefixes)
    ] == []


# --------------------------------------------------------------------------- #
# 四、端点表守卫：前端调的每一条都必须是后端真实存在的路由
# --------------------------------------------------------------------------- #

def test_frontend_endpoint_table_matches_real_routes() -> None:
    spec = create_app().openapi()["paths"]
    available = {
        (method.upper(), _normalized(path))
        for path, operations in spec.items()
        for method in operations
    }

    missing = [
        f"{name} {method} {path}"
        for name, (method, path) in _endpoints().items()
        if (method.upper(), _normalized(path)) not in available
    ]

    assert not missing, "api-client.js 引用了后端不存在的端点：\n" + "\n".join(missing)


def test_health_probe_uses_a_registered_route() -> None:
    """`HEALTH` 历史上写的是 `/health`（后端没有），前端因此恒判"后端离线"。"""
    method, path = _endpoints()["HEALTH"]
    assert method == "GET"
    assert path in create_app().openapi()["paths"]