From 839f616123a97f1a65c25753d7243d9a671293f4 Mon Sep 17 00:00:00 2001 From: Shaggi Date: Wed, 29 Jul 2026 19:09:39 +0300 Subject: [PATCH] feat: preserve runtime dependency provenance --- README.md | 8 + .../analyzer/mypy_analyzer.py | 104 ++- .../models/__init__.py | 20 + .../models/endpoint.py | 199 ++++++ .../output/json_output.py | 8 +- .../output/yaml_output.py | 8 +- .../parser/fastapi_extractor.py | 509 ++++++++++++- .../parser/runtime_worker.py | 12 +- tests/integration/test_di_patterns.py | 3 + tests/unit/test_fastapi_extractor.py | 672 ++++++++++++++++++ tests/unit/test_formatters.py | 26 +- tests/unit/test_models.py | 105 +++ tests/unit/test_mypy_correctness.py | 224 +++++- 13 files changed, 1881 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index aa2778c..d61b3bb 100644 --- a/README.md +++ b/README.md @@ -101,6 +101,14 @@ fastapi-endpoint-detector list --app path/to/main.py --format markdown fastapi-endpoint-detector list --app path/to/main.py --format html -o endpoints.html ``` +### Machine-readable endpoint provenance + +JSON and YAML inventories and analysis reports use schema version 3. Each endpoint may +include the additive, nullable `dependency_graph` field: schema-v1 bounded evidence for +the declared FastAPI dependency tree, with explicit status and limitations. Trusted +runtime extraction transports this evidence through the private protocol-v2 worker; +consumers should use the public inventory/report schema rather than that worker protocol. + ## Commands ### `analyze` - Analyze Code Changes diff --git a/src/fastapi_endpoint_detector/analyzer/mypy_analyzer.py b/src/fastapi_endpoint_detector/analyzer/mypy_analyzer.py index 05e5e6b..f1d0ace 100644 --- a/src/fastapi_endpoint_detector/analyzer/mypy_analyzer.py +++ b/src/fastapi_endpoint_detector/analyzer/mypy_analyzer.py @@ -34,7 +34,11 @@ ResolvedCallSite, ResourceIdentityEvidence, ) -from fastapi_endpoint_detector.models.endpoint import Endpoint +from fastapi_endpoint_detector.models.endpoint import ( + DependencyCallableKind, + DependencyResolutionStatus, + Endpoint, +) from fastapi_endpoint_detector.models.surface_contract import CallbackRangeMode # Type alias for line-level progress callback (file_path, line_number, symbol_name) @@ -372,7 +376,7 @@ class MypyAnalyzer: and extract precise file/line information for all references. """ - CACHE_SCHEMA_VERSION = 16 + CACHE_SCHEMA_VERSION = 18 MAX_POINTS_TO_TARGETS = 8 MAX_FACTORY_RETURNS = 64 MAX_FACTORY_STATES = 512 @@ -957,10 +961,74 @@ def injected_type(expression: ast.expr | None, seen: set[str]) -> str | None: self._python_dependency_cache[cache_key] = set(found) return found + def _runtime_dependency_seeds(self, endpoint: Endpoint) -> dict[str, int]: + """Return only uniquely source-attested project-local runtime graph seeds.""" + graph = endpoint.dependency_graph + if graph is None: + return {} + seeds: dict[str, int] = {} + canonical_root = self.source_root.resolve() + for occurrence in graph.occurrences: + if ( + occurrence.resolution_status != DependencyResolutionStatus.ESTABLISHED + or occurrence.callable_kind + not in {DependencyCallableKind.FUNCTION, DependencyCallableKind.BOUND_METHOD} + or occurrence.module is None + or occurrence.qualname is None + or occurrence.source_span is None + or occurrence.display_name == "" + or "" in occurrence.qualname + ): + continue + fullname = f"{occurrence.module}.{occurrence.qualname}" + resolved = self._resolve_fullname_to_file(fullname) + if resolved is None: + continue + definition_path, definition_module = resolved + if definition_module != occurrence.module and not definition_module.endswith( + f".{occurrence.module}" + ): + continue + try: + runtime_path = occurrence.source_span.file_path.resolve() + mypy_path = Path(definition_path).resolve() + runtime_path.relative_to(canonical_root) + mypy_path.relative_to(canonical_root) + except (OSError, ValueError): + continue + if runtime_path != mypy_path: + continue + dependency_tree = self._trees.get(definition_module) + if dependency_tree is None: + continue + symbol_name = occurrence.qualname.rsplit(".", maxsplit=1)[-1] + result = self._find_func_in_tree( + dependency_tree, + symbol_name, + qualified_name=occurrence.qualname, + ) + if result is None or result[1] != occurrence.qualname: + continue + definition_node = getattr(result[0], "func", result[0]) + definition_start, definition_end = self._get_func_lines(definition_node) + if ( + occurrence.source_span.end_line < definition_start + or occurrence.source_span.start_line > definition_end + ): + continue + canonical_fullname = f"{definition_module}.{result[1]}" + previous = seeds.get(canonical_fullname) + if previous is None or occurrence.depth < previous: + seeds[canonical_fullname] = occurrence.depth + return seeds + def _python_dependency_closure(self, endpoint: Endpoint) -> dict[str, int]: - """Expand explicit FastAPI dependency annotations to the configured depth.""" + """Expand explicit and source-attested runtime dependencies to bounded depth.""" depths: dict[str, int] = {} - queue = [(fullname, 1) for fullname in self._python_dependency_fullnames(endpoint)] + initial = dict.fromkeys(self._python_dependency_fullnames(endpoint), 1) + for fullname, depth in self._runtime_dependency_seeds(endpoint).items(): + initial[fullname] = min(initial.get(fullname, depth), depth) + queue = list(initial.items()) while queue: fullname, depth = queue.pop(0) previous = depths.get(fullname) @@ -976,8 +1044,17 @@ def _python_dependency_closure(self, endpoint: Endpoint) -> dict[str, int]: dependency_tree = self._trees.get(dependency_module) if dependency_tree is None: continue - symbol_name = fullname.rsplit(".", maxsplit=1)[-1] - dependency_result = self._find_func_in_tree(dependency_tree, symbol_name) + qualified_name = ( + fullname[len(dependency_module) + 1 :] + if fullname.startswith(f"{dependency_module}.") + else fullname.rsplit(".", maxsplit=1)[-1] + ) + symbol_name = qualified_name.rsplit(".", maxsplit=1)[-1] + dependency_result = self._find_func_in_tree( + dependency_tree, + symbol_name, + qualified_name=qualified_name, + ) if dependency_result is None: continue node, _qualified = dependency_result @@ -1002,8 +1079,20 @@ def _python_dependency_closure(self, endpoint: Endpoint) -> dict[str, int]: @staticmethod def _endpoint_key(endpoint: Endpoint) -> str: - """Key dependency data by public route and physical handler identity.""" + """Key dependency data by route, handler, and authoritative runtime graph.""" handler = endpoint.handler + graph_payload = ( + None + if endpoint.dependency_graph is None + else endpoint.dependency_graph.model_dump(mode="json") + ) + graph_hash = hashlib.sha256( + json.dumps( + graph_payload, + sort_keys=True, + separators=(",", ":"), + ).encode() + ).hexdigest() return json.dumps( [ endpoint.identifier, @@ -1011,6 +1100,7 @@ def _endpoint_key(endpoint: Endpoint) -> str: handler.line_number, handler.name, handler.module, + graph_hash, ], separators=(",", ":"), ) diff --git a/src/fastapi_endpoint_detector/models/__init__.py b/src/fastapi_endpoint_detector/models/__init__.py index 8b6b3e4..980e2c8 100644 --- a/src/fastapi_endpoint_detector/models/__init__.py +++ b/src/fastapi_endpoint_detector/models/__init__.py @@ -43,7 +43,17 @@ EffectContractCoverage, ) from fastapi_endpoint_detector.models.endpoint import ( + DependencyCallableKind, + DependencyCallableStructure, + DependencyDeclarationKind, + DependencyDeclarationScope, + DependencyGraphLimitation, + DependencyGraphStatus, + DependencyResolutionStatus, + DependencySourceSpan, Endpoint, + EndpointDependencyGraph, + EndpointDependencyOccurrence, EndpointDiscoveryCondition, EndpointDiscoveryStatus, EndpointInventory, @@ -104,7 +114,17 @@ __all__ = [ # noqa: RUF022 - grouped by public model domain # Endpoint models + "DependencyCallableKind", + "DependencyCallableStructure", + "DependencyDeclarationKind", + "DependencyDeclarationScope", + "DependencyGraphLimitation", + "DependencyGraphStatus", + "DependencyResolutionStatus", + "DependencySourceSpan", "Endpoint", + "EndpointDependencyGraph", + "EndpointDependencyOccurrence", "EndpointDiscoveryCondition", "EndpointDiscoveryStatus", "EndpointInventory", diff --git a/src/fastapi_endpoint_detector/models/endpoint.py b/src/fastapi_endpoint_detector/models/endpoint.py index 7e29453..857f61b 100644 --- a/src/fastapi_endpoint_detector/models/endpoint.py +++ b/src/fastapi_endpoint_detector/models/endpoint.py @@ -48,6 +48,201 @@ class EndpointDiscoveryStatus(str, Enum): CONDITIONAL = "conditional" +class DependencyGraphStatus(str, Enum): + """Completeness of runtime dependency-graph evidence.""" + + ESTABLISHED = "established" + CONDITIONAL = "conditional" + UNAVAILABLE = "unavailable" + + +class DependencyDeclarationScope(str, Enum): + """Best runtime-attested declaration scope for one dependency occurrence.""" + + ASSEMBLY = "assembly" + PARAMETER = "parameter" + NESTED = "nested" + UNKNOWN = "unknown" + + +class DependencyDeclarationKind(str, Enum): + """FastAPI declaration construct, without guessing ambiguous empty Security scopes.""" + + DEPENDS = "depends" + SECURITY = "security" + DEPENDS_OR_SECURITY = "depends_or_security" + UNKNOWN = "unknown" + + +class DependencyCallableKind(str, Enum): + """Stable structural category of a dependency callable.""" + + FUNCTION = "function" + BOUND_METHOD = "bound_method" + PARTIAL = "partial" + CALLABLE_INSTANCE = "callable_instance" + UNKNOWN = "unknown" + + +class DependencyResolutionStatus(str, Enum): + """Strength of the identity recorded for one dependency occurrence.""" + + ESTABLISHED = "established" + CONDITIONAL = "conditional" + UNAVAILABLE = "unavailable" + + +class DependencySourceSpan(BaseModel): + """Runtime-inspected source span for a dependency callable.""" + + file_path: Path + start_line: int = Field(ge=1) + end_line: int = Field(ge=1) + + @model_validator(mode="after") + def validate_lines(self) -> "DependencySourceSpan": + if self.end_line < self.start_line: + raise ValueError("dependency source span end must not precede start") + return self + + class Config: + frozen = True + + +class DependencyCallableStructure(BaseModel): + """One bounded, address-free layer in a structured callable.""" + + kind: DependencyCallableKind + module: str | None = Field(default=None, min_length=1, max_length=512) + qualname: str | None = Field(default=None, min_length=1, max_length=1024) + bound_positional_count: int = Field(default=0, ge=0, le=1024) + bound_keyword_names: tuple[str, ...] = Field(default=(), max_length=128) + + @model_validator(mode="after") + def validate_identity(self) -> "DependencyCallableStructure": + if (self.module is None) != (self.qualname is None): + raise ValueError("dependency callable module and qualname must be provided together") + if any(not name or len(name) > 256 for name in self.bound_keyword_names): + raise ValueError("dependency bound keyword names must be nonblank and bounded") + return self + + class Config: + frozen = True + + +class EndpointDependencyOccurrence(BaseModel): + """One ordered occurrence in FastAPI's declared dependency graph.""" + + index_path: tuple[int, ...] = Field(min_length=1, max_length=64) + parent_path: tuple[int, ...] = Field(max_length=63) + depth: int = Field(ge=1, le=64) + order: int = Field(ge=0) + declaration_scope: DependencyDeclarationScope + declaration_kind: DependencyDeclarationKind + callable_kind: DependencyCallableKind + resolution_status: DependencyResolutionStatus + display_name: str = Field(min_length=1, max_length=512) + module: str | None = Field(default=None, min_length=1, max_length=512) + qualname: str | None = Field(default=None, min_length=1, max_length=1024) + source_span: DependencySourceSpan | None = None + security_scopes: tuple[str, ...] = Field(default=(), max_length=256) + use_cache: bool | None = None + callable_structure: tuple[DependencyCallableStructure, ...] = Field(default=(), max_length=8) + + @model_validator(mode="after") + def validate_occurrence(self) -> "EndpointDependencyOccurrence": + if len(self.index_path) != self.depth or self.parent_path != self.index_path[:-1]: + raise ValueError("dependency index path must agree with parent path and depth") + if (self.module is None) != (self.qualname is None): + raise ValueError("dependency module and qualname must be provided together") + if self.resolution_status == DependencyResolutionStatus.ESTABLISHED and ( + self.module is None or self.source_span is None + ): + raise ValueError("established dependency identity requires qualified source evidence") + if any(not scope or len(scope) > 512 for scope in self.security_scopes): + raise ValueError("dependency security scopes must be nonblank and bounded") + return self + + class Config: + frozen = True + + +class DependencyGraphLimitation(BaseModel): + """Source-backed limitation scoped only to dependency graph collection.""" + + code: str = Field(pattern=r"^[a-z][a-z0-9_]*$", max_length=64) + source_path: Path + source_line: int = Field(ge=1) + reason: str = Field(min_length=1, max_length=2048) + + @field_validator("reason") + @classmethod + def reason_must_be_substantive(cls, value: str) -> str: + if not value.strip(): + raise ValueError("dependency graph limitation reason must not be blank") + return value + + class Config: + frozen = True + + +class EndpointDependencyGraph(BaseModel): + """Authoritative declared FastAPI dependency graph for one endpoint.""" + + schema_version: Literal[1] = 1 + status: DependencyGraphStatus + semantics: Literal["declared"] = "declared" + occurrences: tuple[EndpointDependencyOccurrence, ...] = Field(default=(), max_length=65536) + limitations: tuple[DependencyGraphLimitation, ...] = Field(default=(), max_length=65536) + + @model_validator(mode="after") + def validate_strength(self) -> "EndpointDependencyGraph": + if (self.status == DependencyGraphStatus.ESTABLISHED) == bool(self.limitations): + raise ValueError( + "established dependency graph forbids limitations; " + "conditional/unavailable require them" + ) + if self.status == DependencyGraphStatus.UNAVAILABLE and self.occurrences: + raise ValueError("unavailable dependency graph cannot contain occurrences") + expected_order = tuple(range(len(self.occurrences))) + if tuple(item.order for item in self.occurrences) != expected_order: + raise ValueError("dependency occurrence order must be contiguous and deterministic") + paths = [item.index_path for item in self.occurrences] + if len(set(paths)) != len(paths): + raise ValueError("dependency occurrence index paths must be unique") + if any(component < 0 or component > 65535 for path in paths for component in path): + raise ValueError("dependency occurrence index components must be bounded nonnegative") + if paths != sorted(paths): + raise ValueError("dependency occurrences must use deterministic depth-first preorder") + seen: set[tuple[int, ...]] = set() + next_sibling: dict[tuple[int, ...], int] = {} + for item in self.occurrences: + if item.parent_path and item.parent_path not in seen: + raise ValueError("dependency occurrence parent must exist earlier") + expected_index = next_sibling.get(item.parent_path, 0) + if item.index_path[-1] != expected_index: + raise ValueError("dependency root and sibling indexes must be contiguous") + next_sibling[item.parent_path] = expected_index + 1 + seen.add(item.index_path) + if self.status == DependencyGraphStatus.ESTABLISHED and any( + item.resolution_status != DependencyResolutionStatus.ESTABLISHED + for item in self.occurrences + ): + raise ValueError("established graph may contain only established occurrences") + if ( + any( + item.resolution_status != DependencyResolutionStatus.ESTABLISHED + for item in self.occurrences + ) + and not self.limitations + ): + raise ValueError("uncertain dependency occurrences require graph limitations") + return self + + class Config: + frozen = True + + class EndpointDiscoveryCondition(BaseModel): """Source-backed limitation on a conditionally discovered route.""" @@ -147,6 +342,10 @@ class Endpoint(BaseModel): default_factory=list, description="FastAPI Depends() dependencies (function names)", ) + dependency_graph: EndpointDependencyGraph | None = Field( + default=None, + description="Authoritative declared dependency graph; None means not collected", + ) discovery_status: EndpointDiscoveryStatus = EndpointDiscoveryStatus.ESTABLISHED discovery_conditions: tuple[EndpointDiscoveryCondition, ...] = () surface: SurfaceRegistrationEvidence | None = None diff --git a/src/fastapi_endpoint_detector/output/json_output.py b/src/fastapi_endpoint_detector/output/json_output.py index e5b2fea..0ce4973 100644 --- a/src/fastapi_endpoint_detector/output/json_output.py +++ b/src/fastapi_endpoint_detector/output/json_output.py @@ -40,6 +40,11 @@ def _endpoint_to_dict(self, endpoint: Endpoint) -> dict[str, Any]: "name": endpoint.name, "tags": endpoint.tags, "dependencies": endpoint.dependencies, + "dependency_graph": ( + endpoint.dependency_graph.model_dump(mode="json") + if endpoint.dependency_graph is not None + else None + ), "discovery_status": endpoint.discovery_status.value, "discovery_conditions": [ condition.model_dump(mode="json") for condition in endpoint.discovery_conditions @@ -83,6 +88,7 @@ def _affected_to_dict(self, affected: AffectedEndpoint) -> dict[str, Any]: def format(self, report: AnalysisReport) -> str: """Format an analysis report as JSON.""" data = { + "schema_version": 3, "timestamp": report.timestamp.isoformat(), "app_path": report.app_path, "diff_source": report.diff_source, @@ -148,7 +154,7 @@ def format(self, report: AnalysisReport) -> str: def format_inventory(self, inventory: EndpointInventory) -> str: """Format endpoints with whole-inventory strength metadata.""" data = { - "schema_version": 2, + "schema_version": 3, "inventory_status": inventory.status.value, "inventory_limitations": [ limitation.model_dump(mode="json") for limitation in inventory.limitations diff --git a/src/fastapi_endpoint_detector/output/yaml_output.py b/src/fastapi_endpoint_detector/output/yaml_output.py index 8a47dda..99e87bf 100644 --- a/src/fastapi_endpoint_detector/output/yaml_output.py +++ b/src/fastapi_endpoint_detector/output/yaml_output.py @@ -32,6 +32,11 @@ def _endpoint_to_dict(self, endpoint: Endpoint) -> dict[str, Any]: "name": endpoint.name, "tags": endpoint.tags, "dependencies": endpoint.dependencies, + "dependency_graph": ( + endpoint.dependency_graph.model_dump(mode="json") + if endpoint.dependency_graph is not None + else None + ), "discovery_status": endpoint.discovery_status.value, "discovery_conditions": [ condition.model_dump(mode="json") for condition in endpoint.discovery_conditions @@ -75,6 +80,7 @@ def _affected_to_dict(self, affected: AffectedEndpoint) -> dict[str, Any]: def format(self, report: AnalysisReport) -> str: """Format an analysis report as YAML.""" data = { + "schema_version": 3, "timestamp": report.timestamp.isoformat(), "app_path": report.app_path, "diff_source": report.diff_source, @@ -140,7 +146,7 @@ def format(self, report: AnalysisReport) -> str: def format_inventory(self, inventory: EndpointInventory) -> str: """Format endpoints with whole-inventory strength metadata.""" data = { - "schema_version": 2, + "schema_version": 3, "inventory_status": inventory.status.value, "inventory_limitations": [ limitation.model_dump(mode="json") for limitation in inventory.limitations diff --git a/src/fastapi_endpoint_detector/parser/fastapi_extractor.py b/src/fastapi_endpoint_detector/parser/fastapi_extractor.py index d887ed4..f3bcb62 100644 --- a/src/fastapi_endpoint_detector/parser/fastapi_extractor.py +++ b/src/fastapi_endpoint_detector/parser/fastapi_extractor.py @@ -6,6 +6,7 @@ AST parsing as it handles all FastAPI patterns automatically. """ +import functools import importlib.util import inspect import json @@ -25,7 +26,17 @@ from starlette.routing import Mount from fastapi_endpoint_detector.models.endpoint import ( + DependencyCallableKind, + DependencyCallableStructure, + DependencyDeclarationKind, + DependencyDeclarationScope, + DependencyGraphLimitation, + DependencyGraphStatus, + DependencyResolutionStatus, + DependencySourceSpan, Endpoint, + EndpointDependencyGraph, + EndpointDependencyOccurrence, EndpointMethod, HandlerInfo, ) @@ -53,6 +64,9 @@ def __init__( *, timeout_seconds: float = 60.0, output_limit_bytes: int = 4 * 1024 * 1024, + dependency_max_depth: int = 32, + dependency_max_nodes: int = 2048, + dependency_max_work: int = 8192, ) -> None: """ Initialize the extractor. @@ -64,6 +78,9 @@ def __init__( will be derived from app_path. timeout_seconds: Maximum time allowed for runtime import and extraction. output_limit_bytes: Maximum serialized worker response size. + dependency_max_depth: Maximum recursive dependency depth retained. + dependency_max_nodes: Maximum dependency occurrences retained per endpoint. + dependency_max_work: Maximum dependency traversal work units per endpoint. """ try: normalized_timeout = float(timeout_seconds) @@ -80,13 +97,30 @@ def __init__( isinstance(output_limit_bytes, bool) or not isinstance(output_limit_bytes, int) or output_limit_bytes <= 0 + or output_limit_bytes > 64 * 1024 * 1024 ): - raise ValueError("output_limit_bytes must be a positive integer") + raise ValueError("output_limit_bytes must be a positive integer not exceeding 67108864") self.app_path = app_path.resolve() self.app_variable = app_variable self.module_name = module_name self.timeout_seconds = normalized_timeout + for name, value in ( + ("dependency_max_depth", dependency_max_depth), + ("dependency_max_nodes", dependency_max_nodes), + ("dependency_max_work", dependency_max_work), + ): + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise ValueError(f"{name} must be a positive integer") + if dependency_max_depth > 64: + raise ValueError("dependency_max_depth must not exceed 64") + if dependency_max_nodes > 4096: + raise ValueError("dependency_max_nodes must not exceed 4096") + if dependency_max_work > 65536: + raise ValueError("dependency_max_work must not exceed 65536") self.output_limit_bytes = output_limit_bytes + self.dependency_max_depth = dependency_max_depth + self.dependency_max_nodes = dependency_max_nodes + self.dependency_max_work = dependency_max_work self._app: Any = None self._original_sys_path: list[str] = [] @@ -270,6 +304,463 @@ def _extract_dependencies(self, route: Any) -> list[str]: return dependencies + @staticmethod + def _physical_callable(value: Any) -> Any: + """Return the actual function implementing a callable without decorator unwrapping.""" + candidate = value.func if isinstance(value, functools.partial) else value + if inspect.ismethod(candidate): + return candidate.__func__ + if inspect.isfunction(candidate): + return candidate + if callable(candidate) and not inspect.isclass(candidate): + return type(candidate).__call__ + return candidate + + @classmethod + def _dependency_source_span(cls, value: Any) -> DependencySourceSpan | None: + """Return source coordinates for the physical callable, never ``__wrapped__``.""" + try: + candidate = cls._physical_callable(value) + source_lines, start_line = inspect.getsourcelines(candidate) + file_path = Path( + inspect.getsourcefile(candidate) or inspect.getfile(candidate) + ).resolve() + except Exception: + return None + if start_line < 1: + return None + return DependencySourceSpan( + file_path=file_path, + start_line=start_line, + end_line=start_line + len(source_lines) - 1, + ) + + @classmethod + def _callable_identity(cls, value: Any) -> tuple[str | None, str | None]: + """Derive physical identity from code/globals, not copyable wrapper metadata.""" + try: + candidate = cls._physical_callable(value) + if not inspect.isfunction(candidate): + return None, None + globals_dict = getattr(candidate, "__globals__", None) + code = getattr(candidate, "__code__", None) + module = globals_dict.get("__name__") if isinstance(globals_dict, dict) else None + qualname = getattr(code, "co_qualname", None) + if qualname is None: + # ``code.co_qualname`` was added in Python 3.11. On 3.10, + # accept ``__qualname__`` only when copied decorator metadata + # has not changed the physical code object's function name. + runtime_name = getattr(candidate, "__name__", None) + code_name = getattr(code, "co_name", None) + declared_qualname = getattr(candidate, "__qualname__", None) + if runtime_name == code_name: + qualname = declared_qualname + if not isinstance(module, str) or not module or len(module) > 512: + return None, None + if not isinstance(qualname, str) or not qualname or len(qualname) > 1024: + return None, None + except Exception: + return None, None + return module, qualname + + @staticmethod + def _callable_kind(value: Any) -> DependencyCallableKind: + if isinstance(value, functools.partial): + return DependencyCallableKind.PARTIAL + if inspect.ismethod(value) and value.__self__ is not None: + return DependencyCallableKind.BOUND_METHOD + if inspect.isfunction(value): + return DependencyCallableKind.FUNCTION + if callable(value) and not inspect.isclass(value): + return DependencyCallableKind.CALLABLE_INSTANCE + return DependencyCallableKind.UNKNOWN + + def _callable_structure( # noqa: PLR0912, PLR0915 + self, value: Any + ) -> tuple[tuple[DependencyCallableStructure, ...], tuple[str, ...]]: + """Describe callable shape without scanning metadata beyond retention bounds.""" + layers: list[DependencyCallableStructure] = [] + limitations: list[str] = [] + current = value + + def note(code: str) -> None: + if code not in limitations: + limitations.append(code) + + while len(layers) < 8: + kind = self._callable_kind(current) + module, qualname = self._callable_identity(current) + positional_count = 0 + keyword_names: tuple[str, ...] = () + if isinstance(current, functools.partial): + args_available, raw_args = self._safe_attribute(current, "args") + if not args_available or type(raw_args) is not tuple: + note("callable_arguments_invalid_shape") + else: + positional_count = len(raw_args) + if positional_count > 1024: + note("callable_positional_count_truncated") + + keywords_available, raw_keywords = self._safe_attribute(current, "keywords") + if not keywords_available or ( + raw_keywords is not None and type(raw_keywords) is not dict + ): + note("callable_keywords_invalid_shape") + elif raw_keywords is not None: + keyword_count = len(raw_keywords) + if keyword_count > 128: + # Count first: over-cap maps retain no key-level evidence and are + # never iterated, materialized, or sorted. + note("callable_keyword_names_truncated") + else: + bounded_names: list[str] = [] + invalid_name = False + for name in raw_keywords: + if type(name) is not str or not name: + invalid_name = True + else: + bounded_names.append(name) + if invalid_name: + note("callable_keyword_names_invalid_shape") + else: + bounded_names.sort() + if any(len(name) > 256 for name in bounded_names): + note("callable_keyword_name_truncated") + keyword_names = tuple(name[:256] for name in bounded_names) + + layers.append( + DependencyCallableStructure( + kind=kind, + module=module, + qualname=qualname, + bound_positional_count=min(positional_count, 1024), + bound_keyword_names=keyword_names, + ) + ) + if not isinstance(current, functools.partial): + break + func_available, next_callable = self._safe_attribute(current, "func") + if not func_available: + note("callable_structure_invalid_shape") + break + current = next_callable + if len(layers) == 8 and layers[-1].kind == DependencyCallableKind.PARTIAL: + note("callable_structure_truncated") + return tuple(layers), tuple(limitations) + + @staticmethod + def _safe_attribute(value: Any, name: str) -> tuple[bool, Any]: + try: + return True, getattr(value, name) + except Exception: + return False, None + + @classmethod + def _dependency_children(cls, node: Any) -> list[Any] | tuple[Any, ...] | None: + available, raw_children = cls._safe_attribute(node, "dependencies") + if not available: + return None + if raw_children is None: + return () + # FastAPI's supported Dependant implementations expose exact ordered + # list/tuple children. Never probe or materialize arbitrary iterables. + if type(raw_children) not in {list, tuple}: + return None + return cast("list[Any] | tuple[Any, ...]", raw_children) + + @staticmethod + def _limitation_source(handler: HandlerInfo) -> tuple[Path, int]: + return handler.file_path, max(handler.line_number, 1) + + def _extract_dependency_graph( # noqa: PLR0915 + self, route: Any, handler: HandlerInfo + ) -> EndpointDependencyGraph: + """Collect the declared FastAPI Dependant tree with deterministic hard bounds.""" + _effective_available, effective_route = self._safe_attribute(route, "starlette_route") + _original_available, original_route = self._safe_attribute(route, "original_route") + candidates: list[Any] = [] + for candidate in (effective_route, route, original_route): + if candidate is not None and all(candidate is not seen for seen in candidates): + candidates.append(candidate) + root = None + for candidate in candidates: + available, candidate_root = self._safe_attribute(candidate, "dependant") + if available and candidate_root is not None: + root = candidate_root + break + source_path, source_line = self._limitation_source(handler) + if root is None: + return EndpointDependencyGraph( + status=DependencyGraphStatus.UNAVAILABLE, + limitations=( + DependencyGraphLimitation( + code="dependant_unavailable", + source_path=source_path, + source_line=source_line, + reason="FastAPI route exposes no effective dependant graph", + ), + ), + ) + roots = self._dependency_children(root) + if roots is None: + return EndpointDependencyGraph( + status=DependencyGraphStatus.UNAVAILABLE, + limitations=( + DependencyGraphLimitation( + code="dependencies_unavailable", + source_path=source_path, + source_line=source_line, + reason="FastAPI dependant exposes no traversable dependencies", + ), + ), + ) + + occurrences: list[EndpointDependencyOccurrence] = [] + limitations: list[DependencyGraphLimitation] = [] + work = 0 + capped = False + + def limit(code: str, reason: str, span: DependencySourceSpan | None = None) -> None: + path = span.file_path if span is not None else source_path + line = span.start_line if span is not None else source_line + limitations.append( + DependencyGraphLimitation( + code=code, + source_path=path, + source_line=line, + reason=reason, + ) + ) + + def visit( # noqa: PLR0912, PLR0915 + node: Any, index_path: tuple[int, ...], ancestors: frozenset[int] + ) -> None: + nonlocal work, capped + if capped: + return + work += 1 + + has_call, call = self._safe_attribute(node, "call") + if not has_call: + call = None + callable_kind = self._callable_kind(call) + module, qualname = self._callable_identity(call) + span = self._dependency_source_span(call) + _display_available, display = self._safe_attribute(call, "__name__") + if not isinstance(display, str) or not display: + display = type(call).__name__ if call is not None else "" + if len(display) > 512: + limit( + "display_name_truncated", + f"dependency {index_path} display name exceeded its retention bound", + span, + ) + display = display[:512] + established = ( + has_call + and callable_kind != DependencyCallableKind.UNKNOWN + and module is not None + and qualname is not None + and "" not in qualname + and span is not None + ) + resolution = ( + DependencyResolutionStatus.ESTABLISHED + if established + else DependencyResolutionStatus.UNAVAILABLE + if not has_call + else DependencyResolutionStatus.CONDITIONAL + ) + + own_available, raw_own_scopes = self._safe_attribute(node, "own_oauth_scopes") + legacy_available, raw_legacy_scopes = self._safe_attribute(node, "security_scopes") + scopes_available = own_available or legacy_available + declaration_local = own_available + raw_scopes = raw_own_scopes if own_available else raw_legacy_scopes + raw_scope_count = 0 + invalid_scope_member = False + truncated_scope = False + scopes_list: list[str] = [] + if raw_scopes is None: + pass + elif type(raw_scopes) in {list, tuple, set, frozenset}: + raw_scope_count = len(raw_scopes) + unordered = type(raw_scopes) in {set, frozenset} + if unordered: + limit( + "security_scopes_unordered_shape", + f"dependency {index_path} exposes unordered security scopes", + span, + ) + if raw_scope_count > 256: + # Count first: every supported over-cap shape retains no + # member-level evidence and is never iterated or sorted. + limit( + "security_scope_count_truncated", + f"dependency {index_path} security-scope count exceeded 256", + span, + ) + else: + bounded_scopes: list[str] = [] + for raw_scope in raw_scopes: + if type(raw_scope) is not str or not raw_scope: + invalid_scope_member = True + else: + if len(raw_scope) > 512: + truncated_scope = True + bounded_scopes.append(raw_scope) + # Validate the complete bounded collection before sorting or + # retaining any member-level evidence. + if not invalid_scope_member: + if unordered: + bounded_scopes.sort() + scopes_list = [scope[:512] for scope in bounded_scopes] + else: + limit( + "security_scopes_invalid_shape", + f"dependency {index_path} exposes invalid security-scope metadata", + span, + ) + if invalid_scope_member: + limit( + "security_scope_member_invalid", + f"dependency {index_path} contains a non-string or blank security scope", + span, + ) + if truncated_scope: + limit( + "security_scope_string_truncated", + f"dependency {index_path} contains an overlong security scope", + span, + ) + scopes = tuple(scopes_list) + declaration_kind = ( + DependencyDeclarationKind.SECURITY + if declaration_local and scopes + else DependencyDeclarationKind.DEPENDS_OR_SECURITY + if scopes_available + else DependencyDeclarationKind.UNKNOWN + ) + has_use_cache, raw_use_cache = self._safe_attribute(node, "use_cache") + use_cache = raw_use_cache if isinstance(raw_use_cache, bool) else None + _name_available, raw_name = self._safe_attribute(node, "name") + scope = ( + DependencyDeclarationScope.NESTED + if len(index_path) > 1 + else DependencyDeclarationScope.PARAMETER + if isinstance(raw_name, str) and raw_name + else DependencyDeclarationScope.ASSEMBLY + ) + callable_structure, structure_limitations = self._callable_structure(call) + for structure_limitation in structure_limitations: + limit( + structure_limitation, + f"dependency {index_path} callable structure metadata was malformed " + "or exceeded a retention bound", + span, + ) + occurrences.append( + EndpointDependencyOccurrence( + index_path=index_path, + parent_path=index_path[:-1], + depth=len(index_path), + order=len(occurrences), + declaration_scope=scope, + declaration_kind=declaration_kind, + callable_kind=callable_kind, + resolution_status=resolution, + display_name=display, + module=module, + qualname=qualname, + source_span=span, + security_scopes=scopes, + use_cache=use_cache, + callable_structure=callable_structure, + ) + ) + if not established: + limit( + "callable_identity_unavailable", + f"dependency {index_path} has no source-attested qualified callable identity", + span, + ) + if not scopes_available: + limit( + "declaration_kind_unavailable", + f"dependency {index_path} does not expose security-scope metadata", + span, + ) + if not has_use_cache or use_cache is None: + limit( + "cache_semantics_unavailable", + f"dependency {index_path} does not expose boolean use_cache semantics", + span, + ) + + node_id = id(node) + if node_id in ancestors: + limit("cycle", f"dependency {index_path} closes an internal dependant cycle", span) + return + children = self._dependency_children(node) + if children is None: + limit( + "nested_dependencies_unavailable", + f"dependency {index_path} exposes no traversable nested dependencies", + span, + ) + return + if len(index_path) >= self.dependency_max_depth: + if children: + limit("depth_cap", f"dependency {index_path} reached the depth cap", span) + return + next_ancestors = ancestors | {node_id} + for child_index in range(len(children)): + if work >= self.dependency_max_work: + capped = True + limit("work_cap", "dependency graph traversal reached its work cap", span) + break + if len(occurrences) >= self.dependency_max_nodes: + capped = True + limit("node_cap", "dependency graph traversal reached its node cap", span) + break + visit(children[child_index], (*index_path, child_index), next_ancestors) + if capped: + break + + for root_index in range(len(roots)): + if work >= self.dependency_max_work: + capped = True + limit("work_cap", "dependency graph traversal reached its work cap") + break + if len(occurrences) >= self.dependency_max_nodes: + capped = True + limit("node_cap", "dependency graph traversal reached its node cap") + break + visit(roots[root_index], (root_index,), frozenset()) + if capped: + break + + overrides = getattr(self._app, "dependency_overrides", None) + try: + overrides_visible = bool(overrides) + except Exception: + overrides_visible = True + if overrides_visible: + limit( + "dependency_overrides_visible", + "dependency overrides are visible; only the declared graph is modeled", + ) + + return EndpointDependencyGraph( + status=( + DependencyGraphStatus.CONDITIONAL + if limitations + else DependencyGraphStatus.ESTABLISHED + ), + occurrences=tuple(occurrences), + limitations=tuple(limitations), + ) + @staticmethod def _join_paths(prefix: str, path: str) -> str: """Join effective include and mount paths without losing root slashes.""" @@ -346,23 +837,28 @@ def _http_methods(route: Any, original_route: Any) -> list[EndpointMethod]: def _http_endpoint(self, route: Any, original_route: Any, prefix: str) -> Endpoint: path = self._require_route_path(route, original_route) endpoint = self._require_route_endpoint(route, original_route) + handler = self._get_handler_info(endpoint) return Endpoint( path=self._join_paths(prefix, path), methods=self._http_methods(route, original_route), - handler=self._get_handler_info(endpoint), + handler=handler, name=getattr(route, "name", None), tags=list(getattr(route, "tags", None) or []), dependencies=self._extract_dependencies(route), + dependency_graph=self._extract_dependency_graph(route, handler), ) def _websocket_endpoint(self, route: Any, original_route: Any, prefix: str) -> Endpoint: metadata = self._effective_route_metadata(route) + endpoint = self._require_route_endpoint(metadata, original_route) + handler = self._get_handler_info(endpoint) return Endpoint( path=self._join_paths(prefix, self._require_route_path(metadata, original_route)), methods=[EndpointMethod.WEBSOCKET], - handler=self._get_handler_info(self._require_route_endpoint(metadata, original_route)), + handler=handler, name=getattr(metadata, "name", None), dependencies=self._extract_dependencies(metadata), + dependency_graph=self._extract_dependency_graph(route, handler), ) def _endpoints_from_route( @@ -462,7 +958,7 @@ def _read_runtime_result(self, result_path: Path, returncode: int) -> list[Endpo payload = json.loads(result_path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise FastAPIExtractorError("Runtime worker returned an invalid response") from exc - if not isinstance(payload, dict) or payload.get("schema_version") != 1: + if not isinstance(payload, dict) or payload.get("schema_version") != 2: raise FastAPIExtractorError("Runtime worker returned an unsupported response") if payload.get("status") == "error": message = payload.get("message") @@ -490,11 +986,14 @@ def extract_endpoints(self) -> list[Endpoint]: "Runtime subprocess isolation requires POSIX; use secure AST or VM mode" ) request = { - "schema_version": 1, + "schema_version": 2, "app_path": str(self.app_path), "app_variable": self.app_variable, "module_name": self.module_name, "output_limit_bytes": self.output_limit_bytes, + "dependency_max_depth": self.dependency_max_depth, + "dependency_max_nodes": self.dependency_max_nodes, + "dependency_max_work": self.dependency_max_work, } with tempfile.TemporaryDirectory(prefix="fastapi-endpoint-runtime-") as temp_dir: result_path = Path(temp_dir) / "result.json" diff --git a/src/fastapi_endpoint_detector/parser/runtime_worker.py b/src/fastapi_endpoint_detector/parser/runtime_worker.py index c590a23..ac506e0 100644 --- a/src/fastapi_endpoint_detector/parser/runtime_worker.py +++ b/src/fastapi_endpoint_detector/parser/runtime_worker.py @@ -12,7 +12,7 @@ from fastapi_endpoint_detector.parser.fastapi_extractor import FastAPIExtractor -_PROTOCOL_VERSION = 1 +_PROTOCOL_VERSION = 2 _DEFAULT_OUTPUT_LIMIT_BYTES = 4 * 1024 * 1024 @@ -33,6 +33,13 @@ def _required_string(request: dict[str, Any], field: str) -> str: return value +def _positive_integer(request: dict[str, Any], field: str) -> int: + value = request.get(field) + if not isinstance(value, int) or isinstance(value, bool) or value <= 0: + raise ValueError(f"runtime worker request requires positive integer {field}") + return value + + def _write_result(result_path: Path, payload: dict[str, Any], output_limit: int) -> None: encoded = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8") if len(encoded) > output_limit: @@ -64,6 +71,9 @@ def _run(result_path: Path) -> int: Path(_required_string(request, "app_path")), app_variable=_required_string(request, "app_variable"), module_name=module_name, + dependency_max_depth=_positive_integer(request, "dependency_max_depth"), + dependency_max_nodes=_positive_integer(request, "dependency_max_nodes"), + dependency_max_work=_positive_integer(request, "dependency_max_work"), ) with ( Path(os.devnull).open("w", encoding="utf-8") as sink, diff --git a/tests/integration/test_di_patterns.py b/tests/integration/test_di_patterns.py index 60549ed..80d59ea 100644 --- a/tests/integration/test_di_patterns.py +++ b/tests/integration/test_di_patterns.py @@ -166,11 +166,14 @@ def test_token_decode_change_affects_secured_endpoints( assert result.exit_code == 0, f"CLI failed: {result.output}" + # require_scope returns check_scope, whose dependency path is now proven as + # check_scope -> get_current_user -> decode_token for POST /items. assert affected_ids(result.output) == { "DELETE /users/{user_id}", "GET /protected", "GET /users/me", "GET /users/me/items", + "POST /items", } diff --git a/tests/unit/test_fastapi_extractor.py b/tests/unit/test_fastapi_extractor.py index 6f5cc87..a86dd4a 100644 --- a/tests/unit/test_fastapi_extractor.py +++ b/tests/unit/test_fastapi_extractor.py @@ -1,5 +1,6 @@ """Runtime FastAPI extractor parity tests.""" +import functools import os import py_compile import subprocess @@ -10,7 +11,14 @@ from typing import Any import pytest +from fastapi import Depends, FastAPI +from fastapi_endpoint_detector.models.endpoint import ( + DependencyCallableKind, + DependencyDeclarationKind, + DependencyGraphStatus, + HandlerInfo, +) from fastapi_endpoint_detector.parser.fastapi_extractor import ( FastAPIExtractor, FastAPIExtractorError, @@ -171,6 +179,670 @@ async def socket(websocket, token=Depends(parameter_dependency)): assert websocket.handler.name == "socket" +def test_runtime_dependency_graph_preserves_nested_repeated_security_and_callable_shapes( + tmp_path: Path, +) -> None: + app_file = tmp_path / "dependency_graph_app.py" + app_file.write_text( + """from functools import partial +from typing import Annotated +from fastapi import Depends, FastAPI, Security + + +def helper(): + return 1 + + +def nested(value=Depends(helper)): + return value + + +def authorize(): + return 1 + + +class Provider: + def __call__(self, value=Depends(nested)): + return value + + +provider = Provider() +app = FastAPI(dependencies=[Depends(nested), Depends(nested)]) + +@app.get("/graph") +def graph( + auth: Annotated[int, Security(authorize, scopes=["read"])], + partial_value=Depends(partial(nested)), + instance_value=Depends(provider), +): + return auth +""" + ) + + endpoint = FastAPIExtractor(app_file).extract_endpoints()[0] + graph = endpoint.dependency_graph + + assert endpoint.dependencies == ["nested", "nested", "partial", "Provider"] + assert graph is not None + assert graph.status == DependencyGraphStatus.ESTABLISHED + assert graph.semantics == "declared" + assert [item.index_path for item in graph.occurrences[:4]] == [ + (0,), + (0, 0), + (1,), + (1, 0), + ] + assert [item.display_name for item in graph.occurrences].count("nested") == 3 + security = next(item for item in graph.occurrences if item.display_name == "authorize") + assert security.declaration_kind in { + DependencyDeclarationKind.SECURITY, + DependencyDeclarationKind.DEPENDS_OR_SECURITY, + } + assert security.security_scopes == ("read",) + partial_item = next( + item for item in graph.occurrences if item.callable_kind == DependencyCallableKind.PARTIAL + ) + assert [layer.kind for layer in partial_item.callable_structure] == [ + DependencyCallableKind.PARTIAL, + DependencyCallableKind.FUNCTION, + ] + assert any( + item.callable_kind == DependencyCallableKind.CALLABLE_INSTANCE for item in graph.occurrences + ) + assert [item.order for item in graph.occurrences] == list(range(len(graph.occurrences))) + assert all("0x" not in item.display_name for item in graph.occurrences) + + +def test_runtime_dependency_graph_cycle_and_caps_are_conditional_and_deterministic( + tmp_path: Path, +) -> None: + cycle_app = tmp_path / "cycle_app.py" + cycle_app.write_text( + """from fastapi import Depends, FastAPI + + +def dependency(): + return 1 + + +app = FastAPI() + +@app.get("/cycle") +def cycle(value=Depends(dependency)): + return value + +route = next(route for route in app.routes if getattr(route, "path", None) == "/cycle") +node = route.dependant.dependencies[0] +node.dependencies.append(node) +""" + ) + cycle_endpoint = FastAPIExtractor(cycle_app).extract_endpoints()[0] + cycle_graph = cycle_endpoint.dependency_graph + assert cycle_graph is not None + assert cycle_graph.status == DependencyGraphStatus.CONDITIONAL + assert [item.index_path for item in cycle_graph.occurrences] == [(0,), (0, 0)] + assert "cycle" in {limitation.code for limitation in cycle_graph.limitations} + + capped_app = tmp_path / "capped_app.py" + capped_app.write_text( + """from fastapi import Depends, FastAPI + + +def leaf(): + return 1 + + +def middle(value=Depends(leaf)): + return value + + +app = FastAPI() + +@app.get("/capped") +def capped(value=Depends(middle)): + return value +""" + ) + first = FastAPIExtractor(capped_app, dependency_max_depth=1).extract_endpoints()[0] + second = FastAPIExtractor(capped_app, dependency_max_depth=1).extract_endpoints()[0] + assert first.dependency_graph == second.dependency_graph + assert first.dependency_graph is not None + assert first.dependency_graph.status == DependencyGraphStatus.CONDITIONAL + assert [item.index_path for item in first.dependency_graph.occurrences] == [(0,)] + assert "depth_cap" in {limitation.code for limitation in first.dependency_graph.limitations} + + node_capped = FastAPIExtractor(capped_app, dependency_max_nodes=1).extract_endpoints()[0] + assert node_capped.dependency_graph is not None + assert "node_cap" in { + limitation.code for limitation in node_capped.dependency_graph.limitations + } + work_capped = FastAPIExtractor(capped_app, dependency_max_work=1).extract_endpoints()[0] + assert work_capped.dependency_graph is not None + assert "work_cap" in { + limitation.code for limitation in work_capped.dependency_graph.limitations + } + + +def test_runtime_dependency_graph_missing_shape_and_overrides_are_graph_local( + tmp_path: Path, +) -> None: + handler = HandlerInfo( + name="endpoint", module="main", file_path=tmp_path / "main.py", line_number=4 + ) + extractor = FastAPIExtractor(tmp_path / "main.py") + extractor._app = SimpleNamespace(dependency_overrides={}) + unavailable = extractor._extract_dependency_graph(SimpleNamespace(), handler) + assert unavailable.status == DependencyGraphStatus.UNAVAILABLE + assert unavailable.limitations[0].source_path == handler.file_path + + root = SimpleNamespace(dependencies=[]) + extractor._app = SimpleNamespace(dependency_overrides={object(): object()}) + conditional = extractor._extract_dependency_graph(SimpleNamespace(dependant=root), handler) + assert conditional.status == DependencyGraphStatus.CONDITIONAL + assert conditional.occurrences == () + assert {item.code for item in conditional.limitations} == {"dependency_overrides_visible"} + + +def _synthetic_old_dependency() -> int: + return 1 + + +def _synthetic_effective_dependency() -> int: + return 2 + + +def test_dependency_graph_prefers_effective_non_none_root_without_masking( + tmp_path: Path, +) -> None: + handler = HandlerInfo( + name="endpoint", module="main", file_path=tmp_path / "main.py", line_number=1 + ) + extractor = FastAPIExtractor(tmp_path / "main.py") + extractor._app = SimpleNamespace(dependency_overrides={}) + + def node(call: Any) -> SimpleNamespace: + return SimpleNamespace( + call=call, + dependencies=[], + own_oauth_scopes=[], + use_cache=True, + name=None, + ) + + old_root = SimpleNamespace(dependencies=[node(_synthetic_old_dependency)]) + effective_root = SimpleNamespace(dependencies=[node(_synthetic_effective_dependency)]) + wrapper = SimpleNamespace( + dependant=old_root, + starlette_route=SimpleNamespace(dependant=effective_root), + original_route=SimpleNamespace(dependant=old_root), + ) + graph = extractor._extract_dependency_graph(wrapper, handler) + assert [item.display_name for item in graph.occurrences] == ["_synthetic_effective_dependency"] + + wrapper.dependant = None + assert extractor._extract_dependency_graph(wrapper, handler).occurrences[0].display_name == ( + "_synthetic_effective_dependency" + ) + direct = SimpleNamespace(dependant=effective_root) + assert extractor._extract_dependency_graph(direct, handler).occurrences[0].display_name == ( + "_synthetic_effective_dependency" + ) + + class RaisingWrapper: + starlette_route = SimpleNamespace(dependant=effective_root) + + @property + def dependant(self) -> Any: + raise RuntimeError("stale wrapper") + + assert extractor._extract_dependency_graph(RaisingWrapper(), handler).occurrences + + +def test_normalized_http_and_websocket_routes_use_effective_dependency_roots( + tmp_path: Path, +) -> None: + app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None) + + @app.get("/http", dependencies=[Depends(_synthetic_effective_dependency)]) + def http_handler() -> None: + pass + + @app.websocket("/websocket", dependencies=[Depends(_synthetic_effective_dependency)]) + async def websocket_handler(websocket: Any) -> None: + pass + + extractor = FastAPIExtractor(tmp_path / "main.py") + extractor._app = app + for original in app.routes: + direct = extractor._endpoints_from_route(original, "", frozenset())[0] + assert direct.dependency_graph is not None + assert direct.dependency_graph.occurrences[0].display_name == ( + "_synthetic_effective_dependency" + ) + stale = SimpleNamespace( + dependencies=[ + SimpleNamespace(dependency=_synthetic_old_dependency), + ] + ) + wrapper = SimpleNamespace( + original_route=original, + starlette_route=original, + dependant=stale, + path=original.path, + endpoint=original.endpoint, + methods=getattr(original, "methods", None), + name=original.name, + tags=getattr(original, "tags", []), + dependencies=getattr(original, "dependencies", []), + ) + normalized = extractor._endpoints_from_route(wrapper, "", frozenset())[0] + assert normalized.dependency_graph is not None + assert normalized.dependency_graph.occurrences[0].display_name == ( + "_synthetic_effective_dependency" + ) + + +def test_dependency_graph_rejects_arbitrary_dependency_iterables_without_pulling( + tmp_path: Path, +) -> None: + pulls = 0 + + def infinite() -> Any: + nonlocal pulls + while True: + pulls += 1 + yield object() + + handler = HandlerInfo( + name="endpoint", module="main", file_path=tmp_path / "main.py", line_number=1 + ) + extractor = FastAPIExtractor( + tmp_path / "main.py", dependency_max_nodes=1, dependency_max_work=1 + ) + extractor._app = SimpleNamespace(dependency_overrides={}) + graph = extractor._extract_dependency_graph( + SimpleNamespace(dependant=SimpleNamespace(dependencies=infinite())), handler + ) + assert graph.status == DependencyGraphStatus.UNAVAILABLE + assert pulls == 0 + + nested = SimpleNamespace( + call=_synthetic_effective_dependency, + dependencies=infinite(), + own_oauth_scopes=[], + use_cache=True, + name=None, + ) + nested_graph = extractor._extract_dependency_graph( + SimpleNamespace(dependant=SimpleNamespace(dependencies=[nested])), handler + ) + assert nested_graph.status == DependencyGraphStatus.CONDITIONAL + assert pulls == 0 + assert "nested_dependencies_unavailable" in {item.code for item in nested_graph.limitations} + + +@pytest.mark.parametrize( + ("keyword", "value"), + [ + ("dependency_max_depth", 65), + ("dependency_max_nodes", 4097), + ("dependency_max_work", 65537), + ("output_limit_bytes", 64 * 1024 * 1024 + 1), + ], +) +def test_runtime_dependency_constructor_caps_have_finite_upper_bounds( + tmp_path: Path, keyword: str, value: int +) -> None: + with pytest.raises(ValueError, match=r"must not exceed|not exceeding"): + if keyword == "dependency_max_depth": + FastAPIExtractor(tmp_path / "main.py", dependency_max_depth=value) + elif keyword == "dependency_max_nodes": + FastAPIExtractor(tmp_path / "main.py", dependency_max_nodes=value) + elif keyword == "dependency_max_work": + FastAPIExtractor(tmp_path / "main.py", dependency_max_work=value) + else: + FastAPIExtractor(tmp_path / "main.py", output_limit_bytes=value) + + +def test_dependency_security_metadata_is_local_bounded_and_deterministic( + tmp_path: Path, +) -> None: + handler = HandlerInfo( + name="endpoint", module="main", file_path=tmp_path / "main.py", line_number=1 + ) + extractor = FastAPIExtractor(tmp_path / "main.py") + extractor._app = SimpleNamespace(dependency_overrides={}) + node = SimpleNamespace( + call=_synthetic_effective_dependency, + dependencies=[], + own_oauth_scopes={"write", "read"}, + security_scopes=["inherited"], + use_cache=True, + name="value", + ) + graph = extractor._extract_dependency_graph( + SimpleNamespace(dependant=SimpleNamespace(dependencies=[node])), handler + ) + assert graph.occurrences[0].security_scopes == ("read", "write") + assert graph.occurrences[0].declaration_kind == DependencyDeclarationKind.SECURITY + assert "security_scopes_unordered_shape" in {item.code for item in graph.limitations} + + del node.own_oauth_scopes + legacy = extractor._extract_dependency_graph( + SimpleNamespace(dependant=SimpleNamespace(dependencies=[node])), handler + ) + assert legacy.occurrences[0].security_scopes == ("inherited",) + assert legacy.occurrences[0].declaration_kind == DependencyDeclarationKind.DEPENDS_OR_SECURITY + + node.own_oauth_scopes = [1, "x" * 513, *(["retained"] * 254)] + invalid = extractor._extract_dependency_graph( + SimpleNamespace(dependant=SimpleNamespace(dependencies=[node])), handler + ) + assert invalid.occurrences[0].security_scopes == () + assert { + "security_scope_member_invalid", + "security_scope_string_truncated", + } <= {item.code for item in invalid.limitations} + + node.own_oauth_scopes = ["retained"] * 257 + over_cap = extractor._extract_dependency_graph( + SimpleNamespace(dependant=SimpleNamespace(dependencies=[node])), handler + ) + assert over_cap.occurrences[0].security_scopes == () + assert "security_scope_count_truncated" in {item.code for item in over_cap.limitations} + + pulls = 0 + + def scopes_generator() -> Any: + nonlocal pulls + pulls += 1 + yield "read" + + node.own_oauth_scopes = scopes_generator() + invalid_shape = extractor._extract_dependency_graph( + SimpleNamespace(dependant=SimpleNamespace(dependencies=[node])), handler + ) + assert pulls == 0 + assert "security_scopes_invalid_shape" in {item.code for item in invalid_shape.limitations} + + +def test_callable_and_scope_metadata_bounds_precede_iteration_and_sorting( + tmp_path: Path, +) -> None: + keyword_comparisons = 0 + keyword_member_reads = 0 + scope_comparisons = 0 + scope_member_reads = 0 + + class Keyword(str): + def __lt__(self, other: object) -> bool: + nonlocal keyword_comparisons + keyword_comparisons += 1 + return super().__lt__(other) + + def __len__(self) -> int: + nonlocal keyword_member_reads + keyword_member_reads += 1 + return super().__len__() + + class Scope(str): + def __lt__(self, other: object) -> bool: + nonlocal scope_comparisons + scope_comparisons += 1 + return super().__lt__(other) + + def __len__(self) -> int: + nonlocal scope_member_reads + scope_member_reads += 1 + return super().__len__() + + structured = functools.partial(_synthetic_effective_dependency) + assert structured.keywords is not None + for index in range(129): + structured.keywords[Keyword(f"key_{index:03}")] = index + scopes = {Scope(f"scope_{index:03}") for index in range(257)} + keyword_comparisons = keyword_member_reads = 0 + scope_comparisons = scope_member_reads = 0 + + node = SimpleNamespace( + call=structured, + dependencies=[], + own_oauth_scopes=scopes, + use_cache=True, + name=None, + ) + handler = HandlerInfo( + name="endpoint", module="main", file_path=tmp_path / "main.py", line_number=1 + ) + extractor = FastAPIExtractor( + tmp_path / "main.py", dependency_max_nodes=1, dependency_max_work=1 + ) + extractor._app = SimpleNamespace(dependency_overrides={}) + graph = extractor._extract_dependency_graph( + SimpleNamespace(dependant=SimpleNamespace(dependencies=[node])), handler + ) + + occurrence = graph.occurrences[0] + assert occurrence.callable_structure[0].bound_keyword_names == () + assert occurrence.security_scopes == () + assert keyword_comparisons == keyword_member_reads == 0 + assert scope_comparisons == scope_member_reads == 0 + assert { + "callable_keyword_names_truncated", + "security_scope_count_truncated", + "security_scopes_unordered_shape", + } <= {item.code for item in graph.limitations} + + for ordered_scopes in ([Scope("read")] * 257, tuple([Scope("read")] * 257)): + scope_member_reads = 0 + node.own_oauth_scopes = ordered_scopes + ordered_graph = extractor._extract_dependency_graph( + SimpleNamespace(dependant=SimpleNamespace(dependencies=[node])), handler + ) + assert ordered_graph.occurrences[0].security_scopes == () + assert scope_member_reads == 0 + + +def test_at_cap_metadata_validates_types_before_sorting(tmp_path: Path) -> None: + comparisons = 0 + + class ComparedString(str): + def __lt__(self, other: object) -> bool: + nonlocal comparisons + comparisons += 1 + return super().__lt__(other) + + structured = functools.partial(_synthetic_effective_dependency) + assert structured.keywords is not None + for index in range(127): + structured.keywords[ComparedString(f"key_{index:03}")] = index + structured.keywords[1] = "malformed" + scopes: set[object] = {ComparedString(f"scope_{index:03}") for index in range(255)} + scopes.add(1) + comparisons = 0 + node = SimpleNamespace( + call=structured, + dependencies=[], + own_oauth_scopes=scopes, + use_cache=True, + name=None, + ) + handler = HandlerInfo( + name="endpoint", module="main", file_path=tmp_path / "main.py", line_number=1 + ) + extractor = FastAPIExtractor( + tmp_path / "main.py", dependency_max_nodes=1, dependency_max_work=1 + ) + extractor._app = SimpleNamespace(dependency_overrides={}) + + graph = extractor._extract_dependency_graph( + SimpleNamespace(dependant=SimpleNamespace(dependencies=[node])), handler + ) + + assert graph.occurrences[0].callable_structure[0].bound_keyword_names == () + assert graph.occurrences[0].security_scopes == () + assert comparisons == 0 + assert { + "callable_keyword_names_invalid_shape", + "security_scope_member_invalid", + } <= {item.code for item in graph.limitations} + + +@pytest.mark.parametrize( + "mutation", + [ + "registered.keywords[1] = 'malformed-after-registration'", + "registered.keywords[''] = 'malformed-after-registration'", + ], +) +def test_post_registration_malformed_partial_metadata_is_graph_local( + tmp_path: Path, mutation: str +) -> None: + app_file = tmp_path / "mutated_partial_app.py" + app_file.write_text( + "from functools import partial\n" + "from fastapi import Depends, FastAPI\n\n" + "def dependency(*, marker=1):\n return marker\n\n" + "registered = partial(dependency, marker=1)\n" + "app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)\n\n" + "@app.get('/healthy')\n" + "def healthy():\n return True\n\n" + "@app.get('/partial')\n" + "def endpoint(value=Depends(registered)):\n return value\n\n" + "registered.keywords.clear()\n" + f"{mutation}\n" + ) + + endpoints = FastAPIExtractor( + app_file, dependency_max_nodes=1, dependency_max_work=1 + ).extract_endpoints() + + assert [endpoint.identifier for endpoint in endpoints] == ["GET /healthy", "GET /partial"] + partial = next(endpoint for endpoint in endpoints if endpoint.identifier == "GET /partial") + assert partial.discovery_status.value == "established" + graph = partial.dependency_graph + assert graph is not None + assert graph.status == DependencyGraphStatus.CONDITIONAL + assert "callable_keyword_names_invalid_shape" in {item.code for item in graph.limitations} + assert graph.occurrences[0].callable_structure[0].bound_keyword_names == () + + +def test_callable_structure_and_display_truncations_condition_graph(tmp_path: Path) -> None: + def dependency(*args: Any, **kwargs: Any) -> int: + return len(args) + len(kwargs) + + dependency.__name__ = "d" * 513 + structured: Any = functools.partial( + dependency, + *range(1025), + **{f"key_{index}": index for index in range(129)}, + ) + node = SimpleNamespace( + call=structured, + dependencies=[], + own_oauth_scopes=[], + use_cache=True, + name=None, + ) + display_node = SimpleNamespace( + call=dependency, + dependencies=[], + own_oauth_scopes=[], + use_cache=True, + name=None, + ) + handler = HandlerInfo( + name="endpoint", module="main", file_path=tmp_path / "main.py", line_number=1 + ) + extractor = FastAPIExtractor(tmp_path / "main.py") + extractor._app = SimpleNamespace(dependency_overrides={}) + graph = extractor._extract_dependency_graph( + SimpleNamespace(dependant=SimpleNamespace(dependencies=[node, display_node])), handler + ) + assert graph.status == DependencyGraphStatus.CONDITIONAL + assert { + "callable_positional_count_truncated", + "callable_keyword_names_truncated", + "display_name_truncated", + } <= {item.code for item in graph.limitations} + partial_occurrence = next( + occurrence + for occurrence in graph.occurrences + if occurrence.callable_kind == DependencyCallableKind.PARTIAL + ) + assert partial_occurrence.callable_structure[0].bound_keyword_names == () + + +def test_security_set_output_is_hash_seed_deterministic(tmp_path: Path) -> None: + script = f""" +from pathlib import Path +from types import SimpleNamespace +from fastapi_endpoint_detector.models.endpoint import HandlerInfo +from fastapi_endpoint_detector.parser.fastapi_extractor import FastAPIExtractor +node = SimpleNamespace( + call=len, dependencies=[], own_oauth_scopes={{'write', 'read'}}, + use_cache=True, name='v', +) +extractor = FastAPIExtractor(Path({str(tmp_path / "main.py")!r})) +extractor._app = SimpleNamespace(dependency_overrides={{}}) +graph = extractor._extract_dependency_graph( + SimpleNamespace(dependant=SimpleNamespace(dependencies=[node])), + HandlerInfo( + name='handler', module='main', + file_path=Path({str(tmp_path / "main.py")!r}), line_number=1, + ), +) +print(graph.occurrences[0].security_scopes) +print(tuple(item.code for item in graph.limitations)) +""" + outputs = [] + for seed in ("1", "2", "3", "4"): + environment = os.environ.copy() + environment["PYTHONHASHSEED"] = seed + environment["PYTHONPATH"] = str(Path(__file__).parents[2] / "src") + outputs.append( + subprocess.check_output( + [sys.executable, "-c", script], + env=environment, + text=True, + timeout=10, + ) + ) + assert len(set(outputs)) == 1 + + +def test_nested_security_scopes_do_not_reclassify_depends_declarations(tmp_path: Path) -> None: + app_file = tmp_path / "security_app.py" + app_file.write_text( + "from fastapi import Depends, FastAPI, Security\n\n" + "def leaf() -> int:\n return 1\n\n" + "def auth(value: int = Depends(leaf)) -> int:\n return value\n\n" + "app = FastAPI()\n\n" + "@app.get('/security')\n" + "def endpoint(value: int = Security(auth, scopes=['read'])) -> int:\n" + " return value\n" + ) + graph = FastAPIExtractor(app_file).extract_endpoints()[0].dependency_graph + assert graph is not None + leaf = next(item for item in graph.occurrences if item.display_name == "leaf") + assert leaf.declaration_kind == DependencyDeclarationKind.DEPENDS_OR_SECURITY + + +def test_runtime_worker_payload_limit_fails_controlled(tmp_path: Path) -> None: + app_file = tmp_path / "main.py" + app_file.write_text( + """from fastapi import FastAPI +app = FastAPI() +@app.get("/payload") +def payload(): + return {} +""" + ) + + with pytest.raises(FastAPIExtractorError, match="output limit"): + FastAPIExtractor(app_file, output_limit_bytes=128).extract_endpoints() + + def test_runtime_and_secure_extractors_agree_on_static_nested_routes(tmp_path: Path) -> None: app_file = tmp_path / "differential_app.py" app_file.write_text( diff --git a/tests/unit/test_formatters.py b/tests/unit/test_formatters.py index 56a3f03..bdef16d 100644 --- a/tests/unit/test_formatters.py +++ b/tests/unit/test_formatters.py @@ -10,7 +10,9 @@ import yaml from fastapi_endpoint_detector.models.endpoint import ( + DependencyGraphStatus, Endpoint, + EndpointDependencyGraph, EndpointDiscoveryCondition, EndpointDiscoveryStatus, EndpointInventory, @@ -55,7 +57,7 @@ def test_inventory_strength_is_structured_and_visible() -> None: json_result = json.loads(JsonFormatter().format_inventory(inventory)) yaml_result = yaml.safe_load(YamlFormatter().format_inventory(inventory)) - assert json_result["schema_version"] == 2 + assert json_result["schema_version"] == 3 assert json_result["inventory_status"] == "conditional" assert json_result["inventory_limitations"][0]["reason"] == limitation.reason assert json_result["route_conditions"][0]["reason"] == limitation.reason @@ -105,6 +107,28 @@ def test_unavailable_report_inventory_is_visible_in_all_formats() -> None: assert limitation.reason in rendered +def test_json_and_yaml_preserve_optional_dependency_graph() -> None: + endpoint = Endpoint( + path="/graph", + methods=[EndpointMethod.GET], + handler=HandlerInfo( + name="graph", module="main", file_path=Path("/app/main.py"), line_number=3 + ), + dependency_graph=EndpointDependencyGraph(status=DependencyGraphStatus.ESTABLISHED), + ) + + json_endpoint = json.loads(JsonFormatter().format_endpoints([endpoint]))["endpoints"][0] + yaml_endpoint = yaml.safe_load(YamlFormatter().format_endpoints([endpoint]))["endpoints"][0] + assert json_endpoint["dependency_graph"] == yaml_endpoint["dependency_graph"] + assert json_endpoint["dependency_graph"] == { + "schema_version": 1, + "status": "established", + "semantics": "declared", + "occurrences": [], + "limitations": [], + } + + def test_json_and_yaml_preserve_startup_activation_evidence() -> None: condition = EndpointDiscoveryCondition( source_path=Path("/app/main.py"), diff --git a/tests/unit/test_models.py b/tests/unit/test_models.py index cc299c3..0fae2cf 100644 --- a/tests/unit/test_models.py +++ b/tests/unit/test_models.py @@ -2,6 +2,7 @@ Unit tests for the Pydantic models. """ +import json from pathlib import Path import pytest @@ -13,7 +14,16 @@ DiffHunk, ) from fastapi_endpoint_detector.models.endpoint import ( + DependencyCallableKind, + DependencyDeclarationKind, + DependencyDeclarationScope, + DependencyGraphLimitation, + DependencyGraphStatus, + DependencyResolutionStatus, + DependencySourceSpan, Endpoint, + EndpointDependencyGraph, + EndpointDependencyOccurrence, EndpointDiscoveryCondition, EndpointDiscoveryStatus, EndpointInventory, @@ -78,6 +88,101 @@ def test_create_endpoint(self) -> None: assert endpoint.path == "/api/users" assert EndpointMethod.GET in endpoint.methods + def test_dependency_graph_none_and_established_empty_are_distinct(self) -> None: + handler = HandlerInfo( + name="get_users", + module="routers.users", + file_path=Path("/app/routers/users.py"), + line_number=10, + ) + legacy = Endpoint(path="/legacy", methods=[EndpointMethod.GET], handler=handler) + collected = legacy.model_copy( + update={ + "dependency_graph": EndpointDependencyGraph( + status=DependencyGraphStatus.ESTABLISHED + ) + } + ) + + assert legacy.dependency_graph is None + assert collected.dependency_graph is not None + assert collected.dependency_graph.occurrences == () + with pytest.raises(ValidationError): + collected.dependency_graph.status = DependencyGraphStatus.CONDITIONAL + + def test_dependency_graph_uncertainty_requires_graph_scoped_limitation(self) -> None: + limitation = DependencyGraphLimitation( + code="shape_unavailable", + source_path=Path("/app/main.py"), + source_line=4, + reason="FastAPI dependant shape changed", + ) + graph = EndpointDependencyGraph( + status=DependencyGraphStatus.CONDITIONAL, + limitations=(limitation,), + ) + + assert graph.limitations == (limitation,) + with pytest.raises(ValueError, match="conditional/unavailable require"): + EndpointDependencyGraph(status=DependencyGraphStatus.CONDITIONAL) + + def test_dependency_graph_rejects_malformed_tree_and_strength_payloads(self) -> None: + def occurrence(path: tuple[int, ...], order: int) -> EndpointDependencyOccurrence: + return EndpointDependencyOccurrence( + index_path=path, + parent_path=path[:-1], + depth=len(path), + order=order, + declaration_scope=DependencyDeclarationScope.NESTED, + declaration_kind=DependencyDeclarationKind.DEPENDS_OR_SECURITY, + callable_kind=DependencyCallableKind.FUNCTION, + resolution_status=DependencyResolutionStatus.ESTABLISHED, + display_name="dependency", + module="main", + qualname="dependency", + source_span=DependencySourceSpan( + file_path=Path("/app/main.py"), start_line=1, end_line=2 + ), + ) + + malformed = [ + (occurrence((-1,), 0), "bounded nonnegative"), + (occurrence((0, 0), 0), "parent must exist earlier"), + (occurrence((1,), 0), "indexes must be contiguous"), + ] + for item, message in malformed: + with pytest.raises(ValueError, match=message): + EndpointDependencyGraph( + status=DependencyGraphStatus.ESTABLISHED, + occurrences=(item,), + ) + + with pytest.raises(ValueError, match="depth-first preorder"): + EndpointDependencyGraph( + status=DependencyGraphStatus.ESTABLISHED, + occurrences=( + occurrence((0,), 0), + occurrence((1,), 1), + occurrence((0, 0), 2), + ), + ) + + uncertain = occurrence((0,), 0).model_copy( + update={"resolution_status": DependencyResolutionStatus.CONDITIONAL} + ) + with pytest.raises(ValueError, match="only established occurrences"): + EndpointDependencyGraph( + status=DependencyGraphStatus.ESTABLISHED, + occurrences=(uncertain,), + ) + + orphan_payload = { + "status": "established", + "occurrences": [occurrence((0, 0), 0).model_dump(mode="json")], + } + with pytest.raises(ValueError, match="parent must exist earlier"): + EndpointDependencyGraph.model_validate_json(json.dumps(orphan_payload)) + def test_endpoint_identifier(self) -> None: """Test the endpoint identifier property.""" handler = HandlerInfo( diff --git a/tests/unit/test_mypy_correctness.py b/tests/unit/test_mypy_correctness.py index 9003a0e..608d7f3 100644 --- a/tests/unit/test_mypy_correctness.py +++ b/tests/unit/test_mypy_correctness.py @@ -14,7 +14,14 @@ ) from fastapi_endpoint_detector.config import AnalysisConfig, Config, ParserConfig from fastapi_endpoint_detector.models.diff import ChangeType, DiffFile -from fastapi_endpoint_detector.models.endpoint import Endpoint, EndpointMethod, HandlerInfo +from fastapi_endpoint_detector.models.endpoint import ( + DependencyGraphStatus, + DependencySourceSpan, + Endpoint, + EndpointMethod, + HandlerInfo, +) +from fastapi_endpoint_detector.parser.fastapi_extractor import FastAPIExtractor def _endpoint(main: Path) -> Endpoint: @@ -348,6 +355,221 @@ def test_annotated_dependency_alias_traces_provider_and_nested_calls(tmp_path: P assert deps.references_symbol_at_line("dependencies.py", 4) is not None +def test_runtime_graph_seeds_app_dependency_and_nested_helper_by_qualified_source( + tmp_path: Path, +) -> None: + first = tmp_path / "first.py" + first.write_text( + "def helper() -> int:\n return 1\n\ndef auth() -> int:\n return helper()\n" + ) + second = tmp_path / "second.py" + second.write_text("def auth() -> int:\n return 2\n") + main = tmp_path / "main.py" + main.write_text( + "from fastapi import Depends, FastAPI\n" + "from first import auth\n\n" + "app = FastAPI(dependencies=[Depends(auth)])\n\n" + "@app.get('/runtime-seed')\n" + "def handler() -> int:\n" + " return 1\n" + ) + endpoint = FastAPIExtractor(main).extract_endpoints()[0] + + deps = MypyAnalyzer(tmp_path, max_depth=3).analyze_endpoint(endpoint) + + assert deps.references_symbol_at_line("first.py", 4) is not None + assert deps.references_symbol_at_line("first.py", 1) is not None + assert deps.references_symbol_at_line("second.py", 1) is None + + assert endpoint.dependency_graph is not None + mismatched_occurrences = tuple( + occurrence.model_copy( + update={ + "source_span": DependencySourceSpan( + file_path=second, + start_line=1, + end_line=2, + ) + } + ) + for occurrence in endpoint.dependency_graph.occurrences + ) + mismatched = endpoint.model_copy( + update={ + "dependency_graph": endpoint.dependency_graph.model_copy( + update={"occurrences": mismatched_occurrences} + ) + } + ) + mismatched_deps = MypyAnalyzer(tmp_path, max_depth=3).analyze_endpoint(mismatched) + assert mismatched_deps.references_symbol_at_line("first.py", 4) is None + assert mismatched_deps.references_symbol_at_line("first.py", 1) is None + + same_file_wrong_span = endpoint.model_copy( + update={ + "dependency_graph": endpoint.dependency_graph.model_copy( + update={ + "occurrences": tuple( + occurrence.model_copy( + update={ + "source_span": DependencySourceSpan( + file_path=first, + start_line=1, + end_line=2, + ) + } + ) + for occurrence in endpoint.dependency_graph.occurrences + ) + } + ) + } + ) + same_file_deps = MypyAnalyzer(tmp_path, max_depth=3).analyze_endpoint(same_file_wrong_span) + assert same_file_deps.references_symbol_at_line("first.py", 4) is None + + +def test_suppressing_wraps_decorators_are_physical_and_unseeded(tmp_path: Path) -> None: + (tmp_path / "decorators.py").write_text( + "from functools import wraps\n\n" + "def suppress(func):\n" + " @wraps(func)\n" + " def wrapper():\n" + " return 0\n" + " return wrapper\n" + ) + external = tmp_path / "external.py" + external.write_text( + "from decorators import suppress\n" + "from fastapi import Depends, FastAPI\n\n" + "def helper() -> int:\n return 1\n\n" + "@suppress\n" + "def auth() -> int:\n return helper()\n\n" + "app = FastAPI(dependencies=[Depends(auth)])\n\n" + "@app.get('/')\n" + "def handler() -> int:\n return 1\n" + ) + endpoint = FastAPIExtractor(external).extract_endpoints()[0] + occurrence = endpoint.dependency_graph.occurrences[0] # type: ignore[union-attr] + # Python 3.11+ exposes code.co_qualname and can attest the physical local + # wrapper. Python 3.10 cannot recover that qualified identity safely once + # functools.wraps copied the decorated function metadata, so it abstains. + assert (occurrence.module, occurrence.qualname) in { + ("decorators", "suppress..wrapper"), + (None, None), + } + assert (occurrence.module, occurrence.qualname) != ("external", "auth") + assert occurrence.source_span is not None + assert occurrence.source_span.file_path == tmp_path / "decorators.py" + assert endpoint.dependency_graph is not None + assert endpoint.dependency_graph.status == DependencyGraphStatus.CONDITIONAL + deps = MypyAnalyzer(tmp_path, max_depth=3).analyze_endpoint(endpoint) + assert deps.references_symbol_at_line("external.py", 4) is None + + same_file = tmp_path / "same_file.py" + same_file.write_text( + "from functools import wraps\n" + "from fastapi import Depends, FastAPI\n\n" + "def suppress(func):\n" + " @wraps(func)\n" + " def wrapper():\n" + " return 0\n" + " return wrapper\n\n" + "def helper() -> int:\n return 1\n\n" + "@suppress\n" + "def auth() -> int:\n return helper()\n\n" + "app = FastAPI(dependencies=[Depends(auth)])\n\n" + "@app.get('/')\n" + "def handler() -> int:\n return 1\n" + ) + same_endpoint = FastAPIExtractor(same_file).extract_endpoints()[0] + same_deps = MypyAnalyzer(tmp_path, max_depth=3).analyze_endpoint(same_endpoint) + assert same_deps.references_symbol_at_line("same_file.py", 10) is None + assert same_deps.references_symbol_at_line("same_file.py", 14) is None + + +def test_runtime_seeds_exact_children_under_unseedable_parents(tmp_path: Path) -> None: + main = tmp_path / "parents.py" + main.write_text( + "from functools import partial\n" + "from fastapi import Depends, FastAPI\n\n" + "def partial_leaf() -> int:\n return 1\n\n" + "def parent(value: int = Depends(partial_leaf)) -> int:\n return value\n\n" + "def instance_leaf() -> int:\n return 2\n\n" + "class Provider:\n" + " def __call__(self, value: int = Depends(instance_leaf)) -> int:\n" + " return value\n\n" + "provider = Provider()\n" + "app = FastAPI(dependencies=[Depends(partial(parent)), Depends(provider)])\n\n" + "@app.get('/')\n" + "def handler() -> int:\n return 1\n" + ) + endpoint = FastAPIExtractor(main).extract_endpoints()[0] + deps = MypyAnalyzer(tmp_path, max_depth=3).analyze_endpoint(endpoint) + assert deps.references_symbol_at_line("parents.py", 4) is not None + assert deps.references_symbol_at_line("parents.py", 10) is not None + + +def test_bound_method_seed_preserves_exact_class_and_canonical_module(tmp_path: Path) -> None: + other = tmp_path / "other_methods.py" + other.write_text( + "def other_helper() -> int:\n return 3\n\n" + "class First:\n" + " def auth(self) -> int:\n" + " return other_helper()\n" + ) + main = tmp_path / "methods.py" + main.write_text( + "from fastapi import Depends, FastAPI\n\n" + "def first_helper() -> int:\n return 1\n\n" + "def second_helper() -> int:\n return 2\n\n" + "class First:\n" + " def auth(self) -> int:\n" + " return first_helper()\n\n" + "class Second:\n" + " def auth(self) -> int:\n" + " return second_helper()\n\n" + "first = First()\n" + "app = FastAPI(dependencies=[Depends(first.auth)])\n\n" + "@app.get('/')\n" + "def handler() -> int:\n return 1\n" + ) + endpoint = FastAPIExtractor(main).extract_endpoints()[0] + deps = MypyAnalyzer(tmp_path, max_depth=3).analyze_endpoint(endpoint) + assert deps.references_symbol_at_line("methods.py", 3) is not None + assert deps.references_symbol_at_line("methods.py", 10) is not None + assert deps.references_symbol_at_line("methods.py", 6) is None + assert deps.references_symbol_at_line("methods.py", 14) is None + assert deps.references_symbol_at_line("other_methods.py", 1) is None + assert deps.references_symbol_at_line("other_methods.py", 4) is None + + +def test_dependency_graph_hash_prevents_stale_endpoint_cache_reuse(tmp_path: Path) -> None: + main = tmp_path / "cached.py" + main.write_text( + "from fastapi import Depends, FastAPI\n\n" + "def helper() -> int:\n return 1\n\n" + "def auth() -> int:\n return helper()\n\n" + "app = FastAPI(dependencies=[Depends(auth)])\n\n" + "@app.get('/')\n" + "def handler() -> int:\n return 1\n" + ) + endpoint = FastAPIExtractor(main).extract_endpoints()[0] + legacy = endpoint.model_copy(update={"dependency_graph": None}) + cache_path = tmp_path / "analysis-cache.json" + first = MypyAnalyzer(tmp_path, max_depth=3) + first.set_cache_path(cache_path) + first.analyze_endpoints([legacy]) + + second = MypyAnalyzer(tmp_path, max_depth=3) + second.set_cache_path(cache_path) + results = second.analyze_endpoints([endpoint]) + deps = results[second._endpoint_key(endpoint)] + assert deps.references_symbol_at_line("cached.py", 3) is not None + assert deps.references_symbol_at_line("cached.py", 6) is not None + assert second._endpoint_key(legacy) != second._endpoint_key(endpoint) + + def test_imported_global_is_referenced_at_definition(tmp_path: Path) -> None: config = tmp_path / "config.py" config.write_text("DEFAULT = {'used': 1}\nUNRELATED = 2\n")