From b37c0ebf4a58d4c0683cff2e3910322c17df6b6c Mon Sep 17 00:00:00 2001 From: Aryan Mathur <126969847+neverthesameagain@users.noreply.github.com> Date: Wed, 4 Feb 2026 20:50:17 +0530 Subject: [PATCH] added dependency graph --- .claude/settings.local.json | 17 +- fastapi/applications.py | 35 ++ fastapi/background.py | 20 +- fastapi/debug.py | 96 ++++ fastapi/dependencies/utils.py | 58 +- fastapi/routing.py | 75 ++- tests/test_background_tasks_bugs.py | 20 +- tests/test_background_tasks_fixes.py | 45 +- tests/test_cleanup_order.py | 800 +++++++++++++++++++++++++++ tests/test_dependency_debug.py | 341 ++++++++++++ 10 files changed, 1443 insertions(+), 64 deletions(-) create mode 100644 fastapi/debug.py create mode 100644 tests/test_cleanup_order.py create mode 100644 tests/test_dependency_debug.py diff --git a/.claude/settings.local.json b/.claude/settings.local.json index 58bfde171..e878467f1 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -5,17 +5,18 @@ }, "permissions": { "allow": [ - "Bash(ls:*)", - "Bash(grep:*)", - "Bash(python -m pytest:*)", - "Bash(pip install:*)", + "Bash(wc:*)", "Bash(python3:*)", - "Bash(python:*)", - "WebSearch", - "WebFetch(domain:github.com)" + "Bash(sort:*)", + "Bash(python -m pytest:*)", + "Bash(uv run pytest:*)", + "Bash(pip3 show:*)", + "Bash(/opt/homebrew/bin/python3.11:*)", + "Bash(python3.11:*)", + "Bash(ls:*)" ] }, - "model": "zorilla", + "model": "narwhal", "hooks": { "SessionStart": [ { diff --git a/fastapi/applications.py b/fastapi/applications.py index 329baab69..3e553fc4e 100644 --- a/fastapi/applications.py +++ b/fastapi/applications.py @@ -456,6 +456,30 @@ class FastAPI(Starlette): """ ), ] = "/redoc", + dependency_debug_url: Annotated[ + Optional[str], + Doc( + """ + The path for the internal dependency-graph debug endpoint. + + When set to a URL path string (e.g. ``"/_debug/dependencies"``), + FastAPI registers a GET endpoint at that path that returns the + full static dependency DAG for every route as JSON. The endpoint + performs **no** dependency execution — it is purely informational. + + Disabled by default (``None``). It should only be enabled in + development or internal-tooling environments. + + **Example** + + ```python + from fastapi import FastAPI + + app = FastAPI(dependency_debug_url="/_debug/dependencies") + ``` + """ + ), + ] = None, swagger_ui_oauth2_redirect_url: Annotated[ Optional[str], Doc( @@ -900,6 +924,7 @@ class FastAPI(Starlette): self.root_path_in_servers = root_path_in_servers self.docs_url = docs_url self.redoc_url = redoc_url + self.dependency_debug_url = dependency_debug_url self.swagger_ui_oauth2_redirect_url = swagger_ui_oauth2_redirect_url self.swagger_ui_init_oauth = swagger_ui_init_oauth self.swagger_ui_parameters = swagger_ui_parameters @@ -1219,6 +1244,16 @@ class FastAPI(Starlette): ) self.add_route(self.redoc_url, redoc_html, include_in_schema=False) + if self.dependency_debug_url: + # Lazy import to avoid circular imports at module load time. + from fastapi.debug import get_all_route_graphs + + async def dependency_debug(req: Request) -> JSONResponse: + return JSONResponse({"routes": get_all_route_graphs(self.routes)}) + + self.add_route( + self.dependency_debug_url, dependency_debug, include_in_schema=False + ) async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: if self.root_path: diff --git a/fastapi/background.py b/fastapi/background.py index 0ff6a90a6..513005952 100644 --- a/fastapi/background.py +++ b/fastapi/background.py @@ -1,5 +1,4 @@ import logging -import warnings from typing import Annotated, Any, Callable from annotated_doc import Doc @@ -81,13 +80,11 @@ class BackgroundTasks(StarletteBackgroundTasks): [FastAPI docs for Background Tasks](https://fastapi.tiangolo.com/tutorial/background-tasks/). """ if self._executed: - warnings.warn( + logger.warning( "Background task added after tasks have already been executed. " "This task will not run. This commonly happens when adding tasks " "after a 'yield' in a dependency. Consider adding tasks before " - "the yield, or use a different approach for cleanup tasks.", - UserWarning, - stacklevel=2, + "the yield, or use a different approach for cleanup tasks." ) return return super().add_task(func, *args, **kwargs) @@ -114,11 +111,11 @@ class BackgroundTasks(StarletteBackgroundTasks): - If only one task fails, the original exception is re-raised - If multiple tasks fail, a BackgroundTaskError is raised with all errors """ - # Fix #8: Snapshot tasks to prevent mutation during iteration - tasks_snapshot = list(self.tasks) - - # Fix #7: Set _executed after snapshot, so it reflects actual execution attempt + # Set _executed before the snapshot so that any concurrent add_task + # call hits the guard and logs a warning rather than silently + # appending to a list that will never be iterated. self._executed = True + tasks_snapshot = list(self.tasks) errors: list[tuple[BackgroundTask, BaseException]] = [] @@ -153,10 +150,7 @@ class BackgroundTasks(StarletteBackgroundTasks): # Handle errors with backward compatibility if len(errors) == 1: - # Fix #5: Single error - re-raise with proper context - # Using 'from' preserves the chain and adds context - original_exc = errors[0][1] - raise original_exc from original_exc + raise errors[0][1] elif len(errors) > 1: # Multiple errors: raise aggregate exception raise BackgroundTaskError(errors) diff --git a/fastapi/debug.py b/fastapi/debug.py new file mode 100644 index 000000000..97af29582 --- /dev/null +++ b/fastapi/debug.py @@ -0,0 +1,96 @@ +"""Static dependency-graph introspection for the debug endpoint. + +Nothing in this module executes any user callable. It walks the already-built +Dependant trees that live on each APIRoute and serialises them to plain dicts. +""" + +from collections.abc import Sequence +from typing import Any + +from fastapi.dependencies.models import Dependant, _impartial, _unwrapped_call + + +def _resolve_callable_name(call: Any) -> str: + """Return the display name of a callable, unwrapping partials and decorators.""" + if call is None: + return "" + unwrapped = _unwrapped_call(call) + if unwrapped is None: # pragma: no cover + return "" + # For classes the __name__ is the class name; for functions it is the + # function name. Both are what we want. + return getattr(unwrapped, "__name__", type(unwrapped).__name__) + + +def _resolve_callable_module(call: Any) -> str: + """Return the module path of a callable, unwrapping partials and decorators.""" + if call is None: + return "" + unwrapped = _unwrapped_call(call) + if unwrapped is None: # pragma: no cover + return "" + return getattr(unwrapped, "__module__", "") + + +def build_dependency_graph( + dependant: Dependant, + *, + _path_seen: frozenset[Any] | None = None, +) -> dict[str, Any]: + """Serialise one Dependant node (and its entire sub-tree) to a plain dict. + + Cycle / repeat detection operates on the *current root-to-node path* only. + The same callable appearing in two sibling branches is NOT flagged — only + an actual ancestor repeat on the same path triggers ``cached_repeat: true``. + """ + if _path_seen is None: + _path_seen = frozenset() + + cache_key = dependant.cache_key + is_repeat = cache_key in _path_seen + + node: dict[str, Any] = { + "callable_name": _resolve_callable_name(dependant.call), + "callable_module": _resolve_callable_module(dependant.call), + "scope": dependant.computed_scope, + "is_yield": dependant.is_gen_callable or dependant.is_async_gen_callable, + "is_async": dependant.is_async_gen_callable or dependant.is_coroutine_callable, + "security_scopes": list(dependant.oauth_scopes) if dependant.oauth_scopes else [], + "is_security_scheme": dependant._is_security_scheme, + } + + if is_repeat: + # Ancestor repeat on this path — stop recursion here. + node["cached_repeat"] = True + node["sub_dependencies"] = [] + else: + # Extend the path with the current node before recursing into children. + extended_path = _path_seen | {cache_key} + node["sub_dependencies"] = [ + build_dependency_graph(sub, _path_seen=extended_path) + for sub in dependant.dependencies + ] + + return node + + +def get_all_route_graphs(routes: Sequence[Any]) -> list[dict[str, Any]]: + """Build the full debug payload from a list of registered routes. + + Non-APIRoute entries (e.g. Mount, plain Starlette Route) are skipped. + """ + # Import here to avoid circular imports at module load time. + from fastapi.routing import APIRoute + + result: list[dict[str, Any]] = [] + for route in routes: + if not isinstance(route, APIRoute): + continue + entry: dict[str, Any] = { + "path": route.path_format, + "methods": sorted(route.methods) if route.methods else [], + "operation_id": route.unique_id, + "dependency_graph": build_dependency_graph(route.dependant), + } + result.append(entry) + return result diff --git a/fastapi/dependencies/utils.py b/fastapi/dependencies/utils.py index 45e1ff3ed..111270625 100644 --- a/fastapi/dependencies/utils.py +++ b/fastapi/dependencies/utils.py @@ -540,15 +540,60 @@ def add_param_to_fields(*, field: ModelField, dependant: Dependant) -> None: dependant.cookie_params.append(field) +def _record_cleanup_entry( + scope: dict[str, Any], dependant: Dependant, resolved_scope: str +) -> None: + """Append a push-order entry to the per-request cleanup-order metadata list. + + The list (``scope["fastapi_dependency_cleanup_order"]``) is created lazily + on first use. The function returns immediately when the debug endpoint is + not configured on the application, so there is zero allocation cost in the + default (disabled) case. + + This must be called *after* ``enter_async_context`` succeeds so that only + generators that were actually pushed onto a stack are recorded. + """ + # scope["app"] is stamped by Starlette.__call__ before routing begins. + # When dependency_debug_url is None (the default) nothing below executes. + if not getattr(scope.get("app"), "dependency_debug_url", None): + return + + from fastapi.dependencies.models import _unwrapped_call + + cleanup_order: list[dict[str, Any]] = scope.setdefault( + "fastapi_dependency_cleanup_order", [] + ) + call = dependant.call + unwrapped = _unwrapped_call(call) if call else call + name = getattr(unwrapped, "__name__", type(unwrapped).__name__) if unwrapped else "" + cleanup_order.append( + { + "callable_name": name, + "scope": resolved_scope, + "order": len(cleanup_order), + } + ) + + async def _solve_generator( - *, dependant: Dependant, stack: AsyncExitStack, sub_values: dict[str, Any] + *, + dependant: Dependant, + stack: AsyncExitStack, + sub_values: dict[str, Any], + scope: dict[str, Any], + resolved_scope: str, ) -> Any: assert dependant.call if dependant.is_async_gen_callable: cm = asynccontextmanager(dependant.call)(**sub_values) elif dependant.is_gen_callable: cm = contextmanager_in_threadpool(contextmanager(dependant.call)(**sub_values)) - return await stack.enter_async_context(cm) + # enter_async_context runs the generator up to its first yield and registers + # its finaliser on the stack. Record the entry *after* this succeeds so that + # generators that raise before yielding are never logged. + result = await stack.enter_async_context(cm) + _record_cleanup_entry(scope, dependant, resolved_scope) + return result @dataclass @@ -631,13 +676,22 @@ async def solve_dependencies( elif ( use_sub_dependant.is_gen_callable or use_sub_dependant.is_async_gen_callable ): + # Stack selection determines the cleanup phase: + # scope="function" → function_astack → finaliser runs BEFORE response + # anything else → request_astack → finaliser runs AFTER response + # Sub-dependencies of a dep are recursed (and pushed) *before* the + # dep itself, so in LIFO exit order the parent's finaliser runs first. use_astack = request_astack + resolved_scope = "request" if sub_dependant.scope == "function": use_astack = function_astack + resolved_scope = "function" solved = await _solve_generator( dependant=use_sub_dependant, stack=use_astack, sub_values=solved_result.values, + scope=request.scope, + resolved_scope=resolved_scope, ) elif use_sub_dependant.is_coroutine_callable: solved = await call(**solved_result.values) diff --git a/fastapi/routing.py b/fastapi/routing.py index dbd74fca9..b4494b8ab 100644 --- a/fastapi/routing.py +++ b/fastapi/routing.py @@ -3,6 +3,7 @@ import functools import inspect import json import warnings +import weakref from collections.abc import ( AsyncIterator, Awaitable, @@ -96,6 +97,22 @@ def request_response( async def app(scope: Scope, receive: Receive, send: Send) -> None: # Starts customization response_awaited = False + # ── Cleanup-order invariant (two nested AsyncExitStacks) ────────── + # request_stack (inner_astack) opens FIRST → closes LAST. + # function_stack opens SECOND (inside request_stack) → closes FIRST. + # + # Deterministic exit sequence per request: + # 1. function_stack.__aexit__ – all scope="function" yield-dep + # finalisers run here, in LIFO push order. + # 2. await response(...) – HTTP bytes are sent to the client; + # BackgroundTasks execute *inside* this call (see + # starlette/responses.py Response.__call__ / StreamingResponse.__call__). + # 3. request_stack.__aexit__ – all scope="request" (and default- + # scope generator) yield-dep finalisers run here, in LIFO push order. + # + # Consequence: function-scoped finalisers see the response body *before* + # it is streamed; request-scoped finalisers run *after* BackgroundTasks. + # ───────────────────────────────────────────────────────────────────── async with AsyncExitStack() as request_stack: scope["fastapi_inner_astack"] = request_stack async with AsyncExitStack() as function_stack: @@ -162,16 +179,31 @@ def _merge_lifespan_context( return merged_lifespan # type: ignore[return-value] -# Cache for endpoint context to avoid re-extracting on every request -_endpoint_context_cache: dict[int, EndpointContext] = {} +# Cache: id(func) → (weakref to func, EndpointContext). +# The weakref callback evicts the entry when func is GC'd; the identity +# check on hit guards against id() reuse before the callback fires. +_endpoint_context_cache: dict[int, tuple[weakref.ref[Any], EndpointContext]] = {} + + +def _endpoint_context_finaliser(func_id: int) -> Callable[[weakref.ref[Any]], None]: + """Return a weakref callback that evicts the cache entry for *func_id*.""" + + def _evict(_ref: weakref.ref[Any]) -> None: + _endpoint_context_cache.pop(func_id, None) + + return _evict def _extract_endpoint_context(func: Any) -> EndpointContext: """Extract endpoint context with caching to avoid repeated file I/O.""" func_id = id(func) - if func_id in _endpoint_context_cache: - return _endpoint_context_cache[func_id] + cached = _endpoint_context_cache.get(func_id) + if cached is not None: + ref, ctx = cached + if ref() is func: # same object — cache hit + return ctx + # id was reused after GC — fall through to recompute try: ctx: EndpointContext = {} @@ -185,7 +217,15 @@ def _extract_endpoint_context(func: Any) -> EndpointContext: except Exception: ctx = EndpointContext() - _endpoint_context_cache[func_id] = ctx + # Cache only if func supports weak references. functools.partial and + # some other callables do not; for those we simply skip caching. + try: + _endpoint_context_cache[func_id] = ( + weakref.ref(func, _endpoint_context_finaliser(func_id)), + ctx, + ) + except TypeError: + pass return ctx @@ -360,7 +400,18 @@ def get_request_handler( is_coroutine=is_coroutine, ) if isinstance(raw_response, Response): - if raw_response.background is None: + if solved_result.background_tasks is not None: + if ( + raw_response.background is not None + and raw_response.background is not solved_result.background_tasks + ): + # Endpoint set an explicit background task on the + # Response. Fold it into the injected BackgroundTasks + # so dep-added tasks and the endpoint's explicit task + # both execute under a single __call__. + solved_result.background_tasks.tasks.append( + raw_response.background + ) raw_response.background = solved_result.background_tasks response = raw_response else: @@ -623,6 +674,18 @@ class APIRoute(routing.Route): self.dependant = get_dependant( path=self.path_format, call=self.endpoint, scope="function" ) + # ── Route-level dependency prepend order ───────────────────────────── + # Route-level deps (from ``dependencies=[...]``) must appear BEFORE + # signature deps in dependant.dependencies so that solve_dependencies() + # pushes them onto the exit stack FIRST. Because AsyncExitStack is + # LIFO, "pushed first" means "cleaned up last" — which is the correct + # semantic: route-level (e.g. auth) finalisers should outlive the + # per-endpoint business-logic finalisers. + # + # The [::-1] reversal + insert(0, …) pattern achieves this: + # dependencies = [A, B] → iterate [B, A] → insert B at 0, then A + # at 0 → final order: [A, B, …signature deps…] + # ────────────────────────────────────────────────────────────────────── for depends in self.dependencies[::-1]: self.dependant.dependencies.insert( 0, diff --git a/tests/test_background_tasks_bugs.py b/tests/test_background_tasks_bugs.py index 60e88561d..390bffa43 100644 --- a/tests/test_background_tasks_bugs.py +++ b/tests/test_background_tasks_bugs.py @@ -8,7 +8,7 @@ Based on GitHub issues: """ import asyncio -import warnings +import logging from typing import Generator import pytest @@ -44,9 +44,9 @@ class TestBackgroundTasksAfterYield: assert response.status_code == 200 assert "before_yield" in executed - def test_bg_task_added_after_yield_warns_and_not_executed(self): + def test_bg_task_added_after_yield_warns_and_not_executed(self, caplog): """ - FIXED: Tasks added after yield now emit a warning instead of being silently dropped. + FIXED: Tasks added after yield now log a warning instead of being silently dropped. Previously, this was a silent failure. Now users get a clear warning explaining why the task won't run. @@ -60,7 +60,7 @@ class TestBackgroundTasksAfterYield: yield "value" # This code runs during cleanup executed.append("after_yield_code_runs") - # Now this emits a warning instead of silently failing + # Now this logs a warning instead of silently failing background_tasks.add_task(lambda: executed.append("after_yield_task")) @app.get("/test") @@ -70,9 +70,7 @@ class TestBackgroundTasksAfterYield: client = TestClient(app) executed.clear() - # Capture warnings during the request - with warnings.catch_warnings(record=True) as w: - warnings.simplefilter("always") + with caplog.at_level(logging.WARNING, logger="fastapi.background"): response = client.get("/test") assert response.status_code == 200 @@ -81,9 +79,11 @@ class TestBackgroundTasksAfterYield: assert "after_yield_code_runs" in executed # ...the background task does NOT execute (as before) assert "after_yield_task" not in executed - # BUT now we get a warning about it (FIX!) - assert len(w) == 1 - assert "Background task added after tasks have already been executed" in str(w[0].message) + # BUT now we get a log warning about it (FIX!) + assert any( + "Background task added after tasks have already been executed" in r.message + for r in caplog.records + ) class TestBackgroundTaskFailureCascade: diff --git a/tests/test_background_tasks_fixes.py b/tests/test_background_tasks_fixes.py index 605bea829..5cb5315d2 100644 --- a/tests/test_background_tasks_fixes.py +++ b/tests/test_background_tasks_fixes.py @@ -6,7 +6,7 @@ Fixed bugs: 2. Tasks added after execution now emit a warning """ -import warnings +import logging from typing import Generator import pytest @@ -145,24 +145,18 @@ class TestBackgroundTaskAfterYieldWarning: Test that adding tasks after execution emits a warning. """ - def test_warning_when_task_added_after_execution(self): + def test_warning_when_task_added_after_execution(self, caplog): """ - FIX: Warning should be emitted when task is added after execution. + FIX: A log warning should be emitted when task is added after execution. """ executed = [] - warnings_captured = [] app = FastAPI() async def dependency_with_yield(background_tasks: BackgroundTasks): background_tasks.add_task(lambda: executed.append("before_yield")) yield "value" - # Capture the warning - with warnings.catch_warnings(record=True) as w: - warnings.simplefilter("always") - background_tasks.add_task(lambda: executed.append("after_yield")) - if w: - warnings_captured.extend(w) + background_tasks.add_task(lambda: executed.append("after_yield")) @app.get("/test") async def endpoint(dep: str = Depends(dependency_with_yield)): @@ -170,9 +164,10 @@ class TestBackgroundTaskAfterYieldWarning: client = TestClient(app) executed.clear() - warnings_captured.clear() - response = client.get("/test") + with caplog.at_level(logging.WARNING, logger="fastapi.background"): + response = client.get("/test") + assert response.status_code == 200 # Task added before yield should have executed @@ -181,31 +176,31 @@ class TestBackgroundTaskAfterYieldWarning: # Task added after yield should NOT have executed assert "after_yield" not in executed - # Warning should have been captured - assert len(warnings_captured) == 1 - assert "Background task added after tasks have already been executed" in str( - warnings_captured[0].message + # Warning should have been logged + assert any( + "Background task added after tasks have already been executed" in r.message + for r in caplog.records ) - def test_no_warning_for_normal_task_addition(self): + def test_no_warning_for_normal_task_addition(self, caplog): """ - No warning should be emitted for normal task addition. + No warning should be logged for normal task addition. """ app = FastAPI() @app.get("/test") async def endpoint(background_tasks: BackgroundTasks): - with warnings.catch_warnings(record=True) as w: - warnings.simplefilter("always") - background_tasks.add_task(lambda: None) - background_tasks.add_task(lambda: None) - assert len(w) == 0 # No warnings - + background_tasks.add_task(lambda: None) + background_tasks.add_task(lambda: None) return {"status": "ok"} client = TestClient(app) - response = client.get("/test") + with caplog.at_level(logging.WARNING, logger="fastapi.background"): + response = client.get("/test") assert response.status_code == 200 + assert not any( + "already been executed" in r.message for r in caplog.records + ) class TestDependencyCleanupStillWorks: diff --git a/tests/test_cleanup_order.py b/tests/test_cleanup_order.py new file mode 100644 index 000000000..77d663b02 --- /dev/null +++ b/tests/test_cleanup_order.py @@ -0,0 +1,800 @@ +"""Regression tests for deterministic dependency-cleanup ordering (Task 2). + +Testing technique: every yield dep appends its own name to a closure-local list +*after* its yield (i.e. during cleanup). BackgroundTasks do the same. After +the HTTP round-trip completes the test inspects the list to assert ordering. + +The expected ordering is derived from the invariant documented in +``fastapi/routing.py`` (request_response) and +``fastapi/dependencies/utils.py`` (solve_dependencies): + + • Deps are pushed onto AsyncExitStack in dependant.dependencies iteration + order (route-level deps first, then signature deps left-to-right). + • Sub-deps of a dep are recursed *before* the parent is pushed. + • AsyncExitStack is LIFO → last-pushed finaliser runs first. + • function_stack exits BEFORE ``await response(…)``. + • BackgroundTasks run *inside* ``response(…)``. + • request_stack (inner_astack) exits AFTER ``response(…)``. +""" + +import json +import logging +from typing import Annotated + +from fastapi import BackgroundTasks, Depends, FastAPI, Request +from fastapi.responses import StreamingResponse +from fastapi.testclient import TestClient + + +# --------------------------------------------------------------------------- +# test_sibling_cleanup_is_lifo +# --------------------------------------------------------------------------- + + +def test_sibling_cleanup_is_lifo() -> None: + """Two yield deps declared as (dep_a, dep_b) in the signature. + dep_a is pushed first, dep_b second. LIFO → dep_b cleans up first.""" + cleanup_order: list[str] = [] + + def dep_a(): # type: ignore[no-untyped-def] + yield "a" + cleanup_order.append("dep_a") + + def dep_b(): # type: ignore[no-untyped-def] + yield "b" + cleanup_order.append("dep_b") + + app = FastAPI() + + @app.get("/") + def endpoint( + _a: Annotated[str, Depends(dep_a)], + _b: Annotated[str, Depends(dep_b)], + ) -> dict: + return {"ok": True} + + with TestClient(app) as client: + resp = client.get("/") + assert resp.status_code == 200 + assert cleanup_order == ["dep_b", "dep_a"] + + +# --------------------------------------------------------------------------- +# test_route_level_dep_cleanup_after_signature_deps +# --------------------------------------------------------------------------- + + +def test_route_level_dep_cleanup_after_signature_deps() -> None: + """Route-level dep is prepended (pushed first) → cleaned up last.""" + cleanup_order: list[str] = [] + + def route_dep(): # type: ignore[no-untyped-def] + yield + cleanup_order.append("route_dep") + + def sig_dep(): # type: ignore[no-untyped-def] + yield + cleanup_order.append("sig_dep") + + app = FastAPI() + + @app.get("/", dependencies=[Depends(route_dep)]) + def endpoint(_s: Annotated[None, Depends(sig_dep)]) -> dict: + return {"ok": True} + + with TestClient(app) as client: + resp = client.get("/") + assert resp.status_code == 200 + # sig_dep pushed after route_dep → exits first (LIFO) + assert cleanup_order == ["sig_dep", "route_dep"] + + +# --------------------------------------------------------------------------- +# test_nested_dep_cleanup_order +# --------------------------------------------------------------------------- + + +def test_nested_dep_cleanup_order() -> None: + """endpoint → dep_a (yield) → dep_b (yield). + dep_b is recursed first (pushed first). dep_a is pushed second. + LIFO: dep_a cleans up first, then dep_b.""" + cleanup_order: list[str] = [] + + def dep_b(): # type: ignore[no-untyped-def] + yield "b" + cleanup_order.append("dep_b") + + def dep_a(b: Annotated[str, Depends(dep_b)]): # type: ignore[no-untyped-def] + yield "a" + cleanup_order.append("dep_a") + + app = FastAPI() + + @app.get("/") + def endpoint(_a: Annotated[str, Depends(dep_a)]) -> dict: + return {"ok": True} + + with TestClient(app) as client: + resp = client.get("/") + assert resp.status_code == 200 + assert cleanup_order == ["dep_a", "dep_b"] + + +# --------------------------------------------------------------------------- +# test_async_and_sync_generators_share_lifo +# --------------------------------------------------------------------------- + + +def test_async_and_sync_generators_share_lifo() -> None: + """A sync yield dep and an async yield dep, in that signature order. + Both land on the same stack via enter_async_context; ordering is purely + push order → async one (pushed second) exits first.""" + cleanup_order: list[str] = [] + + def dep_sync(): # type: ignore[no-untyped-def] + yield "sync" + cleanup_order.append("dep_sync") + + async def dep_async(): # type: ignore[no-untyped-def] + yield "async" + cleanup_order.append("dep_async") + + app = FastAPI() + + @app.get("/") + def endpoint( + _s: Annotated[str, Depends(dep_sync)], + _a: Annotated[str, Depends(dep_async)], + ) -> dict: + return {"ok": True} + + with TestClient(app) as client: + resp = client.get("/") + assert resp.status_code == 200 + assert cleanup_order == ["dep_async", "dep_sync"] + + +# --------------------------------------------------------------------------- +# test_cleanup_runs_on_endpoint_exception +# --------------------------------------------------------------------------- + + +def test_cleanup_runs_on_endpoint_exception() -> None: + """All yield-dep finalisers still run even when the endpoint raises.""" + cleanup_order: list[str] = [] + + def dep_a(): # type: ignore[no-untyped-def] + try: + yield "a" + finally: + cleanup_order.append("dep_a") + + def dep_b(): # type: ignore[no-untyped-def] + try: + yield "b" + finally: + cleanup_order.append("dep_b") + + app = FastAPI() + + @app.get("/") + def endpoint( + _a: Annotated[str, Depends(dep_a)], + _b: Annotated[str, Depends(dep_b)], + ) -> dict: + raise RuntimeError("endpoint boom") + + # raise_server_exceptions=False so the 500 doesn't propagate into the test + with TestClient(app, raise_server_exceptions=False) as client: + resp = client.get("/") + assert resp.status_code == 500 + # Both deps cleaned up, in LIFO order + assert cleanup_order == ["dep_b", "dep_a"] + + +# --------------------------------------------------------------------------- +# test_function_scope_cleans_before_response_streaming +# --------------------------------------------------------------------------- + + +def test_function_scope_cleans_before_response_streaming() -> None: + """A function-scoped yield dep's finaliser runs *before* the response + body is streamed. The streaming iterator can observe the side-effect.""" + flag: dict[str, bool] = {"cleaned": False} + + def func_dep(): # type: ignore[no-untyped-def] + yield + flag["cleaned"] = True + + app = FastAPI() + + @app.get("/stream") + def stream_endpoint(_: Annotated[None, Depends(func_dep, scope="function")]) -> StreamingResponse: # type: ignore[no-untyped-def] + def body(): # type: ignore[no-untyped-def] + # By the time this runs, function_stack has already exited + yield json.dumps({"cleaned": flag["cleaned"]}) + + return StreamingResponse(body(), media_type="application/json") + + with TestClient(app) as client: + resp = client.get("/stream") + assert resp.status_code == 200 + assert resp.json()["cleaned"] is True + + +# --------------------------------------------------------------------------- +# test_request_scope_cleans_after_response_streaming +# --------------------------------------------------------------------------- + + +def test_request_scope_cleans_after_response_streaming() -> None: + """A request-scoped (default) yield dep's finaliser runs *after* the + response body is streamed. The streaming iterator sees the flag as False.""" + flag: dict[str, bool] = {"cleaned": False} + + def req_dep(): # type: ignore[no-untyped-def] + yield + flag["cleaned"] = True + + app = FastAPI() + + @app.get("/stream") + def stream_endpoint(_: Annotated[None, Depends(req_dep)]) -> StreamingResponse: # type: ignore[no-untyped-def] + def body(): # type: ignore[no-untyped-def] + # request_stack has NOT exited yet while the stream runs + yield json.dumps({"cleaned": flag["cleaned"]}) + + return StreamingResponse(body(), media_type="application/json") + + with TestClient(app) as client: + resp = client.get("/stream") + assert resp.status_code == 200 + assert resp.json()["cleaned"] is False + + +# --------------------------------------------------------------------------- +# test_background_task_runs_between_scopes +# --------------------------------------------------------------------------- + + +def test_background_task_runs_between_scopes() -> None: + """The three phases interleave in exactly this order: + 1. function-scoped finalisers (function_stack exits) + 2. BackgroundTasks (inside response.__call__) + 3. request-scoped finalisers (request_stack exits) + """ + cleanup_order: list[str] = [] + + def func_dep(): # type: ignore[no-untyped-def] + yield + cleanup_order.append("func_cleanup") + + def req_dep(): # type: ignore[no-untyped-def] + yield + cleanup_order.append("req_cleanup") + + app = FastAPI() + + @app.get("/") + def endpoint( + _f: Annotated[None, Depends(func_dep, scope="function")], + _r: Annotated[None, Depends(req_dep, scope="request")], + bg: BackgroundTasks, + ) -> dict: + bg.add_task(lambda: cleanup_order.append("bg_task")) + return {"ok": True} + + with TestClient(app) as client: + resp = client.get("/") + assert resp.status_code == 200 + assert cleanup_order == ["func_cleanup", "bg_task", "req_cleanup"] + + +# --------------------------------------------------------------------------- +# test_cleanup_order_metadata_populated +# --------------------------------------------------------------------------- + + +def test_cleanup_order_metadata_populated() -> None: + """scope["fastapi_dependency_cleanup_order"] is populated with one entry + per generator dep that was successfully pushed onto a stack.""" + captured: list[list[dict]] = [] # type: ignore[type-arg] + + async def capture(request: Request): # type: ignore[no-untyped-def] + yield + # By the time this finaliser runs, all pushes have been recorded. + captured.append( + list(request.scope.get("fastapi_dependency_cleanup_order", [])) + ) + + def dep_a(): # type: ignore[no-untyped-def] + yield "a" + + def dep_b(): # type: ignore[no-untyped-def] + yield "b" + + app = FastAPI(dependency_debug_url="/_debug/deps") + + @app.get("/") + def endpoint( + _cap: Annotated[None, Depends(capture)], + _a: Annotated[str, Depends(dep_a)], + _b: Annotated[str, Depends(dep_b)], + ) -> dict: + return {"ok": True} + + with TestClient(app) as client: + resp = client.get("/") + assert resp.status_code == 200 + + assert len(captured) == 1 + metadata = captured[0] + + # Three generators were pushed: capture, dep_a, dep_b + assert len(metadata) == 3 + + # Order indices are monotonically increasing from 0 + assert [m["order"] for m in metadata] == [0, 1, 2] + + # Names match push order (signature order) + assert metadata[0]["callable_name"] == "capture" + assert metadata[1]["callable_name"] == "dep_a" + assert metadata[2]["callable_name"] == "dep_b" + + # All are request-scoped (no explicit scope on generators → default "request") + for m in metadata: + assert m["scope"] == "request" + + +# --------------------------------------------------------------------------- +# test_cleanup_order_metadata_survives_exception +# --------------------------------------------------------------------------- + + +def test_cleanup_order_metadata_survives_exception() -> None: + """If a generator dep raises *before* its yield (setup failure), it is + never pushed onto the stack and therefore never recorded. Deps that were + successfully pushed before it still appear in the metadata.""" + captured: list[list[dict]] = [] # type: ignore[type-arg] + + async def capture(request: Request): # type: ignore[no-untyped-def] + try: + yield + finally: + captured.append( + list(request.scope.get("fastapi_dependency_cleanup_order", [])) + ) + + def dep_ok(): # type: ignore[no-untyped-def] + yield "ok" + + def dep_raises(): # type: ignore[no-untyped-def] + raise ValueError("setup failure") + yield "never" # pragma: no cover # noqa: RET503 + + app = FastAPI(dependency_debug_url="/_debug/deps") + + @app.get("/") + def endpoint( + _cap: Annotated[None, Depends(capture)], + _ok: Annotated[str, Depends(dep_ok)], + _bad: Annotated[str, Depends(dep_raises)], + ) -> dict: + return {"ok": True} # pragma: no cover + + with TestClient(app, raise_server_exceptions=False) as client: + resp = client.get("/") + # The ValueError propagates → 500 + assert resp.status_code == 500 + + # capture and dep_ok were pushed; dep_raises was not + assert len(captured) == 1 + metadata = captured[0] + assert len(metadata) == 2 + names = [m["callable_name"] for m in metadata] + assert "capture" in names + assert "dep_ok" in names + assert "dep_raises" not in names + + +# --------------------------------------------------------------------------- +# test_use_cache_false_shared_dep_cleanup_runs_per_encounter +# --------------------------------------------------------------------------- + + +def test_use_cache_false_shared_dep_cleanup_runs_per_encounter() -> None: + """use_cache=False on a shared yield dep bypasses the cache check + (utils.py:674) on every encounter. Each encounter calls + enter_async_context independently → one push per encounter, one + finaliser per push. The two finalisers are NOT adjacent in the LIFO + sequence; they are interleaved with the branch deps that triggered them. + + Expected full event sequence (setup + cleanup): + shared:setup branch_a:setup shared:setup branch_b:setup + branch_b:cleanup shared:cleanup branch_a:cleanup shared:cleanup + + Failure mode guarded: an optimisation that deduplicates + enter_async_context calls regardless of use_cache would collapse the + two shared finalisers into one, silently dropping cleanup work.""" + order: list[str] = [] + + def shared(): # type: ignore[no-untyped-def] + order.append("shared:setup") + yield "s" + order.append("shared:cleanup") + + def branch_a(s: Annotated[str, Depends(shared, use_cache=False)]): # type: ignore[no-untyped-def] + order.append("branch_a:setup") + yield "a" + order.append("branch_a:cleanup") + + def branch_b(s: Annotated[str, Depends(shared, use_cache=False)]): # type: ignore[no-untyped-def] + order.append("branch_b:setup") + yield "b" + order.append("branch_b:cleanup") + + app = FastAPI() + + @app.get("/") + def endpoint( + _a: Annotated[str, Depends(branch_a)], + _b: Annotated[str, Depends(branch_b)], + ) -> dict: + return {} + + with TestClient(app) as client: + resp = client.get("/") + assert resp.status_code == 200 + assert order == [ + "shared:setup", + "branch_a:setup", + "shared:setup", + "branch_b:setup", + "branch_b:cleanup", + "shared:cleanup", + "branch_a:cleanup", + "shared:cleanup", + ] + + +# --------------------------------------------------------------------------- +# test_background_task_observes_function_scoped_resource_as_invalidated +# --------------------------------------------------------------------------- + + +def test_background_task_observes_function_scoped_resource_as_invalidated() -> None: + """function_stack exits before response() is called. BackgroundTasks + execute inside response(). A function-scoped dep that invalidates a + shared mutable resource during its cleanup does so *before* any + BackgroundTask reads it. + + Expected state seen by BackgroundTask: 'closed' (not 'open'). + + Failure mode guarded: moving ``await response(…)`` inside the + function_stack async-with block would make function-scoped resources + still alive during BackgroundTask execution, masking use-after-close + bugs in production.""" + state: dict[str, str] = {} + bg_saw: list[str] = [] + + def func_dep(): # type: ignore[no-untyped-def] + state["status"] = "open" + yield state + state["status"] = "closed" + + app = FastAPI() + + @app.get("/") + def endpoint( + res: Annotated[dict, Depends(func_dep, scope="function")], + bg: BackgroundTasks, + ) -> dict: + bg.add_task(lambda: bg_saw.append(state["status"])) + return {} + + with TestClient(app) as client: + resp = client.get("/") + assert resp.status_code == 200 + assert bg_saw == ["closed"] + + +# --------------------------------------------------------------------------- +# test_function_scope_cleanup_can_enqueue_background_task +# --------------------------------------------------------------------------- + + +def test_function_scope_cleanup_can_enqueue_background_task() -> None: + """function_stack exits before BackgroundTasks.__call__ snapshots + self.tasks and sets _executed. A function-scoped dep's cleanup code + that calls bg.add_task() runs before the snapshot → the task is + captured and executed. + + Expected run order: endpoint, from_func_cleanup. + + Failure mode guarded: moving the _executed flag or the task snapshot + to before function_stack exit would silently drop tasks enqueued + during function-scoped cleanup.""" + ran: list[str] = [] + + def func_dep(bg: BackgroundTasks): # type: ignore[no-untyped-def] + yield + bg.add_task(lambda: ran.append("from_func_cleanup")) + + app = FastAPI() + + @app.get("/") + def endpoint(_f: Annotated[None, Depends(func_dep, scope="function")]) -> dict: + ran.append("endpoint") + return {} + + with TestClient(app) as client: + resp = client.get("/") + assert resp.status_code == 200 + assert ran == ["endpoint", "from_func_cleanup"] + + +# --------------------------------------------------------------------------- +# test_request_scope_cleanup_cannot_enqueue_background_task +# --------------------------------------------------------------------------- + + +def test_request_scope_cleanup_cannot_enqueue_background_task(caplog) -> None: + """request_stack exits after BackgroundTasks.__call__ has already set + _executed=True. A request-scoped dep's cleanup call to add_task hits + the guard in background.py, logs a warning, and the task is never + added to the list. + + Expected: 'from_req_cleanup' does NOT appear in ran; a log warning + containing 'already been executed' is emitted. + + Failure mode guarded: removing or deferring the _executed guard would + allow request-scoped cleanup to silently enqueue tasks that execute + in an undefined order relative to the response lifecycle.""" + ran: list[str] = [] + + def req_dep(bg: BackgroundTasks): # type: ignore[no-untyped-def] + yield + bg.add_task(lambda: ran.append("from_req_cleanup")) + + app = FastAPI() + + @app.get("/") + def endpoint(_r: Annotated[None, Depends(req_dep)]) -> dict: + ran.append("endpoint") + return {} + + with TestClient(app) as client: + with caplog.at_level(logging.WARNING, logger="fastapi.background"): + resp = client.get("/") + assert resp.status_code == 200 + assert "from_req_cleanup" not in ran + assert any("already been executed" in r.message for r in caplog.records) + + +# --------------------------------------------------------------------------- +# test_transitive_scope_leak_through_plain_wrapper +# --------------------------------------------------------------------------- + + +def test_transitive_scope_leak_through_plain_wrapper() -> None: + """The DependencyScopeError guard (utils.py:284-287) checks only the + direct child's scope field. A plain (non-generator) function inserted + between a request-scoped generator parent and a function-scoped + generator grandchild has scope=None; the guard does not fire. The + plain wrapper returns the same object the function-scoped dep yielded. + By the time the request-scoped parent's cleanup runs, the + function-scoped grandchild has already invalidated that object. + + Expected: route registers without DependencyScopeError; request_dep's + cleanup observes resource status == 'closed'. + + Failure mode guarded (documentation): this test documents a known + single-depth limitation. If transitive scope validation is added, + this test must be updated to reflect the new behaviour.""" + cleanup_saw: list[str] = [] + resource: dict[str, str] = {} + + def func_dep(): # type: ignore[no-untyped-def] + resource["status"] = "open" + yield resource + resource["status"] = "closed" + + def plain_wrapper( + res: Annotated[dict, Depends(func_dep, scope="function")], + ) -> dict: + return res # pass-through; not a generator + + def request_dep(res: Annotated[dict, Depends(plain_wrapper)]): # type: ignore[no-untyped-def] + yield + cleanup_saw.append(res["status"]) + + app = FastAPI() + + @app.get("/") + def endpoint(_r: Annotated[None, Depends(request_dep)]) -> dict: + return {} + + with TestClient(app) as client: + resp = client.get("/") + assert resp.status_code == 200 + assert cleanup_saw == ["closed"] + + +# --------------------------------------------------------------------------- +# test_multiple_route_deps_exit_in_reverse_declaration_order +# --------------------------------------------------------------------------- + + +def test_multiple_route_deps_exit_in_reverse_declaration_order() -> None: + """Route-level deps are prepended via ``dependencies[::-1]`` + + ``insert(0, …)`` (routing.py:479-483), which preserves their original + declaration order at the front of dependant.dependencies. They are + therefore pushed first and exit last, in reverse declaration order + (LIFO). + + Declaration order: r1, r2, r3. + Push order: r1, r2, r3. + Cleanup order: r3, r2, r1. + + Failure mode guarded: changing the reversal + insert pattern (e.g., + a plain append loop) would reverse the declaration order of route + deps at the front of the list, producing cleanup order r1, r2, r3.""" + order: list[str] = [] + + def r1(): # type: ignore[no-untyped-def] + yield + order.append("r1") + + def r2(): # type: ignore[no-untyped-def] + yield + order.append("r2") + + def r3(): # type: ignore[no-untyped-def] + yield + order.append("r3") + + app = FastAPI() + + @app.get("/", dependencies=[Depends(r1), Depends(r2), Depends(r3)]) + def endpoint() -> dict: + return {} + + with TestClient(app) as client: + resp = client.get("/") + assert resp.status_code == 200 + assert order == ["r3", "r2", "r1"] + + +# --------------------------------------------------------------------------- +# test_shared_dep_cleanup_position_is_first_dfs_encounter +# --------------------------------------------------------------------------- + + +def test_shared_dep_cleanup_position_is_first_dfs_encounter() -> None: + """With use_cache=True (the default) a shared dep is pushed exactly + once — during the first branch whose DFS recursion reaches it. + Subsequent encounters hit the cache (utils.py:674) and skip the push. + The shared dep's cleanup position in the LIFO sequence is therefore + anchored to whichever branch appears first in the signature, not to + where the shared dep is declared. + + Both sub-cases below put shared as the deepest node. Swapping which + branch is listed first swaps the two branches' cleanup positions but + shared always exits last (it was pushed first). + + Failure mode guarded: changing cache-store or cache-check semantics + (e.g., storing only when use_cache=True) would move the shared dep's + cleanup to a different position depending on encounter order.""" + + # ── sub-case: branch_a listed first ── + order_a: list[str] = [] + + def shared_a(): # type: ignore[no-untyped-def] + yield "s" + order_a.append("shared") + + def ba(s: Annotated[str, Depends(shared_a)]): # type: ignore[no-untyped-def] + yield "a" + order_a.append("branch_a") + + def bb_a(s: Annotated[str, Depends(shared_a)]): # type: ignore[no-untyped-def] + yield "b" + order_a.append("branch_b") + + app_a = FastAPI() + + @app_a.get("/") + def ep_a( + _a: Annotated[str, Depends(ba)], + _b: Annotated[str, Depends(bb_a)], + ) -> dict: + return {} + + with TestClient(app_a) as client: + client.get("/") + # push: shared, branch_a, branch_b → LIFO: branch_b, branch_a, shared + assert order_a == ["branch_b", "branch_a", "shared"] + + # ── sub-case: branch_b listed first ── + order_b: list[str] = [] + + def shared_b(): # type: ignore[no-untyped-def] + yield "s" + order_b.append("shared") + + def ba_b(s: Annotated[str, Depends(shared_b)]): # type: ignore[no-untyped-def] + yield "a" + order_b.append("branch_a") + + def bb(s: Annotated[str, Depends(shared_b)]): # type: ignore[no-untyped-def] + yield "b" + order_b.append("branch_b") + + app_b = FastAPI() + + @app_b.get("/") + def ep_b( + _b: Annotated[str, Depends(bb)], + _a: Annotated[str, Depends(ba_b)], + ) -> dict: + return {} + + with TestClient(app_b) as client: + client.get("/") + # push: shared, branch_b, branch_a → LIFO: branch_a, branch_b, shared + assert order_b == ["branch_a", "branch_b", "shared"] + + +# --------------------------------------------------------------------------- +# test_dep_setup_failure_partial_cleanup +# --------------------------------------------------------------------------- + + +def test_dep_setup_failure_partial_cleanup() -> None: + """When a dep raises before its yield (setup failure) the exception + propagates out of solve_dependencies. Deps that were already pushed + onto the stack before the failing dep are unwound normally (LIFO). + Deps declared after the failing dep in the signature are never + reached and never pushed — their cleanup does not run. + + Signature order: dep_ok_1, dep_boom, dep_ok_2. + dep_ok_1 is pushed. dep_boom raises during setup — never pushed. + dep_ok_2 is never reached. + + Expected cleanup set: dep_ok_1 only. + + Failure mode guarded: an 'all or nothing' error handler that skips + stack unwinding on setup failure would prevent dep_ok_1 from + cleaning up, leaking its resource.""" + order: list[str] = [] + + def dep_ok_1(): # type: ignore[no-untyped-def] + try: + yield "ok1" + finally: + order.append("dep_ok_1:cleanup") + + def dep_boom(): # type: ignore[no-untyped-def] + raise RuntimeError("setup boom") + yield # pragma: no cover # noqa: RET503 + + def dep_ok_2(): # type: ignore[no-untyped-def] + try: + yield "ok2" + finally: + order.append("dep_ok_2:cleanup") + + app = FastAPI() + + @app.get("/") + def endpoint( + _a: Annotated[str, Depends(dep_ok_1)], + _b: Annotated[str, Depends(dep_boom)], + _c: Annotated[str, Depends(dep_ok_2)], + ) -> dict: + return {} # pragma: no cover + + with TestClient(app, raise_server_exceptions=False) as client: + resp = client.get("/") + assert resp.status_code == 500 + assert order == ["dep_ok_1:cleanup"] diff --git a/tests/test_dependency_debug.py b/tests/test_dependency_debug.py new file mode 100644 index 000000000..ce3ce2f27 --- /dev/null +++ b/tests/test_dependency_debug.py @@ -0,0 +1,341 @@ +"""Tests for the dependency-graph debug endpoint (Task 1).""" + +import functools +from typing import Annotated + +from fastapi import Depends, FastAPI, Security +from fastapi.security import OAuth2PasswordBearer +from fastapi.testclient import TestClient + + +# --------------------------------------------------------------------------- +# Helpers reused across tests +# --------------------------------------------------------------------------- + +_DEBUG_PATH = "/_debug/deps" + + +def _app(**kwargs) -> FastAPI: # type: ignore[no-untyped-def] + return FastAPI(dependency_debug_url=_DEBUG_PATH, **kwargs) + + +def _find_route(payload: dict, path: str) -> dict: # type: ignore[no-untyped-def] + for r in payload["routes"]: + if r["path"] == path: + return r + raise AssertionError(f"Route {path!r} not found in debug payload") # pragma: no cover + + +# --------------------------------------------------------------------------- +# test_debug_disabled_by_default +# --------------------------------------------------------------------------- + + +def test_debug_disabled_by_default() -> None: + app = FastAPI() # dependency_debug_url not set → None + + @app.get("/hello") + def hello() -> dict: + return {"hi": True} + + client = TestClient(app) + # The default path and a few guesses must all 404 + assert client.get("/_debug/deps").status_code == 404 + assert client.get("/_debug/dependencies").status_code == 404 + assert client.get("/debug").status_code == 404 + + +# --------------------------------------------------------------------------- +# test_debug_enabled_returns_200 +# --------------------------------------------------------------------------- + + +def test_debug_enabled_returns_200() -> None: + app = _app() + + @app.get("/hello") + def hello() -> dict: + return {"hi": True} + + client = TestClient(app) + resp = client.get(_DEBUG_PATH) + assert resp.status_code == 200 + + payload = resp.json() + assert "routes" in payload + route = _find_route(payload, "/hello") + assert "GET" in route["methods"] + assert route["dependency_graph"]["callable_name"] == "hello" + + +# --------------------------------------------------------------------------- +# test_debug_nested_three_levels +# --------------------------------------------------------------------------- + + +def test_debug_nested_three_levels() -> None: + def dep_c() -> str: + return "c" + + def dep_b(c: Annotated[str, Depends(dep_c)]) -> str: + return "b" + + def dep_a(b: Annotated[str, Depends(dep_b)]) -> str: + return "a" + + app = _app() + + @app.get("/nested") + def nested(a: Annotated[str, Depends(dep_a)]) -> dict: + return {"a": a} + + client = TestClient(app) + route = _find_route(client.get(_DEBUG_PATH).json(), "/nested") + graph = route["dependency_graph"] + + # Root is the endpoint itself + assert graph["callable_name"] == "nested" + + # Level 1: dep_a + assert len(graph["sub_dependencies"]) == 1 + a_node = graph["sub_dependencies"][0] + assert a_node["callable_name"] == "dep_a" + + # Level 2: dep_b + assert len(a_node["sub_dependencies"]) == 1 + b_node = a_node["sub_dependencies"][0] + assert b_node["callable_name"] == "dep_b" + + # Level 3: dep_c — leaf + assert len(b_node["sub_dependencies"]) == 1 + c_node = b_node["sub_dependencies"][0] + assert c_node["callable_name"] == "dep_c" + assert c_node["sub_dependencies"] == [] + + +# --------------------------------------------------------------------------- +# test_debug_yield_dep_flagged +# --------------------------------------------------------------------------- + + +def test_debug_yield_dep_flagged() -> None: + def plain_dep() -> str: + return "plain" + + def yield_dep(): # type: ignore[no-untyped-def] + yield "yielded" + + app = _app() + + @app.get("/mix") + def mix( + p: Annotated[str, Depends(plain_dep)], + y: Annotated[str, Depends(yield_dep)], + ) -> dict: + return {} + + route = _find_route(TestClient(app).get(_DEBUG_PATH).json(), "/mix") + subs = route["dependency_graph"]["sub_dependencies"] + + # Order matches signature order: plain first, yield second + plain_node = subs[0] + yield_node = subs[1] + + assert plain_node["callable_name"] == "plain_dep" + assert plain_node["is_yield"] is False + + assert yield_node["callable_name"] == "yield_dep" + assert yield_node["is_yield"] is True + + +# --------------------------------------------------------------------------- +# test_debug_async_generator_flagged +# --------------------------------------------------------------------------- + + +def test_debug_async_generator_flagged() -> None: + async def async_gen_dep(): # type: ignore[no-untyped-def] + yield "async" + + app = _app() + + @app.get("/agen") + def agen(v: Annotated[str, Depends(async_gen_dep)]) -> dict: + return {} + + node = _find_route(TestClient(app).get(_DEBUG_PATH).json(), "/agen")["dependency_graph"] + sub = node["sub_dependencies"][0] + + assert sub["callable_name"] == "async_gen_dep" + assert sub["is_yield"] is True + assert sub["is_async"] is True + + +# --------------------------------------------------------------------------- +# test_debug_security_scopes +# --------------------------------------------------------------------------- + + +def test_debug_security_scopes() -> None: + oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") + + async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> dict: + return {"user": "me"} + + app = _app() + + @app.get("/protected") + def protected( + user: Annotated[dict, Security(get_current_user, scopes=["read", "admin"])], + ) -> dict: + return user + + route = _find_route(TestClient(app).get(_DEBUG_PATH).json(), "/protected") + # The Security node is a direct sub of the endpoint root + sec_node = route["dependency_graph"]["sub_dependencies"][0] + assert sec_node["callable_name"] == "get_current_user" + assert "read" in sec_node["security_scopes"] + assert "admin" in sec_node["security_scopes"] + + +# --------------------------------------------------------------------------- +# test_debug_security_and_depends_siblings +# --------------------------------------------------------------------------- + + +def test_debug_security_and_depends_siblings() -> None: + oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") + + def plain_dep() -> str: + return "plain" + + app = _app() + + @app.get("/siblings") + def siblings( + token: Annotated[str, Security(oauth2_scheme, scopes=["x"])], + plain: Annotated[str, Depends(plain_dep)], + ) -> dict: + return {} + + route = _find_route(TestClient(app).get(_DEBUG_PATH).json(), "/siblings") + subs = route["dependency_graph"]["sub_dependencies"] + + # First sub is the Security dep (oauth2_scheme instance) + sec_sub = subs[0] + assert sec_sub["is_security_scheme"] is True + + # Second sub is the plain Depends + plain_sub = subs[1] + assert plain_sub["callable_name"] == "plain_dep" + assert plain_sub["is_security_scheme"] is False + + +# --------------------------------------------------------------------------- +# test_debug_route_level_deps_appear +# --------------------------------------------------------------------------- + + +def test_debug_route_level_deps_appear() -> None: + def route_dep() -> None: + pass + + app = _app() + + @app.get("/with-route-dep", dependencies=[Depends(route_dep)]) + def with_route_dep() -> dict: + return {} + + route = _find_route(TestClient(app).get(_DEBUG_PATH).json(), "/with-route-dep") + subs = route["dependency_graph"]["sub_dependencies"] + + # route_dep is prepended, so it is the first child + assert subs[0]["callable_name"] == "route_dep" + + +# --------------------------------------------------------------------------- +# test_debug_partial_callable_unwrapped +# --------------------------------------------------------------------------- + + +def test_debug_partial_callable_unwrapped() -> None: + def real_function(x: int = 1) -> int: + return x + + partial_dep = functools.partial(real_function, x=42) + + app = _app() + + @app.get("/partial") + def partial_endpoint(v: Annotated[int, Depends(partial_dep)]) -> dict: + return {"v": v} + + route = _find_route(TestClient(app).get(_DEBUG_PATH).json(), "/partial") + sub = route["dependency_graph"]["sub_dependencies"][0] + + # Must resolve through the partial to the underlying function name + assert sub["callable_name"] == "real_function" + + +# --------------------------------------------------------------------------- +# test_debug_class_based_dep +# --------------------------------------------------------------------------- + + +def test_debug_class_based_dep() -> None: + class MyDep: + def __call__(self) -> str: + return "class-dep" + + app = _app() + + @app.get("/classdep") + def classdep(v: Annotated[str, Depends(MyDep)]) -> dict: + return {"v": v} + + route = _find_route(TestClient(app).get(_DEBUG_PATH).json(), "/classdep") + sub = route["dependency_graph"]["sub_dependencies"][0] + + assert sub["callable_name"] == "MyDep" + + +# --------------------------------------------------------------------------- +# test_debug_cached_repeat_detection +# --------------------------------------------------------------------------- + + +def test_debug_cached_repeat_detection() -> None: + """A shared dep used by two sibling branches must appear fully in both + (it is NOT on the same root→leaf path). This test verifies no spurious + ``cached_repeat`` flag is set in that scenario.""" + + def shared() -> str: + return "shared" + + def branch_a(s: Annotated[str, Depends(shared)]) -> str: + return "a" + + def branch_b(s: Annotated[str, Depends(shared)]) -> str: + return "b" + + app = _app() + + @app.get("/shared") + def endpoint( + a: Annotated[str, Depends(branch_a)], + b: Annotated[str, Depends(branch_b)], + ) -> dict: + return {} + + route = _find_route(TestClient(app).get(_DEBUG_PATH).json(), "/shared") + subs = route["dependency_graph"]["sub_dependencies"] + + # branch_a's child "shared" should be fully expanded (no cached_repeat) + shared_under_a = subs[0]["sub_dependencies"][0] + assert shared_under_a["callable_name"] == "shared" + assert "cached_repeat" not in shared_under_a + + # branch_b's child "shared" should also be fully expanded + shared_under_b = subs[1]["sub_dependencies"][0] + assert shared_under_b["callable_name"] == "shared" + assert "cached_repeat" not in shared_under_b