From 1c023cf9a2bd6185eea4478615c87f81421ffb17 Mon Sep 17 00:00:00 2001 From: Shaggi Date: Wed, 29 Jul 2026 22:53:43 +0300 Subject: [PATCH 1/4] fix: scope framework surfaces to selected app --- README.md | 7 +- .../analyzer/change_mapper.py | 2 + src/fastapi_endpoint_detector/cli.py | 4 + .../parser/custom_surface_extractor.py | 377 +++++++++++++++++- .../presets/framework_v1.yaml | 30 +- tests/unit/test_framework_surfaces.py | 160 +++++++- 6 files changed, 552 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index d61b3bb..13a4a19 100644 --- a/README.md +++ b/README.md @@ -171,8 +171,11 @@ benchmark). `--app-entry` and `--bootstrap-entry` require `--secure-ast` and never execute application code. Bootstrap interpretation is explicitly seeded, bounded to project-local straight-line helpers, and records unsupported object escape, mutation, control flow, or dynamic registration as conditional evidence; -no bootstrap/helper names are guessed. Dynamic behavior remains conservatively -unresolved. +no bootstrap/helper names are guessed. The bundled `framework-v1` surface preset +uses the same selected app/app-factory identity: unused and mounted child app +lifecycles are excluded, router lifecycle handlers use include-time copy evidence, +and dynamic router inclusion makes the surface inventory conditional. Dynamic +behavior remains conservatively unresolved. ### `list` - List Endpoints diff --git a/src/fastapi_endpoint_detector/analyzer/change_mapper.py b/src/fastapi_endpoint_detector/analyzer/change_mapper.py index 0399cae..3d44112 100644 --- a/src/fastapi_endpoint_detector/analyzer/change_mapper.py +++ b/src/fastapi_endpoint_detector/analyzer/change_mapper.py @@ -542,6 +542,8 @@ def _merge_surface_inventory( app_path, self._surface_contracts, bootstrap_entry=self.bootstrap_entry, + app_variable=self.app_variable, + app_entry=self.app_entry, ).extract_inventory() return merge_surface_inventory(native, custom) diff --git a/src/fastapi_endpoint_detector/cli.py b/src/fastapi_endpoint_detector/cli.py index 58597d8..3fabc5b 100644 --- a/src/fastapi_endpoint_detector/cli.py +++ b/src/fastapi_endpoint_detector/cli.py @@ -549,6 +549,8 @@ def audit_effect_contracts_command( app, loaded_surfaces, bootstrap_entry=bootstrap_entry, + app_variable=app_var, + app_entry=app_entry, ).extract_inventory() inventory = merge_surface_inventory(inventory, custom) source_root = app.resolve().parent if app.is_file() else app.resolve() @@ -755,6 +757,8 @@ def list_endpoints( app, loaded_surfaces, bootstrap_entry=bootstrap_entry, + app_variable=app_var, + app_entry=app_entry, ).extract_inventory() inventory = merge_surface_inventory(inventory, custom) endpoints = inventory.endpoints diff --git a/src/fastapi_endpoint_detector/parser/custom_surface_extractor.py b/src/fastapi_endpoint_detector/parser/custom_surface_extractor.py index f0a7b2c..dcf272a 100644 --- a/src/fastapi_endpoint_detector/parser/custom_surface_extractor.py +++ b/src/fastapi_endpoint_detector/parser/custom_surface_extractor.py @@ -72,6 +72,25 @@ class _ResolvedResources: reason: str +_FrameworkToken = tuple[str, int, int] + + +@dataclass(frozen=True) +class _FrameworkRegistrationEvent: + token: _FrameworkToken + endpoint: Endpoint + + +@dataclass(frozen=True) +class _FrameworkIncludeEvent: + parent: _FrameworkToken + child: _FrameworkToken | None + condition: EndpointDiscoveryCondition | None + + +_FrameworkEvent = _FrameworkRegistrationEvent | _FrameworkIncludeEvent + + @dataclass(frozen=True) class _StartupRouteResult: route: ( @@ -688,10 +707,15 @@ def __init__( app_path: Path, contracts: LoadedSurfaceContracts, bootstrap_entry: str | None = None, + *, + app_variable: str = "app", + app_entry: str | None = None, ) -> None: self.app_path = app_path.resolve() self.contracts = contracts self.bootstrap_entry = bootstrap_entry + self.app_variable = app_variable + self.app_entry = app_entry if self.app_path.is_dir(): self.root = self.app_path elif self.app_path.is_file(): @@ -714,6 +738,13 @@ def __init__( self._function_scope_states: list[_FunctionScopeFrame] = [] self._building_states = False self._inventory_unavailable = False + self._framework_events: list[_FrameworkEvent] = [] + self._framework_selected_tokens: set[_FrameworkToken] = set() + self._framework_root_condition: EndpointDiscoveryCondition | None = None + self._framework_factory: tuple[_Module, ast.FunctionDef | ast.AsyncFunctionDef] | None = ( + None + ) + self._scope_framework_surfaces = contracts.document.preset.id == "framework-callbacks" self._declared_receiver_types = { contract.registration.receiver_type for contract in contracts.document.contracts @@ -749,6 +780,7 @@ def extract_inventory(self) -> EndpointInventory: self._module_states[module.name] = state finally: self._building_states = False + self._resolve_framework_root() for module in self._modules.values(): self._process_statements( module, @@ -757,7 +789,9 @@ def extract_inventory(self) -> EndpointInventory: (), evaluate_variable_annotations=not module.postponed_annotations, ) + self._process_app_factory() self._process_bootstrap() + self._filter_framework_surfaces() collapsed: dict[tuple[str, str, int], Endpoint] = {} for endpoint in self._endpoints: key = ( @@ -845,6 +879,224 @@ def extract_inventory(self) -> EndpointInventory: route_conditions=route_conditions, ) + @staticmethod + def _framework_receiver_token(binding: _Binding | None) -> _FrameworkToken | None: + if binding is None or binding.kind != "receiver" or binding.instance_token is None: + return None + if binding.identity not in { + "fastapi.FastAPI", + "starlette.applications.Starlette", + "fastapi.APIRouter", + }: + return None + return binding.instance_token + + def _framework_condition(self, module: _Module, line: int, reason: str) -> None: + condition = EndpointDiscoveryCondition( + source_path=module.path, + source_line=line, + reason=reason, + ) + self._framework_root_condition = condition + + def _resolve_framework_root(self) -> None: # noqa: PLR0912 + """Resolve the exact selected application after source-ordered module binding.""" + if not self._scope_framework_surfaces: + return + candidates: list[tuple[_Module, str]] = [] + if self.app_entry is not None: + parts = self.app_entry.split(":") + if ( + len(parts) != 2 + or not parts[0] + or not parts[1] + or any(not item.isidentifier() for item in parts[0].split(".")) + or not parts[1].isidentifier() + ): + raise CustomSurfaceExtractorError( + "app_entry must use an exact project-local MODULE:SYMBOL" + ) + module = self._modules.get(parts[0]) + if module is None: + raise CustomSurfaceExtractorError( + f"custom surface app entry {self.app_entry!r} has no project module" + ) + candidates.append((module, parts[1])) + elif self.app_path.is_file(): + module = next( + (item for item in self._modules.values() if item.path == self.app_path), + None, + ) + if module is not None: + candidates.append((module, self.app_variable)) + else: + candidates.extend((module, self.app_variable) for module in self._modules.values()) + + selected: set[_FrameworkToken] = set() + unresolved: list[tuple[_Module, str]] = [] + for module, symbol in candidates: + binding = self._module_states.get(module.name, {}).get(symbol) + if binding is not None: + binding = self._follow_project_binding(binding) + token = self._framework_receiver_token(binding) + if ( + token is not None + and binding is not None + and binding.identity != "fastapi.APIRouter" + ): + selected.add(token) + continue + function_candidates = self._functions.get(f"{module.name}.{symbol}", []) + if self.app_entry is not None and len(function_candidates) == 1: + self._framework_factory = function_candidates[0] + continue + if self.app_entry is not None or binding is not None: + unresolved.append((module, symbol)) + + if len(selected) == 1: + self._framework_selected_tokens = selected + return + if len(selected) > 1: + module, _symbol = candidates[0] + self._framework_condition( + module, + 1, + "selected framework application binding is ambiguous across project modules", + ) + return + if self._framework_factory is not None: + return + module, symbol = ( + unresolved or candidates or [(next(iter(self._modules.values())), self.app_variable)] + )[0] + rebound_lines: list[int] = [] + for statement in module.tree.body: + visitor = _EagerStateMutationVisitor() + visitor.visit(statement) + if symbol in visitor.rebound_names: + rebound_lines.append(statement.lineno) + line = max(rebound_lines, default=1) + self._framework_condition( + module, + line, + "selected framework application binding is unresolved or was rebound", + ) + + def _process_app_factory(self) -> None: + """Interpret one explicitly selected zero-argument factory and retain its return token.""" + if self._framework_factory is None: + return + module, function = self._framework_factory + if isinstance(function, ast.AsyncFunctionDef) or function.decorator_list: + raise CustomSurfaceExtractorError( + "custom surface app factory must be synchronous and undecorated" + ) + positional = [*function.args.posonlyargs, *function.args.args] + if ( + len(positional) - len(function.args.defaults) + or any(default is None for default in function.args.kw_defaults) + or function.args.vararg is not None + or function.args.kwarg is not None + ): + raise CustomSurfaceExtractorError( + "custom surface app factory must be callable with zero arguments and not variadic" + ) + returns = [item for item in function.body if isinstance(item, ast.Return)] + if len(returns) != 1 or function.body[-1] is not returns[0] or returns[0].value is None: + self._framework_condition( + module, + function.lineno, + "selected framework app factory has unsupported return control flow", + ) + return + + state = dict(self._module_states[module.name]) + scope = _FunctionScopeBindingVisitor(function) + scope.visit(function) + local_names = scope.rebound_names - scope.globals - scope.nonlocals + for name in local_names: + state[name] = None + frame = _FunctionScopeFrame( + local_state=state, + global_state=self._module_states[module.name], + local_names=frozenset(local_names), + ) + self._function_scope_states.append(frame) + try: + self._process_statements( + module, + function.body, + state, + (), + evaluate_variable_annotations=False, + ) + finally: + self._function_scope_states.pop() + token = self._framework_receiver_token( + self._binding_from_expression(returns[0].value, state, module.name) + ) + if token is None: + self._framework_condition( + module, + returns[0].lineno, + "selected framework app factory return is unresolved or not an application", + ) + return + self._framework_selected_tokens = {token} + + @staticmethod + def _is_framework_endpoint(endpoint: Endpoint) -> bool: + return endpoint.surface is not None and endpoint.surface.surface_kind.startswith( + "framework." + ) + + def _filter_framework_surfaces(self) -> None: + """Apply selected-app identity and APIRouter copy-at-include semantics.""" + if not self._scope_framework_surfaces: + return + live: dict[_FrameworkToken, list[Endpoint]] = {} + conditions: dict[_FrameworkToken, list[EndpointDiscoveryCondition]] = {} + included_by: dict[_FrameworkToken, set[_FrameworkToken]] = {} + for event in self._framework_events: + if isinstance(event, _FrameworkRegistrationEvent): + live.setdefault(event.token, []).append(event.endpoint) + surface = event.endpoint.surface + if surface is not None: + for parent in included_by.get(event.token, ()): + conditions.setdefault(parent, []).append( + EndpointDiscoveryCondition( + source_path=surface.registration_file, + source_line=surface.registration_line, + reason=( + "router lifecycle registered after include_router has " + "runtime-version-dependent execution" + ), + ) + ) + continue + if event.child is None: + if event.condition is not None: + conditions.setdefault(event.parent, []).append(event.condition) + continue + live.setdefault(event.parent, []).extend(live.get(event.child, ())) + conditions.setdefault(event.parent, []).extend(conditions.get(event.child, ())) + included_by.setdefault(event.child, set()).add(event.parent) + + accepted = { + id(endpoint) + for token in self._framework_selected_tokens + for endpoint in live.get(token, ()) + } + self._endpoints = [ + endpoint + for endpoint in self._endpoints + if not self._is_framework_endpoint(endpoint) or id(endpoint) in accepted + ] + for token in self._framework_selected_tokens: + self._limitations.extend(conditions.get(token, ())) + if self._framework_root_condition is not None: + self._limitations.append(self._framework_root_condition) + def _process_bootstrap(self) -> None: if self.bootstrap_entry is None: return @@ -2103,6 +2355,86 @@ def _resolve_captured_class_method( required_base, ) + def _framework_call_token( + self, + call: ast.Call, + evaluation: _CallEvaluation | None, + state: dict[str, _Binding | None], + ) -> _FrameworkToken | None: + if not isinstance(call.func, ast.Attribute): + return None + lookup = evaluation.callable_state if evaluation is not None else state + return self._framework_receiver_token(self._expression_binding(call.func.value, lookup)) + + def _record_framework_include( + self, + module: _Module, + call: ast.Call, + state: dict[str, _Binding | None], + evaluation: _CallEvaluation | None, + ) -> None: + if not self._scope_framework_surfaces or not isinstance(call.func, ast.Attribute): + return + if call.func.attr != "include_router": + return + parent = self._framework_call_token(call, evaluation, state) + if parent is None: + return + capture = ( + evaluation.positional[0] if evaluation is not None and evaluation.positional else None + ) + if capture is None and evaluation is not None: + capture = next( + ( + item + for keyword, item in zip(call.keywords, evaluation.keywords, strict=True) + if keyword.arg == "router" + ), + None, + ) + child = self._framework_receiver_token(capture.binding if capture is not None else None) + condition = None + if child is None: + condition = EndpointDiscoveryCondition( + source_path=module.path, + source_line=call.lineno, + reason=( + "selected application include_router target is dynamic or unresolved; " + "framework surface inventory is incomplete" + ), + ) + self._framework_events.append( + _FrameworkIncludeEvent(parent=parent, child=child, condition=condition) + ) + + def _record_framework_registration( + self, + module: _Module, + call: ast.Call, + state: dict[str, _Binding | None], + evaluation: _CallEvaluation | None, + endpoint: Endpoint, + ) -> None: + if not self._scope_framework_surfaces or not self._is_framework_endpoint(endpoint): + return + token = ( + self._framework_call_token(call, evaluation, state) + if isinstance(call.func, ast.Attribute) + else None + ) + if ( + token is None + and endpoint.surface is not None + and endpoint.surface.registration_symbol + in { + "fastapi.FastAPI", + "starlette.applications.Starlette", + } + ): + token = (module.name, call.lineno, call.col_offset) + if token is not None: + self._framework_events.append(_FrameworkRegistrationEvent(token, endpoint)) + def _inspect_registration( # noqa: PLR0912, PLR0915 self, module: _Module, @@ -2144,6 +2476,7 @@ def _inspect_registration( # noqa: PLR0912, PLR0915 ) return symbol, invocation, receiver_type = resolved + self._record_framework_include(module, call, state, evaluation) for contract in self.contracts.document.contracts: if not self._matches(contract, symbol, invocation, receiver_type): continue @@ -2177,6 +2510,12 @@ def _inspect_registration( # noqa: PLR0912, PLR0915 handler_expression = call.keywords[keyword_index].value if evaluation is not None and keyword_index < len(evaluation.keywords): capture = evaluation.keywords[keyword_index] + if ( + contract.handler_optional + and isinstance(handler_expression, ast.Constant) + and handler_expression.value is None + ): + continue if handler_expression is not None and capture is not None: if contract.handler.kind == HandlerSelectorKind.ARGUMENT_CLASS_METHOD: handler_result = self._resolve_captured_class_method( @@ -2286,26 +2625,26 @@ def _inspect_registration( # noqa: PLR0912, PLR0915 contract_hash=self.contracts.contract_hashes[contract.id], conditions=contract.conditions, ) - self._endpoints.append( - Endpoint( - path=surface_id, - methods=[EndpointMethod.CUSTOM], - handler=HandlerInfo( - name=function.name, - module=handler_module.name, - file_path=handler_module.path, - line_number=handler_range[0], - end_line_number=handler_range[1], - ), - discovery_status=( - EndpointDiscoveryStatus.CONDITIONAL - if merged - else EndpointDiscoveryStatus.ESTABLISHED - ), - discovery_conditions=merged, - surface=evidence, - ) + endpoint = Endpoint( + path=surface_id, + methods=[EndpointMethod.CUSTOM], + handler=HandlerInfo( + name=function.name, + module=handler_module.name, + file_path=handler_module.path, + line_number=handler_range[0], + end_line_number=handler_range[1], + ), + discovery_status=( + EndpointDiscoveryStatus.CONDITIONAL + if merged + else EndpointDiscoveryStatus.ESTABLISHED + ), + discovery_conditions=merged, + surface=evidence, ) + self._endpoints.append(endpoint) + self._record_framework_registration(module, call, state, evaluation, endpoint) if contract.activates_routes and resources == ("startup",): self._emit_startup_routes( contract, diff --git a/src/fastapi_endpoint_detector/presets/framework_v1.yaml b/src/fastapi_endpoint_detector/presets/framework_v1.yaml index b3636a5..6d2c421 100644 --- a/src/fastapi_endpoint_detector/presets/framework_v1.yaml +++ b/src/fastapi_endpoint_detector/presets/framework_v1.yaml @@ -1,11 +1,11 @@ schema_version: 5 preset: id: framework-callbacks - version: "4" + version: "5" provenance: kind: preset source: fastapi-endpoint-detector - revision: "4" + revision: "5" contracts: - id: fastapi-lifespan-startup registration: @@ -64,6 +64,32 @@ contracts: execution_mode: framework activates_routes: true + - id: fastapi-router-on-event + registration: + symbol: fastapi.APIRouter.on_event + invocation: instance_method + receiver_type: fastapi.APIRouter + handler: {kind: decorated_function} + surface: + kind: framework.lifecycle + id_template: "event:{resource}" + resource: {kind: argument, index: 0} + callback_mode: either + execution_mode: framework + + - id: fastapi-router-add-event-handler + registration: + symbol: fastapi.APIRouter.add_event_handler + invocation: instance_method + receiver_type: fastapi.APIRouter + handler: {kind: argument, index: 1} + surface: + kind: framework.lifecycle + id_template: "event:{resource}" + resource: {kind: argument, index: 0} + callback_mode: either + execution_mode: framework + - id: starlette-on-event registration: symbol: starlette.applications.Starlette.on_event diff --git a/tests/unit/test_framework_surfaces.py b/tests/unit/test_framework_surfaces.py index bfc5221..c6d5be6 100644 --- a/tests/unit/test_framework_surfaces.py +++ b/tests/unit/test_framework_surfaces.py @@ -1,8 +1,10 @@ """Exact FastAPI and Starlette lifecycle/middleware surface contracts.""" +import asyncio from pathlib import Path import pytest +from fastapi import APIRouter, FastAPI from fastapi_endpoint_detector.config import AnalysisConfig, Config from fastapi_endpoint_detector.models.endpoint import ( @@ -20,10 +22,11 @@ ) -def _extract(tmp_path: Path) -> EndpointInventory: +def _extract(tmp_path: Path, *, app_entry: str | None = None) -> EndpointInventory: return CustomSurfaceExtractor( tmp_path, load_surface_preset("framework-v1"), + app_entry=app_entry, ).extract_inventory() @@ -56,6 +59,27 @@ def test_fastapi_lifespan_splits_exact_pre_and_post_yield_ranges(tmp_path: Path) assert shutdown.surface.callback_range.value == "after_yield" +def test_module_qualified_lifespan_selected_and_literal_none_is_absent(tmp_path: Path) -> None: + (tmp_path / "main.py").write_text( + "from contextlib import asynccontextmanager\n" + "import fastapi\n\n" + "@asynccontextmanager\n" + "async def lifespan(app):\n" + " yield\n\n" + "unused = fastapi.FastAPI(lifespan=None)\n" + "app = fastapi.FastAPI(lifespan=lifespan)\n", + encoding="utf-8", + ) + + inventory = _extract(tmp_path) + + assert [endpoint.identifier for endpoint in inventory.endpoints] == [ + "FRAMEWORK.LIFECYCLE lifespan:shutdown", + "FRAMEWORK.LIFECYCLE lifespan:startup", + ] + assert inventory.status == InventoryStatus.ESTABLISHED + + def test_lifespan_with_conditional_yield_fails_closed(tmp_path: Path) -> None: (tmp_path / "main.py").write_text( "from contextlib import asynccontextmanager\n" @@ -117,6 +141,134 @@ def test_starlette_imperative_lifecycle_callbacks_resolve_exact_handlers( assert [endpoint.handler.name for endpoint in inventory.endpoints] == ["stop", "start"] +def test_runtime_oracle_mount_excludes_child_lifespan_and_router_include_copies() -> None: + calls: list[str] = [] + child = FastAPI() + + @child.on_event("startup") + async def child_startup() -> None: + calls.append("child") + + parent = FastAPI() + parent.mount("/child", child) + + router = APIRouter() + + @router.on_event("startup") + async def copied() -> None: + calls.append("copied") + + parent.include_router(router) + + @router.on_event("startup") + async def too_late() -> None: + calls.append("too-late") + + async def enter_lifespan() -> None: + async with parent.router.lifespan_context(parent): + pass + + asyncio.run(enter_lifespan()) + + # Mounted applications do not contribute child lifespan execution. Router + # lifecycle behavior after inclusion differs across supported FastAPI versions, + # so the static adapter treats that later registration as conditional. + assert "child" not in calls + assert "copied" in calls + + +def test_framework_surfaces_are_scoped_to_selected_app_not_mounted_lifespan( + tmp_path: Path, +) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n\n" + "child = FastAPI()\n" + "@child.on_event('startup')\n" + "async def child_startup(): pass\n\n" + "app = FastAPI()\n" + "@app.on_event('startup')\n" + "async def parent_startup(): pass\n" + "app.mount('/child', child)\n", + encoding="utf-8", + ) + + inventory = _extract(tmp_path) + + assert [endpoint.handler.name for endpoint in inventory.endpoints] == ["parent_startup"] + assert inventory.status == InventoryStatus.ESTABLISHED + + +def test_router_lifecycle_uses_copy_at_include_order(tmp_path: Path) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import APIRouter, FastAPI\n\n" + "router = APIRouter()\n" + "@router.on_event('startup')\n" + "async def copied(): pass\n\n" + "app = FastAPI()\n" + "app.include_router(router=router)\n\n" + "@router.on_event('shutdown')\n" + "async def too_late(): pass\n", + encoding="utf-8", + ) + + inventory = _extract(tmp_path) + + assert [endpoint.handler.name for endpoint in inventory.endpoints] == ["copied"] + assert inventory.endpoints[0].identifier == "FRAMEWORK.LIFECYCLE event:startup" + assert inventory.status == InventoryStatus.CONDITIONAL + assert any("runtime-version-dependent" in item.reason for item in inventory.limitations) + + +def test_rebound_default_app_does_not_leave_stale_framework_surface(tmp_path: Path) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n\n" + "app = FastAPI()\n" + "@app.on_event('startup')\n" + "async def stale(): pass\n" + "app = build_app()\n", + encoding="utf-8", + ) + + inventory = _extract(tmp_path) + + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.CONDITIONAL + assert any("unresolved or was rebound" in item.reason for item in inventory.limitations) + + +def test_explicit_factory_root_selects_only_returned_app_surfaces(tmp_path: Path) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n\n" + "unused = FastAPI()\n" + "@unused.on_event('startup')\n" + "async def unused_startup(): pass\n\n" + "def create_app():\n" + " selected = FastAPI()\n" + " @selected.on_event('startup')\n" + " async def selected_startup(): pass\n" + " return selected\n", + encoding="utf-8", + ) + + inventory = _extract(tmp_path, app_entry="main:create_app") + + assert [endpoint.handler.name for endpoint in inventory.endpoints] == ["selected_startup"] + assert inventory.status == InventoryStatus.ESTABLISHED + + +def test_dynamic_include_on_selected_app_fails_closed(tmp_path: Path) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import FastAPI\n\napp = FastAPI()\napp.include_router(build_router())\n", + encoding="utf-8", + ) + + inventory = _extract(tmp_path) + + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.CONDITIONAL + assert any("include_router target is dynamic" in item.reason for item in inventory.limitations) + + def test_fastapi_http_middleware_is_exact_async_surface(tmp_path: Path) -> None: (tmp_path / "main.py").write_text( "from fastapi import FastAPI\n\n" @@ -381,10 +533,8 @@ def test_startup_route_receiver_rebinding_fails_closed(tmp_path: Path) -> None: inventory = _extract(tmp_path) - assert [endpoint.identifier for endpoint in inventory.endpoints] == [ - "FRAMEWORK.LIFECYCLE event:startup" - ] - assert any("receiver was rebound" in item.reason for item in inventory.limitations) + assert inventory.endpoints == [] + assert any("unresolved or was rebound" in item.reason for item in inventory.limitations) def test_lifespan_adds_pre_yield_route_but_not_shutdown_route(tmp_path: Path) -> None: From 92add6291e7e80590e423c2d9500f5561e735a87 Mon Sep 17 00:00:00 2001 From: shaggitza Date: Thu, 30 Jul 2026 00:54:44 +0300 Subject: [PATCH 2/4] fix: propagate framework include limitations --- .../parser/custom_surface_extractor.py | 41 +++++++++++++----- tests/unit/test_framework_surfaces.py | 42 +++++++++++++++++++ 2 files changed, 72 insertions(+), 11 deletions(-) diff --git a/src/fastapi_endpoint_detector/parser/custom_surface_extractor.py b/src/fastapi_endpoint_detector/parser/custom_surface_extractor.py index dcf272a..147ca16 100644 --- a/src/fastapi_endpoint_detector/parser/custom_surface_extractor.py +++ b/src/fastapi_endpoint_detector/parser/custom_surface_extractor.py @@ -1050,6 +1050,24 @@ def _is_framework_endpoint(endpoint: Endpoint) -> bool: "framework." ) + @staticmethod + def _framework_include_ancestors( + token: _FrameworkToken, + included_by: dict[_FrameworkToken, set[_FrameworkToken]], + ) -> tuple[_FrameworkToken, ...]: + """Return prior include ancestors in deterministic, graph-bounded order.""" + ancestors: list[_FrameworkToken] = [] + seen = {token} + pending = sorted(included_by.get(token, ()), reverse=True) + while pending: + ancestor = pending.pop() + if ancestor in seen: + continue + seen.add(ancestor) + ancestors.append(ancestor) + pending.extend(sorted(included_by.get(ancestor, ()), reverse=True)) + return tuple(ancestors) + def _filter_framework_surfaces(self) -> None: """Apply selected-app identity and APIRouter copy-at-include semantics.""" if not self._scope_framework_surfaces: @@ -1062,21 +1080,22 @@ def _filter_framework_surfaces(self) -> None: live.setdefault(event.token, []).append(event.endpoint) surface = event.endpoint.surface if surface is not None: - for parent in included_by.get(event.token, ()): - conditions.setdefault(parent, []).append( - EndpointDiscoveryCondition( - source_path=surface.registration_file, - source_line=surface.registration_line, - reason=( - "router lifecycle registered after include_router has " - "runtime-version-dependent execution" - ), - ) - ) + condition = EndpointDiscoveryCondition( + source_path=surface.registration_file, + source_line=surface.registration_line, + reason=( + "router lifecycle registered after include_router has " + "runtime-version-dependent execution" + ), + ) + for ancestor in self._framework_include_ancestors(event.token, included_by): + conditions.setdefault(ancestor, []).append(condition) continue if event.child is None: if event.condition is not None: conditions.setdefault(event.parent, []).append(event.condition) + for ancestor in self._framework_include_ancestors(event.parent, included_by): + conditions.setdefault(ancestor, []).append(event.condition) continue live.setdefault(event.parent, []).extend(live.get(event.child, ())) conditions.setdefault(event.parent, []).extend(conditions.get(event.child, ())) diff --git a/tests/unit/test_framework_surfaces.py b/tests/unit/test_framework_surfaces.py index c6d5be6..799c746 100644 --- a/tests/unit/test_framework_surfaces.py +++ b/tests/unit/test_framework_surfaces.py @@ -219,6 +219,31 @@ def test_router_lifecycle_uses_copy_at_include_order(tmp_path: Path) -> None: assert any("runtime-version-dependent" in item.reason for item in inventory.limitations) +def test_nested_router_lifecycle_late_registration_reaches_selected_app( + tmp_path: Path, +) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import APIRouter, FastAPI\n\n" + "child = APIRouter()\n" + "@child.on_event('startup')\n" + "async def copied(): pass\n\n" + "parent = APIRouter()\n" + "parent.include_router(child)\n" + "app = FastAPI()\n" + "app.include_router(parent)\n\n" + "@child.on_event('shutdown')\n" + "async def too_late(): pass\n", + encoding="utf-8", + ) + + inventory = _extract(tmp_path) + + assert [endpoint.handler.name for endpoint in inventory.endpoints] == ["copied"] + assert inventory.endpoints[0].identifier == "FRAMEWORK.LIFECYCLE event:startup" + assert inventory.status == InventoryStatus.CONDITIONAL + assert any("runtime-version-dependent" in item.reason for item in inventory.limitations) + + def test_rebound_default_app_does_not_leave_stale_framework_surface(tmp_path: Path) -> None: (tmp_path / "main.py").write_text( "from fastapi import FastAPI\n\n" @@ -269,6 +294,23 @@ def test_dynamic_include_on_selected_app_fails_closed(tmp_path: Path) -> None: assert any("include_router target is dynamic" in item.reason for item in inventory.limitations) +def test_nested_dynamic_include_reaches_already_included_selected_app(tmp_path: Path) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import APIRouter, FastAPI\n\n" + "parent = APIRouter()\n" + "app = FastAPI()\n" + "app.include_router(parent)\n" + "parent.include_router(build_router())\n", + encoding="utf-8", + ) + + inventory = _extract(tmp_path) + + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.CONDITIONAL + assert any("inventory is incomplete" in item.reason for item in inventory.limitations) + + def test_fastapi_http_middleware_is_exact_async_surface(tmp_path: Path) -> None: (tmp_path / "main.py").write_text( "from fastapi import FastAPI\n\n" From d96e6f2a14b8fae8f524d3f7e3ff58416a7f5e25 Mon Sep 17 00:00:00 2001 From: shaggitza Date: Thu, 30 Jul 2026 01:03:22 +0300 Subject: [PATCH 3/4] fix: propagate late resolved router includes --- .../parser/custom_surface_extractor.py | 34 +++++- tests/unit/test_framework_surfaces.py | 100 ++++++++++++++++++ 2 files changed, 132 insertions(+), 2 deletions(-) diff --git a/src/fastapi_endpoint_detector/parser/custom_surface_extractor.py b/src/fastapi_endpoint_detector/parser/custom_surface_extractor.py index 147ca16..703d60a 100644 --- a/src/fastapi_endpoint_detector/parser/custom_surface_extractor.py +++ b/src/fastapi_endpoint_detector/parser/custom_surface_extractor.py @@ -1068,6 +1068,28 @@ def _framework_include_ancestors( pending.extend(sorted(included_by.get(ancestor, ()), reverse=True)) return tuple(ancestors) + @staticmethod + def _framework_copied_lifecycle_conditions( + endpoints: tuple[Endpoint, ...], + ) -> tuple[EndpointDiscoveryCondition, ...]: + """Describe lifecycle copies whose ancestor execution varies by runtime.""" + conditions: list[EndpointDiscoveryCondition] = [] + for endpoint in endpoints: + surface = endpoint.surface + if surface is None or surface.surface_kind != "framework.lifecycle": + continue + conditions.append( + EndpointDiscoveryCondition( + source_path=surface.registration_file, + source_line=surface.registration_line, + reason=( + "router lifecycle copied into an already-included router has " + "runtime-version-dependent execution" + ), + ) + ) + return tuple(conditions) + def _filter_framework_surfaces(self) -> None: """Apply selected-app identity and APIRouter copy-at-include semantics.""" if not self._scope_framework_surfaces: @@ -1097,8 +1119,16 @@ def _filter_framework_surfaces(self) -> None: for ancestor in self._framework_include_ancestors(event.parent, included_by): conditions.setdefault(ancestor, []).append(event.condition) continue - live.setdefault(event.parent, []).extend(live.get(event.child, ())) - conditions.setdefault(event.parent, []).extend(conditions.get(event.child, ())) + copied_endpoints = tuple(live.get(event.child, ())) + copied_conditions = tuple(conditions.get(event.child, ())) + live.setdefault(event.parent, []).extend(copied_endpoints) + conditions.setdefault(event.parent, []).extend(copied_conditions) + copied_lifecycle_conditions = self._framework_copied_lifecycle_conditions( + copied_endpoints + ) + for ancestor in self._framework_include_ancestors(event.parent, included_by): + conditions.setdefault(ancestor, []).extend(copied_conditions) + conditions[ancestor].extend(copied_lifecycle_conditions) included_by.setdefault(event.child, set()).add(event.parent) accepted = { diff --git a/tests/unit/test_framework_surfaces.py b/tests/unit/test_framework_surfaces.py index 799c746..3a327d4 100644 --- a/tests/unit/test_framework_surfaces.py +++ b/tests/unit/test_framework_surfaces.py @@ -1,6 +1,7 @@ """Exact FastAPI and Starlette lifecycle/middleware surface contracts.""" import asyncio +from importlib.metadata import version from pathlib import Path import pytest @@ -177,6 +178,64 @@ async def enter_lifespan() -> None: assert "copied" in calls +def test_runtime_oracle_late_nested_router_include_is_version_dependent() -> None: + calls: list[str] = [] + child = APIRouter() + parent = APIRouter() + app = FastAPI() + app.include_router(parent) + + @child.on_event("startup") + async def child_startup() -> None: + calls.append("child") + + parent.include_router(child) + + async def enter_lifespan() -> None: + async with app.router.lifespan_context(app): + pass + + asyncio.run(enter_lifespan()) + + expected_calls = { + "0.100.0": [], + "0.139.0": ["child"], + } + installed_version = version("fastapi") + assert installed_version in expected_calls + assert calls == expected_calls[installed_version] + + +def test_runtime_oracle_late_nested_known_leaf_is_version_dependent() -> None: + calls: list[str] = [] + leaf = APIRouter() + + @leaf.on_event("startup") + async def leaf_startup() -> None: + calls.append("leaf") + + child = APIRouter() + parent = APIRouter() + app = FastAPI() + app.include_router(parent) + child.include_router(leaf) + parent.include_router(child) + + async def enter_lifespan() -> None: + async with app.router.lifespan_context(app): + pass + + asyncio.run(enter_lifespan()) + + expected_calls = { + "0.100.0": [], + "0.139.0": ["leaf"], + } + installed_version = version("fastapi") + assert installed_version in expected_calls + assert calls == expected_calls[installed_version] + + def test_framework_surfaces_are_scoped_to_selected_app_not_mounted_lifespan( tmp_path: Path, ) -> None: @@ -244,6 +303,28 @@ def test_nested_router_lifecycle_late_registration_reaches_selected_app( assert any("runtime-version-dependent" in item.reason for item in inventory.limitations) +def test_late_nested_router_include_reaches_every_existing_ancestor(tmp_path: Path) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import APIRouter, FastAPI\n\n" + "child = APIRouter()\n" + "parent = APIRouter()\n" + "root = APIRouter()\n" + "root.include_router(parent)\n" + "app = FastAPI()\n" + "app.include_router(root)\n\n" + "@child.on_event('startup')\n" + "async def version_dependent(): pass\n" + "parent.include_router(child)\n", + encoding="utf-8", + ) + + inventory = _extract(tmp_path) + + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.CONDITIONAL + assert any("runtime-version-dependent" in item.reason for item in inventory.limitations) + + def test_rebound_default_app_does_not_leave_stale_framework_surface(tmp_path: Path) -> None: (tmp_path / "main.py").write_text( "from fastapi import FastAPI\n\n" @@ -311,6 +392,25 @@ def test_nested_dynamic_include_reaches_already_included_selected_app(tmp_path: assert any("inventory is incomplete" in item.reason for item in inventory.limitations) +def test_late_resolved_nested_include_propagates_child_limitation(tmp_path: Path) -> None: + (tmp_path / "main.py").write_text( + "from fastapi import APIRouter, FastAPI\n\n" + "child = APIRouter()\n" + "parent = APIRouter()\n" + "app = FastAPI()\n" + "app.include_router(parent)\n" + "child.include_router(build_router())\n" + "parent.include_router(child)\n", + encoding="utf-8", + ) + + inventory = _extract(tmp_path) + + assert inventory.endpoints == [] + assert inventory.status == InventoryStatus.CONDITIONAL + assert any("inventory is incomplete" in item.reason for item in inventory.limitations) + + def test_fastapi_http_middleware_is_exact_async_surface(tmp_path: Path) -> None: (tmp_path / "main.py").write_text( "from fastapi import FastAPI\n\n" From eccd3e36a676998714d9129d91127d3a64eb4aac Mon Sep 17 00:00:00 2001 From: shaggitza Date: Thu, 30 Jul 2026 01:46:53 +0300 Subject: [PATCH 4/4] test: cover supported FastAPI lifecycle ranges --- tests/unit/test_framework_surfaces.py | 29 ++++++++++++++------------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/tests/unit/test_framework_surfaces.py b/tests/unit/test_framework_surfaces.py index 3a327d4..b104259 100644 --- a/tests/unit/test_framework_surfaces.py +++ b/tests/unit/test_framework_surfaces.py @@ -1,6 +1,7 @@ """Exact FastAPI and Starlette lifecycle/middleware surface contracts.""" import asyncio +import re from importlib.metadata import version from pathlib import Path @@ -31,6 +32,18 @@ def _extract(tmp_path: Path, *, app_entry: str | None = None) -> EndpointInvento ).extract_inventory() +def _expected_late_nested_calls(callback: str) -> list[str]: + installed_version = version("fastapi") + release = re.match(r"^(\d+)\.(\d+)", installed_version) + assert release is not None + major_minor = (int(release.group(1)), int(release.group(2))) + if major_minor <= (0, 100): + return [] + if major_minor >= (0, 139): + return [callback] + pytest.skip(f"nested lifecycle copy behavior is not calibrated for FastAPI {installed_version}") + + def test_fastapi_lifespan_splits_exact_pre_and_post_yield_ranges(tmp_path: Path) -> None: (tmp_path / "main.py").write_text( "from contextlib import asynccontextmanager\n" @@ -197,13 +210,7 @@ async def enter_lifespan() -> None: asyncio.run(enter_lifespan()) - expected_calls = { - "0.100.0": [], - "0.139.0": ["child"], - } - installed_version = version("fastapi") - assert installed_version in expected_calls - assert calls == expected_calls[installed_version] + assert calls == _expected_late_nested_calls("child") def test_runtime_oracle_late_nested_known_leaf_is_version_dependent() -> None: @@ -227,13 +234,7 @@ async def enter_lifespan() -> None: asyncio.run(enter_lifespan()) - expected_calls = { - "0.100.0": [], - "0.139.0": ["leaf"], - } - installed_version = version("fastapi") - assert installed_version in expected_calls - assert calls == expected_calls[installed_version] + assert calls == _expected_late_nested_calls("leaf") def test_framework_surfaces_are_scoped_to_selected_app_not_mounted_lifespan(