mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-26 17:56:27 +08:00
test+docs: pin precedence and leak fixes, restore POST OpenAPI schema
From the ultracode review: - The 422 stream-protocol precedence was untested (all cases used an existing flow). Pin it: a host whose get_flow 404s plus a bad protocol must still 422. - The UUID-leak test exercised the helper directly, not the host wiring that supplies requested_id. Route it through LangflowWorkflowHost.authorize so a regression in that wiring is caught. - The authenticated POST lost its responses= OpenAPI schema when the route moved to the shared router. Thread responses through create_workflow_router and restore WORKFLOW_EXECUTION_RESPONSES on the langflow mount. - Soften the router docstring: SSE framing is single-sourced only for hosts on the lfx default (langflow overrides stream_response).
This commit is contained in:
@ -1,5 +1,6 @@
|
||||
# Router for base api
|
||||
from fastapi import APIRouter
|
||||
from lfx.schema.workflow import WORKFLOW_EXECUTION_RESPONSES
|
||||
from lfx.services.settings.feature_flags import FEATURE_FLAGS
|
||||
from lfx.workflow.host import WorkflowHost
|
||||
from lfx.workflow.router import create_workflow_router
|
||||
@ -138,7 +139,12 @@ router_v2.include_router(registration_router_v2)
|
||||
_workflow_host = LangflowWorkflowHost()
|
||||
assert isinstance(_workflow_host, WorkflowHost) # noqa: S101
|
||||
router_v2.include_router(
|
||||
create_workflow_router(_workflow_host, developer_api_guard=False, auto_register_job_routes=False)
|
||||
create_workflow_router(
|
||||
_workflow_host,
|
||||
developer_api_guard=False,
|
||||
auto_register_job_routes=False,
|
||||
responses=WORKFLOW_EXECUTION_RESPONSES,
|
||||
)
|
||||
)
|
||||
# The langflow-owned durable routes: GET status, POST /stop, GET /{job_id}/events.
|
||||
router_v2.include_router(workflow_background_router_v2)
|
||||
|
||||
@ -340,20 +340,22 @@ class TestV2WorkflowAdmission:
|
||||
async def test_denial_echoes_requested_identifier_not_resolved_uuid(self, monkeypatch: pytest.MonkeyPatch):
|
||||
"""A denial on a flow referenced by endpoint name must not leak the resolved UUID."""
|
||||
from langflow.api.v2 import workflow as workflow_module
|
||||
from langflow.api.v2.workflow_host import LangflowWorkflowHost
|
||||
from lfx.workflow.actions import WorkflowAction
|
||||
from lfx.workflow.host import ResolvedFlow
|
||||
|
||||
resolved_uuid = uuid4()
|
||||
flow = SimpleNamespace(id=resolved_uuid, user_id=uuid4(), workspace_id=None, folder_id=None)
|
||||
resolved = ResolvedFlow(flow_id="my-endpoint", graph=flow)
|
||||
|
||||
async def _deny(*_args, **_kwargs):
|
||||
raise HTTPException(status_code=403, detail="denied")
|
||||
|
||||
monkeypatch.setattr(workflow_module, "ensure_flow_permission", _deny)
|
||||
|
||||
# Through the host, so the requested-id wiring is what's exercised, not just the helper.
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await workflow_module.authorize_flow_action(
|
||||
SimpleNamespace(id=uuid4()), flow, WorkflowAction.EXECUTE, requested_id="my-endpoint"
|
||||
)
|
||||
await LangflowWorkflowHost().authorize(SimpleNamespace(id=uuid4()), resolved, WorkflowAction.EXECUTE)
|
||||
|
||||
assert exc_info.value.status_code == 404
|
||||
assert exc_info.value.detail["code"] == "FLOW_NOT_FOUND"
|
||||
|
||||
@ -1,15 +1,17 @@
|
||||
"""The shared v2 workflow HTTP router.
|
||||
|
||||
lfx owns the entire env-neutral handler body: request parsing, stream-protocol
|
||||
validation, sync/stream dispatch, the single SSE framing loop, error -> HTTP
|
||||
mapping, and the ``developer_api_enabled`` router guard (which reads lfx's own
|
||||
settings service). The host (:mod:`lfx.workflow.host`) supplies only the
|
||||
DB/tenant-bound capabilities — caller resolution, fetch-and-authorize, the
|
||||
request session, and whether durable background runs exist.
|
||||
lfx owns the env-neutral handler body: request parsing, stream-protocol
|
||||
validation, sync/stream dispatch, error -> HTTP mapping, the lfx-default SSE
|
||||
framing loop, and the ``developer_api_enabled`` router guard (which reads lfx's
|
||||
own settings service). The host (:mod:`lfx.workflow.host`) supplies the
|
||||
DB/tenant-bound capabilities (caller resolution, fetch-and-authorize, the
|
||||
request session, durable background runs) and may override sync/stream
|
||||
execution; the langflow host does, keeping its richer SSE pipeline.
|
||||
|
||||
Both runtimes (langflow backend and bare ``lfx serve``) mount the same router
|
||||
via :func:`create_workflow_router`, so neither the wire contract nor the SSE
|
||||
shape can drift by construction.
|
||||
Both runtimes mount the same router via :func:`create_workflow_router`, so the
|
||||
request/response contract and the admission + error mapping are single-sourced.
|
||||
The SSE framing is single-sourced only for hosts using the lfx default (bare
|
||||
``lfx serve``); langflow overrides it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@ -250,6 +252,7 @@ def create_workflow_router(
|
||||
tags: tuple[str, ...] = ("Workflow",),
|
||||
developer_api_guard: bool = True,
|
||||
auto_register_job_routes: bool = True,
|
||||
responses: dict | None = None,
|
||||
) -> APIRouter:
|
||||
"""Build the v2 workflow router bound to ``host``.
|
||||
|
||||
@ -284,6 +287,7 @@ def create_workflow_router(
|
||||
@router.post(
|
||||
"",
|
||||
response_model=None,
|
||||
responses=responses or {},
|
||||
summary="Execute Workflow (v2 sync or stream)",
|
||||
)
|
||||
async def execute_workflow(request: WorkflowRunRequest, http_request: Request, background_tasks: BackgroundTasks):
|
||||
|
||||
@ -13,7 +13,7 @@ from typing import Any
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi import FastAPI, HTTPException, Request
|
||||
from fastapi.testclient import TestClient
|
||||
from lfx.components.input_output import ChatInput, ChatOutput
|
||||
from lfx.graph import Graph
|
||||
@ -113,7 +113,25 @@ def test_authorize_receives_workflow_action_member():
|
||||
assert action is WorkflowAction.EXECUTE
|
||||
|
||||
|
||||
def test_unknown_protocol_422_precedes_flow_fetch():
|
||||
"""An unknown stream_protocol must 422 before get_flow runs, even if the flow is missing."""
|
||||
|
||||
class _NoFlowHost(_FakeHost):
|
||||
async def get_flow(self, flow_id, caller): # noqa: ARG002
|
||||
raise HTTPException(status_code=404, detail="get_flow should not run for a bad protocol")
|
||||
|
||||
client = _client(_NoFlowHost(_echo_graph()))
|
||||
resp = client.post(
|
||||
"/workflows",
|
||||
json={"flow_id": _FLOW_ID, "input_value": "hi", "mode": "stream", "stream_protocol": "nope"},
|
||||
)
|
||||
assert resp.status_code == 422, resp.text
|
||||
assert resp.json()["detail"]["code"] == "UNKNOWN_STREAM_PROTOCOL"
|
||||
|
||||
|
||||
def test_sse_frame_shape_and_terminal_frame():
|
||||
# Pins the lfx-default framing (id line first), which bare serve uses. The langflow
|
||||
# host overrides stream_response, so this does not assert cross-runtime SSE identity.
|
||||
host = _FakeHost(_echo_graph())
|
||||
client = _client(host)
|
||||
resp = client.post(
|
||||
|
||||
Reference in New Issue
Block a user