deer-flow/backend/tests/test_run_request_validation.py

"""Contract tests for the LangGraph-compatible run request boundary."""

from __future__ import annotations

import asyncio
from typing import Any

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.gateway.routers import runs

SUPPORTED_STREAM_MODES = {
    "values",
    "messages-tuple",
    "updates",
    "debug",
    "tasks",
    "checkpoints",
    "custom",
}


@pytest.fixture
def client() -> TestClient:
    app = FastAPI()
    app.include_router(runs.router)
    return TestClient(app, raise_server_exceptions=False)


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("webhook", "https://example.com/callback"),
        ("stream_resumable", True),
        ("on_completion", "complete"),
        ("on_completion", "continue"),
        ("on_completion", "keep"),
        ("on_completion", "delete"),
        ("after_seconds", 1.5),
        ("if_not_exists", "reject"),
        ("feedback_keys", []),
        ("multitask_strategy", "enqueue"),
    ],
)
def test_run_request_rejects_each_unsupported_option_with_exact_422(
    client: TestClient,
    field: str,
    value: Any,
) -> None:
    response = client.post("/api/runs/stream", json={field: value})

    assert response.status_code == 422
    assert response.json() == {
        "detail": [
            {
                "type": "unsupported_run_option",
                "loc": ["body", field],
                "msg": f"Run option '{field}' is not supported by DeerFlow",
                "input": value,
                "ctx": {"option": field},
            }
        ]
    }


@pytest.mark.parametrize(
    "stream_mode",
    [
        "messages",
        "events",
        "tools",
        ["values", "events"],
        ["events", "tools"],
    ],
)
def test_run_request_rejects_unsupported_stream_modes_with_exact_422(
    client: TestClient,
    stream_mode: str | list[str],
) -> None:
    response = client.post("/api/runs/stream", json={"stream_mode": stream_mode})
    requested = [stream_mode] if isinstance(stream_mode, str) else stream_mode
    unsupported = list(dict.fromkeys(mode for mode in requested if mode not in SUPPORTED_STREAM_MODES))
    mode_list = ", ".join(unsupported)

    assert response.status_code == 422
    assert response.json() == {
        "detail": [
            {
                "type": "unsupported_stream_mode",
                "loc": ["body", "stream_mode"],
                "msg": f"Unsupported stream mode(s): {mode_list}",
                "input": stream_mode,
                "ctx": {"modes": mode_list},
            }
        ]
    }


def test_run_request_keeps_supported_modes_and_compatibility_defaults() -> None:
    from app.gateway.routers.thread_runs import RunCreateRequest

    body = RunCreateRequest(
        stream_mode=list(SUPPORTED_STREAM_MODES),
        webhook=None,
        stream_resumable=None,
        on_completion=None,
        after_seconds=None,
        if_not_exists="create",
        feedback_keys=None,
    )

    assert set(body.stream_mode or []) == SUPPORTED_STREAM_MODES
    assert body.on_completion is None
    assert body.if_not_exists == "create"


@pytest.mark.parametrize(
    "artifact_delivery",
    [
        {"required": True, "tool": "present_files"},
        {},
        None,
        "present_files",
        [],
    ],
)
def test_run_request_rejects_removed_artifact_delivery_override(
    client: TestClient,
    artifact_delivery: Any,
) -> None:
    response = client.post(
        "/api/runs/stream",
        json={"artifact_delivery": artifact_delivery},
    )

    assert response.status_code == 422


def _sdk_default_payload(method: str) -> dict[str, Any]:
    """Capture the body a stock ``langgraph_sdk`` client sends for a default run."""
    from langgraph_sdk.client import get_client

    client = get_client(url="http://gateway.invalid")
    captured: dict[str, Any] = {}

    def capture(*_args: Any, json: dict[str, Any] | None = None, **_kwargs: Any) -> None:
        captured.update(json or {})

    async def capture_post(*args: Any, **kwargs: Any) -> dict[str, Any]:
        capture(*args, **kwargs)
        return {}

    if method == "stream":
        client.runs.http.stream = capture  # type: ignore[method-assign]
        # ``stream()`` is a sync factory: the payload is built and handed to the
        # transport before the returned async iterator is consumed.
        client.runs.stream("thread-id", "deerflow", input={"messages": []})
    else:
        client.runs.http.post = capture_post  # type: ignore[method-assign]
        asyncio.run(client.runs.create("thread-id", "deerflow", input={"messages": []}))
    return captured


@pytest.mark.parametrize("method", ["stream", "create"])
def test_gateway_accepts_langgraph_sdk_default_payload(client: TestClient, method: str) -> None:
    """The SDK's own defaults must never be rejected as unsupported options (#4466).

    ``stream_resumable`` defaults to ``False`` rather than ``None``, so the SDK's
    drop-``None`` payload filter always forwards it; every IM channel run goes
    through this path.
    """
    payload = _sdk_default_payload(method)

    assert payload["stream_resumable"] is False, "SDK contract changed; update this test"
    response = client.post("/api/runs/stream", json=payload)

    assert response.status_code != 422, response.text


@pytest.mark.parametrize(
    "payload",
    [
        {"multitask_strategy": []},
        {"stream_mode": [{}]},
    ],
)
def test_malformed_option_types_remain_validation_errors(client: TestClient, payload: dict[str, Any]) -> None:
    response = client.post("/api/runs/stream", json=payload)

    assert response.status_code == 422


@pytest.mark.parametrize(
    "field",
    [
        "multitask_strategy",
        "if_not_exists",
    ],
)
def test_string_run_options_keep_native_type_errors(client: TestClient, field: str) -> None:
    response = client.post("/api/runs/stream", json={field: []})

    assert response.status_code == 422
    assert response.json()["detail"][0]["type"] == "literal_error"


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("checkpoint_during", False),
        ("durability", "sync"),
    ],
)
def test_run_request_rejects_unknown_sdk_options_with_exact_422(
    client: TestClient,
    field: str,
    value: Any,
) -> None:
    response = client.post("/api/runs/stream", json={field: value})

    assert response.status_code == 422
    assert response.json() == {
        "detail": [
            {
                "type": "extra_forbidden",
                "loc": ["body", field],
                "msg": "Extra inputs are not permitted",
                "input": value,
            }
        ]
    }


def test_openapi_stream_mode_enum_matches_runtime_support() -> None:
    app = FastAPI()
    app.include_router(runs.router)
    openapi = app.openapi()
    schemas = openapi["components"]["schemas"]
    stream_mode_schema = schemas["RunCreateRequest"]["properties"]["stream_mode"]

    enums: list[str] = []

    def collect_enums(value: Any) -> None:
        if isinstance(value, dict):
            enum = value.get("enum")
            if isinstance(enum, list):
                enums.extend(item for item in enum if isinstance(item, str))
            for child in value.values():
                collect_enums(child)
        elif isinstance(value, list):
            for child in value:
                collect_enums(child)

    collect_enums(stream_mode_schema)
    collect_enums(schemas["RunStreamMode"])

    assert set(enums) == SUPPORTED_STREAM_MODES
    assert "events" not in enums


def test_openapi_run_option_schema_exposes_only_supported_values() -> None:
    app = FastAPI()
    app.include_router(runs.router)
    schema = app.openapi()["components"]["schemas"]["RunCreateRequest"]
    properties = schema["properties"]

    assert schema["additionalProperties"] is False
    for field in ("webhook", "after_seconds", "feedback_keys"):
        assert properties[field]["type"] == "null"
    assert properties["stream_resumable"]["anyOf"] == [{"const": False, "type": "boolean"}, {"type": "null"}]
    assert properties["on_completion"]["type"] == "null"
    assert properties["if_not_exists"]["const"] == "create"
    assert properties["multitask_strategy"]["enum"] == ["reject", "rollback", "interrupt"]