From ed336ec3dd3e5e76d6fc76f442209c0e3ebdd5c4 Mon Sep 17 00:00:00 2001 From: rayhpeng Date: Thu, 27 Aug 2026 14:59:31 +0800 Subject: [PATCH] fix(gateway): enforce run-create authorization on stateless endpoints (#5030) * fix(gateway): enforce authz on stateless runs * fix(gateway): guard scheduled run creation --- README.md | 2 +- backend/app/gateway/AGENTS.md | 2 +- backend/app/gateway/routers/runs.py | 2 + .../app/gateway/routers/scheduled_tasks.py | 4 + .../test_authorization_route_permissions.py | 106 +++++++++++++++++- .../tests/test_run_creation_route_contract.py | 66 +++++++++++ .../test_scheduled_task_router_behavior.py | 55 ++++++--- .../tests/test_thread_regenerate_prepare.py | 13 ++- 8 files changed, 222 insertions(+), 28 deletions(-) create mode 100644 backend/tests/test_run_creation_route_contract.py diff --git a/README.md b/README.md index a3c285558..fabd18416 100644 --- a/README.md +++ b/README.md @@ -905,7 +905,7 @@ uv run python -m deerflow.skills.review.cli ../skills/public/data-analysis --for Tools follow the same philosophy. DeerFlow comes with a core toolset — web search, web fetch, rendered web capture, file operations, bash execution — and supports custom tools via MCP servers and Python functions. Swap anything. Add anything. -Advanced deployments can enable pluggable authorization with `authorization.enabled` in `config.yaml`. A configured `AuthorizationProvider` filters denied tools before they reach the model or deferred-tool catalog, then the same provider is checked again before every business-tool execution through the existing guardrail middleware. Gateway `threads:*` and `runs:*` route permissions are derived from the same provider, while existing owner checks and admin-only management gates remain in force. A generated `tool_search` may bypass the second tool check only when it fronts the current build's already-filtered deferred catalog. Model access follows the same provider: the Gateway `models` list is filtered per principal, `model:use` is enforced on model detail requests and again when the runtime resolves the agent's model, and a denied default model falls back to the first remaining candidate that also passes `model:use`. The built-in RBAC provider supports per-role `tools`, `routes`, `models`, `skills`, and `sandbox` allow/deny policies and validates that `default_role` names a configured role; authorization is disabled by default. See `config.example.yaml` and the [authorization RFC](docs/plans/2026-07-10-pluggable-authorization-rfc.md). +Advanced deployments can enable pluggable authorization with `authorization.enabled` in `config.yaml`. A configured `AuthorizationProvider` filters denied tools before they reach the model or deferred-tool catalog, then the same provider is checked again before every business-tool execution through the existing guardrail middleware. Gateway `threads:*` and `runs:*` route permissions are derived from the same provider, while existing owner checks and admin-only management gates remain in force. Every HTTP route that starts or enables a future Agent run requires `runs:create`: this includes the stateless `POST /api/runs/stream` and `POST /api/runs/wait` endpoints plus scheduled-task create, update, resume, and manual-trigger mutations. Scheduled-task mutations retain their existing `threads:write` requirement, and the stateless routes separately enforce ownership when the optional thread ID is supplied in the request body. A generated `tool_search` may bypass the second tool check only when it fronts the current build's already-filtered deferred catalog. Model access follows the same provider: the Gateway `models` list is filtered per principal, `model:use` is enforced on model detail requests and again when the runtime resolves the agent's model, and a denied default model falls back to the first remaining candidate that also passes `model:use`. The built-in RBAC provider supports per-role `tools`, `routes`, `models`, `skills`, and `sandbox` allow/deny policies and validates that `default_role` names a configured role; authorization is disabled by default. See `config.example.yaml` and the [authorization RFC](docs/plans/2026-07-10-pluggable-authorization-rfc.md). Advanced deployments can also extend the agent runtime itself by declaring zero-argument `AgentMiddleware` classes under `extensions.middlewares` in `config.yaml` or `extensions_config.json`. DeerFlow loads the same configured class list into the lead-agent and subagent pipelines after their built-in runtime middlewares and loop/token guards, but before the terminal-response/safety/clarification tail, so enterprise forks can add domain guardrails, tool-call governance, or observability hooks without patching the built-in middleware builders. Missing packages, invalid classes, and broken modules fail loudly at agent creation. Treat `config.yaml` and `extensions_config.json` as trusted operator-controlled files: middleware paths are code execution, just like custom tool, model, sandbox, guardrail, MCP server, and MCP interceptor declarations. Gateway skill/MCP toggle endpoints preserve this field but do not expose an API write path for `extensions.middlewares`. Per-context parameterization and separate lead-only/subagent-only middleware lists are not supported yet. diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 9ccaf293c..0fed15652 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -59,7 +59,7 @@ reads/searches. | **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) | | **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block. Before the first journaled run, an empty run-event message feed is seeded from an existing checkpoint head so legacy checkpoint-only history receives earlier thread-global sequence numbers and remains visible after the new run; a thread with no checkpoint or an already-populated feed skips this compatibility path. `POST /regenerate/prepare` - prepare clean input + checkpoint metadata for regenerating the latest completed or interrupted assistant answer, carrying the latest non-empty thread title in graph input so resuming an older checkpoint cannot roll back a later manual rename (#4457); `POST /edit-regenerate/prepare` - prepare a checkpoint replay from the latest editable human turn with a replacement user message and edit replay metadata; it carries the current thread title the same way, but only when the replay base already has one — an untitled base belongs to a thread the title middleware has not named yet, so pinning the current title there would keep a name generated from the prompt the edit just replaced; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/messages` - paginated per-run messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /{rid}/workspace-changes` - workspace/output file change summary and optional diffs; `GET /../messages` - legacy thread message array; `GET /../messages/page` - backward thread-global `seq` history page with middleware/subagent-AI/successful-regenerate/edit-replay filtering and page-run-scoped feedback enrichment; subagent AI callbacks remain available through run events while parent `task` ToolMessages stay visible for card restoration; `GET /../token-usage` - aggregate tokens plus an optional `context_usage` percentage. Context usage approximately counts messages from the latest materialized thread state through `build_thread_checkpoint_state_accessor`, so full and delta checkpoint modes expose the same input. The percentage uses the latest run's model and its configured `context_window`. | | **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific | -| **Runs** (`/api/runs`) | `POST /stream` - stateless run + SSE; `POST /wait` - stateless run + block; `GET /{rid}/messages` - paginated messages by run_id `{data, has_more}` (cursor: `after_seq`/`before_seq`); `GET /{rid}/feedback` - list feedback by run_id | +| **Runs** (`/api/runs`) | `POST /stream`, `/wait` - stateless runs requiring `runs:create`; optional body `thread_id` is owner-checked. Scheduled-task create/update/resume/trigger also require `threads:write` plus `runs:create`. `GET /{rid}/messages`, `/feedback` - run messages/feedback | | **GitHub Webhooks** (`/api/webhooks/github`) | `POST /` - receive GitHub App / repo webhook deliveries. Verifies `X-Hub-Signature-256` against `GITHUB_WEBHOOK_SECRET`; exempt from auth + CSRF because authenticity is enforced by HMAC. The route is fail-closed: mounted only when `GITHUB_WEBHOOK_SECRET` is set, or when explicit dev opt-in `DEER_FLOW_ALLOW_UNVERIFIED_GITHUB_WEBHOOKS=1` is set. Recognized events include `ping`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, and `pull_request_review_comment`; unknown events return 200 with `handled=false`. Fan-out runtime failures return 503, keeping the delivery recorded as failed for manual/API/scripted redelivery (GitHub does not automatically retry any failed delivery, 5xx included); permanent/non-retryable conditions such as `channels.github.enabled: false`, unknown events, malformed payloads, or unavailable channel service return 200 with a skipped/handled response. | | **GitHub Event-Driven Agents** | Custom agents can declare a `github:` block in their `config.yaml` to bind to repos and event triggers. Webhook fan-out publishes one `InboundMessage` per matching binding to the channel bus; `GitHubChannel` routes those messages through `ChannelManager`. The response `dispatch` summarizes matched/fired/skipped agents. | diff --git a/backend/app/gateway/routers/runs.py b/backend/app/gateway/routers/runs.py index dc80a60f1..846913b2f 100644 --- a/backend/app/gateway/routers/runs.py +++ b/backend/app/gateway/routers/runs.py @@ -31,6 +31,7 @@ def _resolve_thread_id(body: RunCreateRequest) -> str: @router.post("/stream") +@require_permission("runs", "create") async def stateless_stream(body: RunCreateRequest, request: Request) -> StreamingResponse: """Create a run and stream events via SSE. @@ -56,6 +57,7 @@ async def stateless_stream(body: RunCreateRequest, request: Request) -> Streamin @router.post("/wait", response_model=dict) +@require_permission("runs", "create") async def stateless_wait(body: RunCreateRequest, request: Request) -> dict: """Create a run and block until completion. diff --git a/backend/app/gateway/routers/scheduled_tasks.py b/backend/app/gateway/routers/scheduled_tasks.py index 1e1ea7a16..496155763 100644 --- a/backend/app/gateway/routers/scheduled_tasks.py +++ b/backend/app/gateway/routers/scheduled_tasks.py @@ -81,6 +81,7 @@ async def list_scheduled_tasks(request: Request): @router.post("/scheduled-tasks") @require_permission("threads", "write") +@require_permission("runs", "create") async def create_scheduled_task(request: Request, body: ScheduledTaskCreateRequest): config = get_config() repo = get_scheduled_task_repo(request) @@ -153,6 +154,7 @@ async def get_scheduled_task(task_id: str, request: Request): @router.patch("/scheduled-tasks/{task_id}") @require_permission("threads", "write") +@require_permission("runs", "create") async def update_scheduled_task(task_id: str, request: Request, body: ScheduledTaskUpdateRequest): config = get_config() repo = get_scheduled_task_repo(request) @@ -272,6 +274,7 @@ async def pause_scheduled_task(task_id: str, request: Request): @router.post("/scheduled-tasks/{task_id}/resume") @require_permission("threads", "write") +@require_permission("runs", "create") async def resume_scheduled_task(task_id: str, request: Request): repo = get_scheduled_task_repo(request) user = await get_optional_user_from_request(request) @@ -300,6 +303,7 @@ async def resume_scheduled_task(task_id: str, request: Request): @router.post("/scheduled-tasks/{task_id}/trigger") @require_permission("threads", "write") +@require_permission("runs", "create") async def trigger_scheduled_task(task_id: str, request: Request): repo = get_scheduled_task_repo(request) service = get_scheduled_task_service(request) diff --git a/backend/tests/test_authorization_route_permissions.py b/backend/tests/test_authorization_route_permissions.py index 77ba2b7d8..8318fd0db 100644 --- a/backend/tests/test_authorization_route_permissions.py +++ b/backend/tests/test_authorization_route_permissions.py @@ -1,10 +1,10 @@ """Route-level authorization tests for the Gateway permission decorators.""" from types import SimpleNamespace -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, MagicMock import pytest -from fastapi import FastAPI, Request +from fastapi import FastAPI, HTTPException, Request from fastapi.testclient import TestClient from app.gateway.auth.models import User @@ -16,6 +16,7 @@ from app.gateway.authz import ( require_permission, resolve_route_permissions, ) +from app.gateway.routers import runs, scheduled_tasks from deerflow.authz.provider import AuthzDecision, AuthzReason from deerflow.authz.rbac import RbacAuthorizationProvider from deerflow.config.authorization_config import AuthorizationConfig, AuthorizationProviderConfig @@ -268,6 +269,107 @@ def test_auth_middleware_marks_internal_route_principal(monkeypatch): assert permission_resolver.await_args.kwargs == {"is_internal": True} +_STATELESS_RUN_PATHS = ("/api/runs/stream", "/api/runs/wait") + + +def _enable_auth_disabled_for_route_test(monkeypatch) -> None: + monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1") + monkeypatch.delenv("DEER_FLOW_ENV", raising=False) + monkeypatch.delenv("ENVIRONMENT", raising=False) + + +def _make_stateless_runs_app() -> FastAPI: + app = FastAPI() + app.add_middleware(AuthMiddleware) + app.include_router(runs.router) + app.state.stream_bridge = MagicMock() + app.state.run_manager = MagicMock() + return app + + +@pytest.mark.parametrize("path", _STATELESS_RUN_PATHS) +def test_stateless_run_creation_requires_runs_create(monkeypatch, path): + _enable_auth_disabled_for_route_test(monkeypatch) + monkeypatch.setattr( + "app.gateway.auth_middleware.resolve_route_permissions", + AsyncMock(return_value=[Permissions.RUNS_READ]), + ) + start_run = AsyncMock(side_effect=HTTPException(status_code=418, detail="run creation reached")) + monkeypatch.setattr(runs, "start_run", start_run) + + with TestClient(_make_stateless_runs_app()) as client: + response = client.post(path, json={}) + + assert response.status_code == 403 + assert response.json() == {"detail": "Permission denied: runs:create"} + start_run.assert_not_awaited() + + +@pytest.mark.parametrize("path", _STATELESS_RUN_PATHS) +def test_stateless_run_creation_allows_runs_create(monkeypatch, path): + _enable_auth_disabled_for_route_test(monkeypatch) + monkeypatch.setattr( + "app.gateway.auth_middleware.resolve_route_permissions", + AsyncMock(return_value=[Permissions.RUNS_CREATE]), + ) + start_run = AsyncMock(side_effect=HTTPException(status_code=418, detail="run creation reached")) + monkeypatch.setattr(runs, "start_run", start_run) + + with TestClient(_make_stateless_runs_app()) as client: + response = client.post(path, json={}) + + assert response.status_code == 418 + assert response.json() == {"detail": "run creation reached"} + start_run.assert_awaited_once() + + +_SCHEDULED_RUN_CREATION_REQUESTS = ( + ( + "POST", + "/api/scheduled-tasks", + { + "title": "Daily summary", + "prompt": "Summarize the latest activity", + "schedule_type": "cron", + "schedule_spec": {"cron": "0 9 * * *"}, + "timezone": "UTC", + }, + ), + ("PATCH", "/api/scheduled-tasks/task-1", {"title": "Updated summary"}), + ("POST", "/api/scheduled-tasks/task-1/resume", None), + ("POST", "/api/scheduled-tasks/task-1/trigger", None), +) + + +def _make_scheduled_tasks_app() -> FastAPI: + app = FastAPI() + app.add_middleware(AuthMiddleware) + app.include_router(scheduled_tasks.router) + return app + + +@pytest.mark.parametrize(("method", "path", "payload"), _SCHEDULED_RUN_CREATION_REQUESTS) +@pytest.mark.parametrize( + ("permissions", "denied_permission"), + [ + ([Permissions.THREADS_WRITE], Permissions.RUNS_CREATE), + ([Permissions.RUNS_CREATE], Permissions.THREADS_WRITE), + ], +) +def test_scheduled_run_creation_requires_thread_write_and_runs_create(monkeypatch, method, path, payload, permissions, denied_permission): + _enable_auth_disabled_for_route_test(monkeypatch) + monkeypatch.setattr( + "app.gateway.auth_middleware.resolve_route_permissions", + AsyncMock(return_value=permissions), + ) + + with TestClient(_make_scheduled_tasks_app()) as client: + response = client.request(method, path, json=payload) + + assert response.status_code == 403 + assert response.json() == {"detail": f"Permission denied: {denied_permission}"} + + # ── Provider cache tests ──────────────────────────────────────────────── diff --git a/backend/tests/test_run_creation_route_contract.py b/backend/tests/test_run_creation_route_contract.py new file mode 100644 index 000000000..6ca1d3ae4 --- /dev/null +++ b/backend/tests/test_run_creation_route_contract.py @@ -0,0 +1,66 @@ +"""Static authorization contract for HTTP routes that can create Agent runs.""" + +from __future__ import annotations + +import ast +import re +from pathlib import Path + +ROUTERS_DIR = Path(__file__).resolve().parent.parent / "app" / "gateway" / "routers" + +# These scheduled-task mutations create or re-enable work that the background +# scheduler later launches through the normal Gateway run lifecycle. +SCHEDULED_RUN_ENABLING_HANDLERS = { + "create_scheduled_task", + "update_scheduled_task", + "resume_scheduled_task", +} + +_ROUTE_DECORATOR_RE = re.compile(r"router\.(get|post|delete|put|patch)") + + +def _is_route_handler(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool: + return any( + isinstance(decorator, ast.Call) and isinstance(decorator.func, ast.Attribute) and isinstance(decorator.func.value, ast.Name) and _ROUTE_DECORATOR_RE.fullmatch(f"{decorator.func.value.id}.{decorator.func.attr}") + for decorator in node.decorator_list + ) + + +def _calls_run_launcher(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool: + for child in ast.walk(node): + if not isinstance(child, ast.Call): + continue + if isinstance(child.func, ast.Name) and child.func.id == "start_run": + return True + if isinstance(child.func, ast.Attribute) and child.func.attr == "dispatch_task": + return True + return False + + +def _requires_runs_create(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool: + for decorator in node.decorator_list: + if not isinstance(decorator, ast.Call): + continue + if not isinstance(decorator.func, ast.Name) or decorator.func.id != "require_permission": + continue + if len(decorator.args) < 2: + continue + resource, action = decorator.args[:2] + if isinstance(resource, ast.Constant) and resource.value == "runs" and isinstance(action, ast.Constant) and action.value == "create": + return True + return False + + +def test_every_run_creation_route_requires_runs_create(): + violations = [] + for path in sorted(ROUTERS_DIR.glob("*.py")): + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) or not _is_route_handler(node): + continue + if node.name not in SCHEDULED_RUN_ENABLING_HANDLERS and not _calls_run_launcher(node): + continue + if not _requires_runs_create(node): + violations.append(f"{path.name}:{node.name}") + + assert not violations, "run-creating routes without runs:create:\n" + "\n".join(violations) diff --git a/backend/tests/test_scheduled_task_router_behavior.py b/backend/tests/test_scheduled_task_router_behavior.py index 18690a5ef..ba7c45c03 100644 --- a/backend/tests/test_scheduled_task_router_behavior.py +++ b/backend/tests/test_scheduled_task_router_behavior.py @@ -4,6 +4,7 @@ from types import SimpleNamespace from unittest.mock import AsyncMock import pytest +from _router_auth_helpers import call_unwrapped from fastapi import HTTPException from app.gateway.routers import scheduled_tasks @@ -188,7 +189,8 @@ async def test_create_scheduled_task_uses_repo(): scheduled_tasks.get_config = lambda: config scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) - created = await scheduled_tasks.create_scheduled_task.__wrapped__( + created = await call_unwrapped( + scheduled_tasks.create_scheduled_task, request=request, body=body, ) @@ -231,7 +233,8 @@ async def test_create_fresh_thread_task_does_not_require_thread_id(): scheduled_tasks.get_config = lambda: config scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) - created = await scheduled_tasks.create_scheduled_task.__wrapped__( + created = await call_unwrapped( + scheduled_tasks.create_scheduled_task, request=request, body=body, ) @@ -273,7 +276,8 @@ async def test_trigger_scheduled_task_dispatches_manual_run(): scheduled_tasks.get_scheduled_task_service = lambda _request: service scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) - result = await scheduled_tasks.trigger_scheduled_task.__wrapped__( + result = await call_unwrapped( + scheduled_tasks.trigger_scheduled_task, task_id=task["id"], request=request, ) @@ -317,7 +321,8 @@ async def test_trigger_scheduled_task_returns_conflict_when_dispatch_conflicts() scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) with pytest.raises(Exception) as exc_info: - await scheduled_tasks.trigger_scheduled_task.__wrapped__( + await call_unwrapped( + scheduled_tasks.trigger_scheduled_task, task_id=task["id"], request=request, ) @@ -360,7 +365,8 @@ async def test_update_scheduled_task_writes_repo(): scheduled_tasks.get_config = lambda: config scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) - result = await scheduled_tasks.update_scheduled_task.__wrapped__( + result = await call_unwrapped( + scheduled_tasks.update_scheduled_task, task_id=task["id"], request=request, body=scheduled_tasks.ScheduledTaskUpdateRequest(title="Updated title"), @@ -419,7 +425,8 @@ async def test_update_rechecks_atomic_mutability_after_router_precheck(tmp_path) scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=SimpleNamespace(id="user-1")) patch_call = asyncio.create_task( - scheduled_tasks.update_scheduled_task.__wrapped__( + call_unwrapped( + scheduled_tasks.update_scheduled_task, task_id=task["id"], request=SimpleNamespace(), body=scheduled_tasks.ScheduledTaskUpdateRequest(prompt="changed after admission"), @@ -476,7 +483,8 @@ async def test_delete_scheduled_task_deletes_repo_row(): scheduled_tasks.get_scheduled_task_repo = lambda _request: repo scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) - result = await scheduled_tasks.delete_scheduled_task.__wrapped__( + result = await call_unwrapped( + scheduled_tasks.delete_scheduled_task, task_id=task["id"], request=request, ) @@ -513,12 +521,14 @@ async def test_pause_and_resume_scheduled_task_update_status(): scheduled_tasks.get_scheduled_task_repo = lambda _request: repo scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) - paused = await scheduled_tasks.pause_scheduled_task.__wrapped__( + paused = await call_unwrapped( + scheduled_tasks.pause_scheduled_task, task_id=task["id"], request=request, ) paused_status = paused["status"] - resumed = await scheduled_tasks.resume_scheduled_task.__wrapped__( + resumed = await call_unwrapped( + scheduled_tasks.resume_scheduled_task, task_id=task["id"], request=request, ) @@ -555,7 +565,8 @@ async def test_pause_cancels_waiting_occurrence_before_pausing_task(): try: scheduled_tasks.get_scheduled_task_repo = lambda _request: repo scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) - result = await scheduled_tasks.pause_scheduled_task.__wrapped__( + result = await call_unwrapped( + scheduled_tasks.pause_scheduled_task, task_id=task["id"], request=request, ) @@ -593,7 +604,8 @@ async def test_delete_rejects_occurrence_that_has_started_launching(): scheduled_tasks.get_scheduled_task_repo = lambda _request: repo scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) with pytest.raises(Exception) as exc_info: - await scheduled_tasks.delete_scheduled_task.__wrapped__( + await call_unwrapped( + scheduled_tasks.delete_scheduled_task, task_id=task["id"], request=request, ) @@ -632,7 +644,8 @@ async def test_pause_rejects_running_task(): scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) with pytest.raises(Exception) as exc_info: - await scheduled_tasks.pause_scheduled_task.__wrapped__( + await call_unwrapped( + scheduled_tasks.pause_scheduled_task, task_id=task["id"], request=request, ) @@ -676,7 +689,8 @@ async def test_update_rejects_running_task(): scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) with pytest.raises(Exception) as exc_info: - await scheduled_tasks.update_scheduled_task.__wrapped__( + await call_unwrapped( + scheduled_tasks.update_scheduled_task, task_id=task["id"], request=request, body=scheduled_tasks.ScheduledTaskUpdateRequest(title="Updated title"), @@ -720,7 +734,8 @@ async def test_update_rejects_queued_task_definition_until_occurrence_finishes() scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) with pytest.raises(Exception) as exc_info: - await scheduled_tasks.update_scheduled_task.__wrapped__( + await call_unwrapped( + scheduled_tasks.update_scheduled_task, task_id=task["id"], request=request, body=scheduled_tasks.ScheduledTaskUpdateRequest(prompt="Changed while queued"), @@ -773,7 +788,8 @@ async def test_list_thread_scheduled_tasks_filters_by_thread_id(): scheduled_tasks.get_scheduled_task_repo = lambda _request: repo scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) - result = await scheduled_tasks.list_thread_scheduled_tasks.__wrapped__( + result = await call_unwrapped( + scheduled_tasks.list_thread_scheduled_tasks, thread_id="thread-1", request=request, ) @@ -825,7 +841,8 @@ async def test_list_scheduled_task_runs_returns_persisted_rows_without_side_effe scheduled_tasks.get_scheduled_task_run_repo = lambda _request: run_repo scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) - result = await scheduled_tasks.list_scheduled_task_runs.__wrapped__( + result = await call_unwrapped( + scheduled_tasks.list_scheduled_task_runs, task_id=task["id"], request=request, ) @@ -864,7 +881,8 @@ async def test_create_once_task_enforces_minimum_delay(): scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) with pytest.raises(Exception) as exc_info: - await scheduled_tasks.create_scheduled_task.__wrapped__( + await call_unwrapped( + scheduled_tasks.create_scheduled_task, request=request, body=body, ) @@ -909,7 +927,8 @@ async def test_update_terminal_once_task_with_future_run_at_rearms_it(): scheduled_tasks.get_config = lambda: _Config() scheduled_tasks.get_optional_user_from_request = AsyncMock(return_value=user) - result = await scheduled_tasks.update_scheduled_task.__wrapped__( + result = await call_unwrapped( + scheduled_tasks.update_scheduled_task, task_id=task["id"], request=request, body=scheduled_tasks.ScheduledTaskUpdateRequest(schedule_spec={"run_at": future_run_at}), diff --git a/backend/tests/test_thread_regenerate_prepare.py b/backend/tests/test_thread_regenerate_prepare.py index 589207121..c99bf63f6 100644 --- a/backend/tests/test_thread_regenerate_prepare.py +++ b/backend/tests/test_thread_regenerate_prepare.py @@ -6,6 +6,7 @@ from types import SimpleNamespace from unittest.mock import AsyncMock, patch import pytest +from _router_auth_helpers import call_unwrapped from fastapi import HTTPException from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage from langgraph.checkpoint.base import empty_checkpoint, uuid6 @@ -251,8 +252,8 @@ def test_run_wait_readers_return_materialized_final_values() -> None: return_value=(accessor, snapshot.config), ), ): - thread_result = await thread_runs.wait_run.__wrapped__("thread-1", body, request) - stateless_result = await runs.stateless_wait(body, request) + thread_result = await call_unwrapped(thread_runs.wait_run, "thread-1", body, request) + stateless_result = await call_unwrapped(runs.stateless_wait, body, request) return thread_result, stateless_result thread_result, stateless_result = asyncio.run(_scenario()) @@ -303,8 +304,8 @@ def test_run_wait_readers_preserve_terminal_error_without_checkpoint() -> None: return_value=(accessor, snapshot.config), ), ): - thread_result = await thread_runs.wait_run.__wrapped__("thread-1", body, request) - stateless_result = await runs.stateless_wait(body, request) + thread_result = await call_unwrapped(thread_runs.wait_run, "thread-1", body, request) + stateless_result = await call_unwrapped(runs.stateless_wait, body, request) return thread_result, stateless_result thread_result, stateless_result = asyncio.run(_scenario()) @@ -340,7 +341,7 @@ def test_run_wait_readers_preserve_terminal_error_when_accessor_builder_fails(ro side_effect=RuntimeError("graph construction failed"), ), ): - return await thread_runs.wait_run.__wrapped__("thread-1", body, request) + return await call_unwrapped(thread_runs.wait_run, "thread-1", body, request) with ( patch.object(runs, "get_stream_bridge", return_value=object()), @@ -352,7 +353,7 @@ def test_run_wait_readers_preserve_terminal_error_when_accessor_builder_fails(ro side_effect=RuntimeError("graph construction failed"), ), ): - return await runs.stateless_wait(body, request) + return await call_unwrapped(runs.stateless_wait, body, request) result = asyncio.run(_scenario())