diff --git a/README.md b/README.md index d61b3bb..7cc181f 100644 --- a/README.md +++ b/README.md @@ -103,11 +103,14 @@ fastapi-endpoint-detector list --app path/to/main.py --format html -o endpoints. ### 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. +JSON and YAML inventories and analysis reports use schema version 4. Each endpoint may +include two additive nullable evidence fields. `dependency_graph` is schema-v1 bounded +runtime evidence for the declared FastAPI dependency tree. `native_provenance` is +schema-v1 secure-AST evidence for the snapshot side, selected root, exact decorator or +imperative registration span, app/router constructor chain, and ordered include/mount +chain. Source columns use Python AST offsets and end coordinates are exclusive. Trusted +runtime extraction transports dependency evidence through the private protocol-v2 +worker; it does not guess source-level route assembly occurrences. ## Commands diff --git a/src/fastapi_endpoint_detector/analyzer/change_mapper.py b/src/fastapi_endpoint_detector/analyzer/change_mapper.py index 0399cae..f5a6731 100644 --- a/src/fastapi_endpoint_detector/analyzer/change_mapper.py +++ b/src/fastapi_endpoint_detector/analyzer/change_mapper.py @@ -42,6 +42,7 @@ Endpoint, EndpointDiscoveryStatus, EndpointInventory, + SnapshotSide, ) from fastapi_endpoint_detector.models.report import ( AffectedEndpoint, @@ -98,14 +99,33 @@ } -def _endpoint_result_key(endpoint: Endpoint) -> tuple[str, str, int, str, str]: +EndpointResultKey = tuple[str, str, int, str, str, str] + + +def _endpoint_result_key(endpoint: Endpoint) -> EndpointResultKey: handler = endpoint.handler + provenance = endpoint.native_provenance + occurrence = "" + if provenance is not None: + registration = provenance.registration + occurrence = ":".join( + ( + provenance.side.value, + str(registration.source_span.file_path.resolve()), + str(registration.occurrence_order), + *( + f"{edge.source_span.file_path.resolve()}@{edge.occurrence_order}" + for edge in provenance.assembly_chain + ), + ) + ) return ( endpoint.identifier, str(handler.file_path.resolve()), handler.line_number, handler.name, handler.module, + occurrence, ) @@ -207,7 +227,7 @@ def materialize(self) -> AffectedEndpoint: def _merge_affected( - accumulated: dict[tuple[str, str, int, str, str], _AffectedAccumulator], + accumulated: dict[EndpointResultKey, _AffectedAccumulator], candidate: AffectedEndpoint, ) -> None: if ( @@ -572,6 +592,7 @@ def baseline_registry(self) -> EndpointRegistry: app_variable=self.app_variable, app_entry=self.app_entry, bootstrap_entry=self.bootstrap_entry, + snapshot_side=SnapshotSide.BASELINE, ) self._baseline_registry = EndpointRegistry() native = extractor.extract_inventory() @@ -878,14 +899,55 @@ def _analyze_diff_file( Tuple of (affected endpoints, processed added lines, processed removed lines). Processed lines are those that were matched to any endpoint. """ - affected: dict[tuple[str, str, int, str, str], _AffectedAccumulator] = {} + affected: dict[EndpointResultKey, _AffectedAccumulator] = {} processed_added_lines: set[int] = set() processed_removed_lines: set[int] = set() # Get changed lines added_lines, removed_lines = DiffParser.get_changed_line_numbers(diff_file) - # Find endpoints in the changed file + # Native route registrations and exact include/mount/object occurrences own + # their materialized descendants. Only target additions are queried here: + # removed coordinates require the explicit baseline path handled by SCIP. + for endpoint, kinds, overlap in self.registry.get_structural_overlaps( + diff_file.path, set(added_lines) + ): + matched_kinds = ", ".join(kinds) + changed_line = min(overlap) + _merge_affected( + affected, + AffectedEndpoint( + endpoint=endpoint, + confidence=ConfidenceLevel.HIGH, + reason=( + f"Native route assembly occurrence modified ({matched_kinds}) " + f"in {diff_file.path}" + ), + dependency_chain=[str(diff_file.path), *kinds], + changed_files=[str(diff_file.path)], + effect_evidence=[ + EffectEvidence( + producer=EvidenceProducer.STRUCTURAL, + status=EvidenceStatus.ESTABLISHED, + effect=ChangeEffectKind.ROUTE_ASSEMBLY, + channel=ImpactChannel.UNKNOWN, + disposition=EffectDisposition.INTERNAL_EFFECT, + summary=( + "Changed source overlaps exact secure-AST route assembly " + "provenance for this endpoint occurrence." + ), + changed_location=CodeReference( + file_path=str(diff_file.path), + line_number=changed_line, + symbol=matched_kinds, + ), + ) + ], + ), + ) + processed_added_lines.update(overlap) + + # Find endpoints whose handlers are defined in the changed file. file_endpoints = self.registry.get_by_file(diff_file.path) # Check for direct handler changes @@ -948,7 +1010,7 @@ def _analyze_with_scip( if has_removed: self.baseline_scip_analyzer.ensure_index(force=not self.use_cache) - affected: dict[tuple[str, str, int, str, str], _AffectedAccumulator] = {} + affected: dict[EndpointResultKey, _AffectedAccumulator] = {} orphan_evidence: dict[str, _OrphanAccumulator] = {} target_root = self.app_path.parent if self.app_path.is_file() else self.app_path baseline_root = ( @@ -969,6 +1031,55 @@ def target_equivalent(endpoint: Endpoint) -> Endpoint: ] return matches[0] if len(matches) == 1 else endpoint + def analyze_structural_side( + registry: EndpointRegistry, + file_path: Path, + lines: list[int], + side: str, + ) -> set[int]: + processed: set[int] = set() + for discovered, kinds, overlap in registry.get_structural_overlaps( + file_path, set(lines) + ): + # Structural evidence remains attached to its source snapshot. + # Lifecycle reconciliation is deliberately deferred rather than + # replacing a baseline occurrence by a same-identifier target. + endpoint = discovered + matched_kinds = ", ".join(kinds) + _merge_affected( + affected, + AffectedEndpoint( + endpoint=endpoint, + confidence=ConfidenceLevel.HIGH, + reason=( + f"Secure-AST {side} route assembly occurrence modified " + f"({matched_kinds}) in {file_path}" + ), + dependency_chain=[str(file_path), *kinds], + changed_files=[str(file_path)], + effect_evidence=[ + EffectEvidence( + producer=EvidenceProducer.STRUCTURAL, + status=EvidenceStatus.ESTABLISHED, + effect=ChangeEffectKind.ROUTE_ASSEMBLY, + channel=ImpactChannel.UNKNOWN, + disposition=EffectDisposition.INTERNAL_EFFECT, + summary=( + f"Changed {side} source overlaps exact secure-AST route " + "assembly provenance for this endpoint occurrence." + ), + changed_location=CodeReference( + file_path=str(file_path), + line_number=min(overlap), + symbol=matched_kinds, + ), + ) + ], + ), + ) + processed.update(overlap) + return processed + def analyze_side( analyzer: SCIPAnalyzer, registry: EndpointRegistry, @@ -1060,25 +1171,45 @@ def analyze_side( added_lines, removed_lines = DiffParser.get_changed_line_numbers(diff_file) processed_added: set[int] = set() if diff_file.path.suffix == ".py" and added_lines: - processed_added = analyze_side( - self.scip_analyzer, - self.registry, - target_root, - diff_file.path, - added_lines, - "target", + processed_added.update( + analyze_structural_side( + self.registry, + diff_file.path, + added_lines, + "target", + ) + ) + processed_added.update( + analyze_side( + self.scip_analyzer, + self.registry, + target_root, + diff_file.path, + added_lines, + "target", + ) ) processed_removed: set[int] = set() source_path = diff_file.source_path or diff_file.path if removed_lines and source_path.suffix == ".py": assert baseline_root is not None - processed_removed = analyze_side( - self.baseline_scip_analyzer, - self.baseline_registry, - baseline_root, - source_path, - removed_lines, - "baseline", + processed_removed.update( + analyze_structural_side( + self.baseline_registry, + source_path, + removed_lines, + "baseline", + ) + ) + processed_removed.update( + analyze_side( + self.baseline_scip_analyzer, + self.baseline_registry, + baseline_root, + source_path, + removed_lines, + "baseline", + ) ) reason = "Changed lines did not resolve through SCIP to a registered endpoint" @@ -1427,7 +1558,7 @@ def report_progress(current: int, total: int, desc: str) -> None: # Analyze each Python file report_progress(70, 100, f"Checking {len(python_files)} changed files...") - all_affected: dict[tuple[str, str, int, str, str], _AffectedAccumulator] = {} + all_affected: dict[EndpointResultKey, _AffectedAccumulator] = {} orphan_evidence: dict[str, _OrphanAccumulator] = {} for i, diff_file in enumerate(python_files): diff --git a/src/fastapi_endpoint_detector/analyzer/endpoint_registry.py b/src/fastapi_endpoint_detector/analyzer/endpoint_registry.py index e99234c..990e29a 100644 --- a/src/fastapi_endpoint_detector/analyzer/endpoint_registry.py +++ b/src/fastapi_endpoint_detector/analyzer/endpoint_registry.py @@ -21,6 +21,7 @@ def __init__(self) -> None: self._by_path: dict[str, list[Endpoint]] = {} self._by_file: dict[Path, list[Endpoint]] = {} self._by_module: dict[str, list[Endpoint]] = {} + self._by_structural_file: dict[Path, list[tuple[Endpoint, str, int, int]]] = {} def register(self, endpoint: Endpoint) -> None: """ @@ -48,6 +49,34 @@ def register(self, endpoint: Endpoint) -> None: self._by_module[module] = [] self._by_module[module].append(endpoint) + provenance = endpoint.native_provenance + if provenance is not None: + occurrences = [ + ( + f"route_{provenance.registration.operation}", + provenance.registration.source_span, + ), + *( + (f"object_{item.object_kind}", item.source_span) + for item in provenance.object_chain + ), + *( + (f"assembly_{item.operation}", item.source_span) + for item in provenance.assembly_chain + ), + ] + if provenance.root.bootstrap_span is not None: + occurrences.append(("bootstrap", provenance.root.bootstrap_span)) + seen: set[tuple[str, Path, int, int]] = set() + for kind, span in occurrences: + identity = (kind, span.file_path, span.start_line, span.end_line) + if identity in seen: + continue + seen.add(identity) + self._by_structural_file.setdefault(span.file_path, []).append( + (endpoint, kind, span.start_line, span.end_line) + ) + def register_many(self, endpoints: list[Endpoint]) -> None: """ Register multiple endpoints. @@ -122,6 +151,48 @@ def _has_path_suffix(path: tuple[str, ...], suffix: tuple[str, ...]) -> bool: """Return whether *suffix* identifies whole trailing path components.""" return bool(suffix) and len(path) >= len(suffix) and path[-len(suffix) :] == suffix + def get_structural_overlaps( + self, + file_path: Path | str, + changed_lines: set[int], + ) -> list[tuple[Endpoint, tuple[str, ...], set[int]]]: + """Return exact native assembly occurrences intersecting target-side lines.""" + if not changed_lines: + return [] + query = Path(file_path) + try: + resolved = query.resolve() + except OSError: + resolved = query + if resolved in self._by_structural_file: + buckets = [self._by_structural_file[resolved]] + elif query in self._by_structural_file: + buckets = [self._by_structural_file[query]] + else: + query_parts = self._path_parts(query) + buckets = [ + occurrences + for registered_path, occurrences in self._by_structural_file.items() + if self._has_path_suffix(self._path_parts(registered_path), query_parts) + ] + if len(buckets) != 1: + return [] + matches: dict[int, tuple[Endpoint, set[str], set[int]]] = {} + for endpoint, kind, start_line, end_line in buckets[0]: + overlap = {line for line in changed_lines if start_line <= line <= end_line} + if not overlap: + continue + key = id(endpoint) + current = matches.get(key) + if current is None: + matches[key] = (endpoint, {kind}, overlap) + else: + current[1].add(kind) + current[2].update(overlap) + return [ + (endpoint, tuple(sorted(kinds)), lines) for endpoint, kinds, lines in matches.values() + ] + def get_by_module(self, module: str) -> list[Endpoint]: """ Get endpoints defined in a specific module. diff --git a/src/fastapi_endpoint_detector/models/__init__.py b/src/fastapi_endpoint_detector/models/__init__.py index 980e2c8..6f4c023 100644 --- a/src/fastapi_endpoint_detector/models/__init__.py +++ b/src/fastapi_endpoint_detector/models/__init__.py @@ -60,7 +60,17 @@ EndpointMethod, HandlerInfo, InventoryStatus, + NativeAssemblyMode, + NativeRegistrationKind, + NativeRootSelectionKind, + NativeRouteAssemblyEdgeEvidence, + NativeRouteObjectEvidence, + NativeRouteProvenance, + NativeRouteRegistrationEvidence, + NativeRouteRootEvidence, + NativeSourceSpan, RouteActivationEvidence, + SnapshotSide, SurfaceRegistrationEvidence, ) from fastapi_endpoint_detector.models.report import ( @@ -131,7 +141,17 @@ "EndpointMethod", "HandlerInfo", "InventoryStatus", + "NativeAssemblyMode", + "NativeRegistrationKind", + "NativeRootSelectionKind", + "NativeRouteAssemblyEdgeEvidence", + "NativeRouteObjectEvidence", + "NativeRouteProvenance", + "NativeRouteRegistrationEvidence", + "NativeRouteRootEvidence", + "NativeSourceSpan", "RouteActivationEvidence", + "SnapshotSide", "SurfaceRegistrationEvidence", # Dependency models "Dependency", diff --git a/src/fastapi_endpoint_detector/models/endpoint.py b/src/fastapi_endpoint_detector/models/endpoint.py index 857f61b..2586982 100644 --- a/src/fastapi_endpoint_detector/models/endpoint.py +++ b/src/fastapi_endpoint_detector/models/endpoint.py @@ -48,6 +48,36 @@ class EndpointDiscoveryStatus(str, Enum): CONDITIONAL = "conditional" +class SnapshotSide(str, Enum): + """Source snapshot that produced one native route occurrence.""" + + TARGET = "target" + BASELINE = "baseline" + + +class NativeRegistrationKind(str, Enum): + """Python construct that installed one native route.""" + + DECORATOR = "decorator" + IMPERATIVE = "imperative" + + +class NativeAssemblyMode(str, Enum): + """Composition semantics of one native route-assembly edge.""" + + COPY = "copy" + LIVE = "live" + + +class NativeRootSelectionKind(str, Enum): + """How secure discovery selected the root that owns a native route.""" + + APP_VARIABLE = "app_variable" + ROUTER_VARIABLE = "router_variable" + APP_ENTRY_OBJECT = "app_entry_object" + APP_ENTRY_FACTORY = "app_entry_factory" + + class DependencyGraphStatus(str, Enum): """Completeness of runtime dependency-graph evidence.""" @@ -277,6 +307,155 @@ class Config: frozen = True +class NativeSourceSpan(BaseModel): + """Column-complete Python AST span; end coordinates are exclusive.""" + + file_path: Path + start_line: int = Field(ge=1) + start_column: int = Field(ge=0) + end_line: int = Field(ge=1) + end_column: int = Field(ge=0) + + @model_validator(mode="after") + def validate_span(self) -> "NativeSourceSpan": + if (self.end_line, self.end_column) < (self.start_line, self.start_column): + raise ValueError("native source span end must not precede start") + return self + + def overlaps_lines(self, lines: set[int]) -> set[int]: + """Return changed lines intersecting this source occurrence.""" + return {line for line in lines if self.start_line <= line <= self.end_line} + + class Config: + frozen = True + + +class NativeRouteRegistrationEvidence(BaseModel): + """Exact source occurrence that registered one native endpoint.""" + + side: SnapshotSide + kind: NativeRegistrationKind + operation: str = Field(pattern=r"^[a-z][a-z0-9_]*$", max_length=64) + owner_module: str = Field(min_length=1, max_length=512) + owner_symbol: str = Field(min_length=1, max_length=2048) + occurrence_order: int = Field(ge=0) + source_span: NativeSourceSpan + + class Config: + frozen = True + + +class NativeRouteAssemblyEdgeEvidence(BaseModel): + """One ordered include or mount occurrence on a public route path.""" + + side: SnapshotSide + operation: Literal["include_router", "mount"] + mode: NativeAssemblyMode + parent_module: str = Field(min_length=1, max_length=512) + parent_symbol: str = Field(min_length=1, max_length=2048) + child_module: str = Field(min_length=1, max_length=512) + child_symbol: str = Field(min_length=1, max_length=2048) + occurrence_order: int = Field(ge=0) + resolved_prefix: str = Field(max_length=4096) + source_span: NativeSourceSpan + + class Config: + frozen = True + + +class NativeRouteObjectEvidence(BaseModel): + """App/router constructor occurrence whose prefix contributes to a route.""" + + side: SnapshotSide + object_kind: Literal["app", "router"] + module: str = Field(min_length=1, max_length=512) + symbol: str = Field(min_length=1, max_length=2048) + resolved_prefix: str | None = Field(default=None, max_length=4096) + source_span: NativeSourceSpan + + class Config: + frozen = True + + +class NativeRouteRootEvidence(BaseModel): + """Selected secure-AST root for one native endpoint occurrence.""" + + side: SnapshotSide + selection_kind: NativeRootSelectionKind + module: str = Field(min_length=1, max_length=512) + symbol: str = Field(min_length=1, max_length=2048) + source_span: NativeSourceSpan | None = None + bootstrap_span: NativeSourceSpan | None = None + + class Config: + frozen = True + + +class NativeRouteProvenance(BaseModel): + """Immutable route-registration and assembly chain decided by secure AST.""" + + schema_version: Literal[1] = 1 + side: SnapshotSide + root: NativeRouteRootEvidence + registration: NativeRouteRegistrationEvidence + object_chain: tuple[NativeRouteObjectEvidence, ...] = Field(min_length=1, max_length=256) + assembly_chain: tuple[NativeRouteAssemblyEdgeEvidence, ...] = Field(default=(), max_length=256) + + @model_validator(mode="after") + def validate_chain(self) -> "NativeRouteProvenance": + evidence = [ + self.root.side, + self.registration.side, + *(item.side for item in self.object_chain), + *(item.side for item in self.assembly_chain), + ] + if any(side != self.side for side in evidence): + raise ValueError("native route provenance cannot mix snapshot sides") + if len(self.object_chain) != len(self.assembly_chain) + 1: + raise ValueError("native object chain must contain one object per assembly hop") + root_object = self.object_chain[0] + if (self.root.module, self.root.symbol) != ( + root_object.module, + root_object.symbol, + ): + raise ValueError("native route root must be the first object occurrence") + expected_root_kind = ( + "router" + if self.root.selection_kind == NativeRootSelectionKind.ROUTER_VARIABLE + else "app" + ) + if root_object.object_kind != expected_root_kind: + raise ValueError("native route root selection must match the first object role") + for index, edge in enumerate(self.assembly_chain): + parent = self.object_chain[index] + child = self.object_chain[index + 1] + if (edge.parent_module, edge.parent_symbol) != (parent.module, parent.symbol) or ( + edge.child_module, + edge.child_symbol, + ) != (child.module, child.symbol): + raise ValueError("native assembly edge must connect adjacent object occurrences") + expected_mode = ( + NativeAssemblyMode.COPY + if edge.operation == "include_router" + else NativeAssemblyMode.LIVE + ) + if edge.mode != expected_mode: + raise ValueError("native assembly operation must use its declared composition mode") + expected_child_kind = "router" if edge.operation == "include_router" else "app" + if child.object_kind != expected_child_kind: + raise ValueError("native assembly operation must target a compatible object role") + owner = self.object_chain[-1] + if (self.registration.owner_module, self.registration.owner_symbol) != ( + owner.module, + owner.symbol, + ): + raise ValueError("native registration owner must be the final object occurrence") + return self + + class Config: + frozen = True + + class SurfaceRegistrationEvidence(BaseModel): """Data-only registration and contract provenance for a custom surface.""" @@ -346,6 +525,10 @@ class Endpoint(BaseModel): default=None, description="Authoritative declared dependency graph; None means not collected", ) + native_provenance: NativeRouteProvenance | None = Field( + default=None, + description="Secure-AST native registration and assembly evidence when collected", + ) discovery_status: EndpointDiscoveryStatus = EndpointDiscoveryStatus.ESTABLISHED discovery_conditions: tuple[EndpointDiscoveryCondition, ...] = () surface: SurfaceRegistrationEvidence | None = None @@ -363,6 +546,8 @@ def validate_discovery_provenance(self) -> "Endpoint": raise ValueError("CUSTOM must be the only method on a custom surface") if custom != (self.surface is not None): raise ValueError("custom endpoints require CUSTOM method and surface provenance") + if custom and self.native_provenance is not None: + raise ValueError("custom endpoints cannot carry native route provenance") if self.activation is not None and ( custom or self.surface is not None diff --git a/src/fastapi_endpoint_detector/models/report.py b/src/fastapi_endpoint_detector/models/report.py index 9a43e9a..ac427a1 100644 --- a/src/fastapi_endpoint_detector/models/report.py +++ b/src/fastapi_endpoint_detector/models/report.py @@ -57,6 +57,7 @@ class EvidenceProducer(str, Enum): """Analyzer that produced an evidence record.""" DIRECT = "direct" + STRUCTURAL = "structural" MYPY = "mypy" SCIP = "scip" DATA_FLOW = "data_flow" @@ -76,6 +77,7 @@ class ChangeEffectKind(str, Enum): """Semantic shape of a source change.""" HANDLER_IMPLEMENTATION = "handler_implementation" + ROUTE_ASSEMBLY = "route_assembly" DEFENSIVE_COPY_ADDED = "defensive_copy_added" ARGUMENT_MUTATION_ISOLATED = "argument_mutation_isolated" RETURN_VALUE_CHANGED = "return_value_changed" diff --git a/src/fastapi_endpoint_detector/output/json_output.py b/src/fastapi_endpoint_detector/output/json_output.py index 0ce4973..d56a6e8 100644 --- a/src/fastapi_endpoint_detector/output/json_output.py +++ b/src/fastapi_endpoint_detector/output/json_output.py @@ -45,6 +45,11 @@ def _endpoint_to_dict(self, endpoint: Endpoint) -> dict[str, Any]: if endpoint.dependency_graph is not None else None ), + "native_provenance": ( + endpoint.native_provenance.model_dump(mode="json") + if endpoint.native_provenance is not None + else None + ), "discovery_status": endpoint.discovery_status.value, "discovery_conditions": [ condition.model_dump(mode="json") for condition in endpoint.discovery_conditions @@ -88,7 +93,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, + "schema_version": 4, "timestamp": report.timestamp.isoformat(), "app_path": report.app_path, "diff_source": report.diff_source, @@ -154,7 +159,7 @@ def format(self, report: AnalysisReport) -> str: def format_inventory(self, inventory: EndpointInventory) -> str: """Format endpoints with whole-inventory strength metadata.""" data = { - "schema_version": 3, + "schema_version": 4, "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 99e87bf..67965e6 100644 --- a/src/fastapi_endpoint_detector/output/yaml_output.py +++ b/src/fastapi_endpoint_detector/output/yaml_output.py @@ -37,6 +37,11 @@ def _endpoint_to_dict(self, endpoint: Endpoint) -> dict[str, Any]: if endpoint.dependency_graph is not None else None ), + "native_provenance": ( + endpoint.native_provenance.model_dump(mode="json") + if endpoint.native_provenance is not None + else None + ), "discovery_status": endpoint.discovery_status.value, "discovery_conditions": [ condition.model_dump(mode="json") for condition in endpoint.discovery_conditions @@ -80,7 +85,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, + "schema_version": 4, "timestamp": report.timestamp.isoformat(), "app_path": report.app_path, "diff_source": report.diff_source, @@ -146,7 +151,7 @@ def format(self, report: AnalysisReport) -> str: def format_inventory(self, inventory: EndpointInventory) -> str: """Format endpoints with whole-inventory strength metadata.""" data = { - "schema_version": 3, + "schema_version": 4, "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/secure_ast_extractor.py b/src/fastapi_endpoint_detector/parser/secure_ast_extractor.py index 5e1b177..0ac8922 100644 --- a/src/fastapi_endpoint_detector/parser/secure_ast_extractor.py +++ b/src/fastapi_endpoint_detector/parser/secure_ast_extractor.py @@ -20,6 +20,16 @@ EndpointMethod, HandlerInfo, InventoryStatus, + NativeAssemblyMode, + NativeRegistrationKind, + NativeRootSelectionKind, + NativeRouteAssemblyEdgeEvidence, + NativeRouteObjectEvidence, + NativeRouteProvenance, + NativeRouteRegistrationEvidence, + NativeRouteRootEvidence, + NativeSourceSpan, + SnapshotSide, ) from fastapi_endpoint_detector.parser._static_evaluation import ( MAX_STATIC_STRING_CHARS, @@ -39,6 +49,24 @@ class SecureASTExtractorError(Exception): _EAGER_DEFINITION_MAX_WORK = 2_048 _EAGER_DEFINITION_MAX_DEPTH = 32 + +def _native_span(path: Path, node: ast.AST) -> NativeSourceSpan: + """Convert one parser occurrence to the public exclusive-end span contract.""" + start_line = getattr(node, "lineno", 1) + start_column = getattr(node, "col_offset", 0) + end_line = getattr(node, "end_lineno", None) or start_line + end_column = getattr(node, "end_col_offset", None) + if end_column is None: + end_column = start_column + 1 + return NativeSourceSpan( + file_path=path, + start_line=start_line, + start_column=start_column, + end_line=end_line, + end_column=end_column, + ) + + _HTTP_ROUTE_METADATA_KEYWORDS = frozenset( { "callbacks", @@ -123,6 +151,7 @@ class _Object: prefix: str | None line: int discovery_conditions: tuple[EndpointDiscoveryCondition, ...] = () + source_span: NativeSourceSpan | None = None def _uses_router_receiver(owner: _Object, receiver: ast.expr | None) -> bool: @@ -250,6 +279,9 @@ class _Route: handler: HandlerInfo line: int discovery_conditions: tuple[EndpointDiscoveryCondition, ...] = () + registration_kind: NativeRegistrationKind | None = None + operation: str | None = None + source_span: NativeSourceSpan | None = None @dataclass(frozen=True) @@ -261,6 +293,8 @@ class _Edge: child_cutoff: int | None limitation_cutoff: tuple[str, int] | None mode: CompositionMode + operation: Literal["include_router", "mount"] | None = None + source_span: NativeSourceSpan | None = None @dataclass(frozen=True) @@ -675,11 +709,13 @@ def __init__( app_variable: str = "app", app_entry: str | None = None, bootstrap_entry: str | None = None, + snapshot_side: SnapshotSide | Literal["target", "baseline"] = SnapshotSide.TARGET, ) -> None: self.app_path = app_path.resolve() self.app_variable = app_variable self.app_entry = app_entry self.bootstrap_entry = bootstrap_entry + self.snapshot_side = SnapshotSide(snapshot_side) self._app_entry_parts = self._parse_entry(app_entry, "--app-entry") self._bootstrap_entry_parts = self._parse_entry(bootstrap_entry, "--bootstrap-entry") @@ -878,6 +914,7 @@ def extract_inventory(self) -> EndpointInventory: # noqa: PLR0912, PLR0915 limitations=(limitation,), ) + bootstrap_span: NativeSourceSpan | None = None if self._bootstrap_entry_parts is not None: if len(set(roots)) != 1: raise SecureASTExtractorError( @@ -900,6 +937,7 @@ def extract_inventory(self) -> EndpointInventory: # noqa: PLR0912, PLR0915 or bootstrap_function.decorator_list ): raise SecureASTExtractorError("bootstrap entry must be synchronous and undecorated") + bootstrap_span = _native_span(bootstrap_module.path, bootstrap_function) selected_root = objects[roots[0]] if selected_root.kind != "app": raise SecureASTExtractorError("bootstrap_entry requires a FastAPI app root") @@ -941,6 +979,30 @@ def extract_inventory(self) -> EndpointInventory: # noqa: PLR0912, PLR0915 tuple(parse_failures) if self.app_path.is_dir() else () ) + def object_evidence(item: _Object) -> NativeRouteObjectEvidence: + source_span = item.source_span + if source_span is None: + source_line = item.line + if source_line >= _ORDER_SCALE and source_line < 2**31 - 1: + source_line //= _ORDER_SCALE + if source_line >= 2**31 - 1: + source_line = 1 + source_span = NativeSourceSpan( + file_path=modules[item.key[0]].path, + start_line=max(source_line, 1), + start_column=0, + end_line=max(source_line, 1), + end_column=1, + ) + return NativeRouteObjectEvidence( + side=self.snapshot_side, + object_kind=item.kind, + module=item.key[0], + symbol=item.key[1], + resolved_prefix=item.prefix, + source_span=source_span, + ) + def visit( owner: ObjectKey, inherited: str, @@ -948,11 +1010,16 @@ def visit( limitation_cutoff: tuple[str, int] | None, stack: frozenset[ObjectKey], inherited_conditions: tuple[EndpointDiscoveryCondition, ...], + root_evidence: NativeRouteRootEvidence, + object_chain: tuple[NativeRouteObjectEvidence, ...], + assembly_chain: tuple[NativeRouteAssemblyEdgeEvidence, ...], + provenance_available: bool, ) -> None: nonlocal inventory_limitations if owner in stack: return item = objects[owner] + current_object_chain = (*object_chain, object_evidence(item)) prefix = _join_paths(inherited, item.prefix) if item.prefix is not None else None object_conditions = _merge_discovery_conditions( inherited_conditions, @@ -1026,11 +1093,34 @@ def visit( ), ) continue + native_provenance = None + if ( + provenance_available + and route.registration_kind is not None + and route.operation is not None + and route.source_span is not None + ): + native_provenance = NativeRouteProvenance( + side=self.snapshot_side, + root=root_evidence, + registration=NativeRouteRegistrationEvidence( + side=self.snapshot_side, + kind=route.registration_kind, + operation=route.operation, + owner_module=route.owner[0], + owner_symbol=route.owner[1], + occurrence_order=route.line, + source_span=route.source_span, + ), + object_chain=current_object_chain, + assembly_chain=assembly_chain, + ) found.append( Endpoint( path=endpoint_path, methods=[EndpointMethod(method) for method in route.methods], handler=route.handler, + native_provenance=native_provenance, discovery_status=( EndpointDiscoveryStatus.CONDITIONAL if discovery_conditions @@ -1068,6 +1158,29 @@ def visit( ), ) continue + edge_available = ( + provenance_available + and edge.operation is not None + and edge.source_span is not None + ) + next_assembly_chain = assembly_chain + if edge_available: + assert edge.operation is not None and edge.source_span is not None + next_assembly_chain = ( + *assembly_chain, + NativeRouteAssemblyEdgeEvidence( + side=self.snapshot_side, + operation=edge.operation, + mode=NativeAssemblyMode(edge.mode), + parent_module=edge.parent[0], + parent_symbol=edge.parent[1], + child_module=edge.child[0], + child_symbol=edge.child[1], + occurrence_order=edge.line, + resolved_prefix=edge.prefix, + source_span=edge.source_span, + ), + ) visit( edge.child, edge_prefix, @@ -1078,10 +1191,31 @@ def visit( object_conditions, edge_effect_conditions, ), + root_evidence, + current_object_chain, + next_assembly_chain, + edge_available, ) for root in sorted(set(roots)): - visit(root, "", None, None, frozenset(), ()) + root_item = objects[root] + if explicit_object is not None: + selection_kind = NativeRootSelectionKind.APP_ENTRY_OBJECT + elif explicit_variable is not None: + selection_kind = NativeRootSelectionKind.APP_ENTRY_FACTORY + elif root_item.kind == "router": + selection_kind = NativeRootSelectionKind.ROUTER_VARIABLE + else: + selection_kind = NativeRootSelectionKind.APP_VARIABLE + root_evidence = NativeRouteRootEvidence( + side=self.snapshot_side, + selection_kind=selection_kind, + module=root[0], + symbol=root[1], + source_span=root_item.source_span, + bootstrap_span=bootstrap_span, + ) + visit(root, "", None, None, frozenset(), (), root_evidence, (), (), True) endpoints = sorted( found, key=lambda endpoint: ( @@ -1257,6 +1391,7 @@ def invalidate_additional_bindings( kind="app", prefix="", line=binding_line, + source_span=_native_span(module.path, node), ) module.objects.setdefault(conditional_name, []).append(item) for name in bound_names: @@ -1327,6 +1462,7 @@ def invalidate_additional_bindings( kind=constructor, prefix=prefix, line=node.lineno, + source_span=_native_span(module.path, node), ) module.objects.setdefault(assigned_name, []).append(item) elif isinstance(value, ast.Call): @@ -1530,6 +1666,7 @@ def _collect_factory_graphs( graph.root.prefix, call.line, graph.root.discovery_conditions, + graph.root.source_span, ) history.append(root) history.sort(key=lambda item: item.line) @@ -1749,6 +1886,7 @@ def snapshot_local_object(item: _Object, operation: ast.AST) -> _Object: item.prefix, call_line, item.discovery_conditions, + _native_span(module.path, operation), ) emitted_objects.append(snapshot) routes.extend( @@ -1759,6 +1897,9 @@ def snapshot_local_object(item: _Object, operation: ast.AST) -> _Object: route.handler, call_order, route.discovery_conditions, + route.registration_kind, + route.operation, + route.source_span, ) for route in list(routes) if route.owner == item.key @@ -1772,6 +1913,8 @@ def snapshot_local_object(item: _Object, operation: ast.AST) -> _Object: edge.child_cutoff, edge.limitation_cutoff, edge.mode, + edge.operation, + edge.source_span, ) for edge in list(edges) if edge.parent == item.key @@ -1990,7 +2133,14 @@ def exact_route_decorator(decorator: ast.expr) -> bool: ) if prefix is None: conditionalize(statement, "factory router prefix is unresolved") - item = _Object(key, variable, constructor, prefix, call_line) + item = _Object( + key, + variable, + constructor, + prefix, + call_line, + source_span=_native_span(module.path, statement), + ) local_objects[assigned] = item local_router_views.discard(assigned) local_created_keys.add(item.key) @@ -2150,6 +2300,9 @@ def resolve_nested_argument( methods, handler, call_order, + registration_kind=NativeRegistrationKind.DECORATOR, + operation=method, + source_span=_native_span(module.path, decorator), ) ) local_objects.pop(statement.name, None) @@ -2328,6 +2481,8 @@ def resolve_nested_argument( cutoff, (module.name, call_order), "copy", + "include_router", + _native_span(module.path, call), ) ) elif call_function.attr == "mount": @@ -2351,6 +2506,8 @@ def resolve_nested_argument( None, None, "live", + "mount", + _native_span(module.path, call), ) ) elif call_function.attr in { @@ -2384,6 +2541,9 @@ def resolve_nested_argument( methods, imperative_handler, call_order, + registration_kind=NativeRegistrationKind.IMPERATIVE, + operation=call_function.attr, + source_span=_native_span(module.path, call), ) ) elif call_function.attr not in { @@ -2413,6 +2573,7 @@ def resolve_nested_argument( root.prefix, call_line, combined_conditions, + root.source_span, ) emitted_objects = [item for item in emitted_objects if item.key != old_key] routes = [ @@ -2423,6 +2584,9 @@ def resolve_nested_argument( route.handler, route.line, route.discovery_conditions, + route.registration_kind, + route.operation, + route.source_span, ) for route in routes ] @@ -2435,6 +2599,8 @@ def resolve_nested_argument( edge.child_cutoff, edge.limitation_cutoff, edge.mode, + edge.operation, + edge.source_span, ) for edge in edges ] @@ -2446,6 +2612,7 @@ def resolve_nested_argument( root.prefix, root.line, combined_conditions, + root.source_span, ) emitted_objects = [root if item.key == root.key else item for item in emitted_objects] return _FactoryGraph(root, emitted_objects, routes, edges) @@ -2798,6 +2965,9 @@ def clear_binding(name: str) -> None: decorator_methods, decorator_handler, next_order(), + registration_kind=NativeRegistrationKind.DECORATOR, + operation=operation, + source_span=_native_span(current_module.path, decorator), ) ) displace_global_binding(statement.name, statement) @@ -3065,6 +3235,8 @@ def clear_binding(name: str) -> None: order, (current_module.name, order), "copy", + "include_router", + _native_span(current_module.path, call), ) ) elif operation == "mount": @@ -3092,6 +3264,8 @@ def clear_binding(name: str) -> None: None, None, "live", + "mount", + _native_span(current_module.path, call), ) ) else: @@ -3123,7 +3297,18 @@ def clear_binding(name: str) -> None: inventory_only=True, ) else: - routes.append(_Route(parent.key, path, methods, handler, order)) + routes.append( + _Route( + parent.key, + path, + methods, + handler, + order, + registration_kind=NativeRegistrationKind.IMPERATIVE, + operation=operation, + source_span=_native_span(current_module.path, call), + ) + ) continue helper_target: tuple[_Module, ast.FunctionDef | ast.AsyncFunctionDef] | None = None @@ -3779,7 +3964,18 @@ def classify_direct_call( # noqa: PLR0911, PLR0912, PLR0915 inventory_only=True, ) return _DirectEffectResult("limited") - routes.append(_Route(owner.key, path, methods, handler, route_order)) + routes.append( + _Route( + owner.key, + path, + methods, + handler, + route_order, + registration_kind=NativeRegistrationKind.DECORATOR, + operation=operation, + source_span=_native_span(module.path, call), + ) + ) return _DirectEffectResult("modeled") assert effect_node is not None @@ -3813,7 +4009,18 @@ def classify_direct_call( # noqa: PLR0911, PLR0912, PLR0915 inventory_only=True, ) return _DirectEffectResult("limited") - routes.append(_Route(owner.key, path, methods, imperative_handler, effect_order)) + routes.append( + _Route( + owner.key, + path, + methods, + imperative_handler, + effect_order, + registration_kind=NativeRegistrationKind.IMPERATIVE, + operation=operation, + source_span=_native_span(module.path, call), + ) + ) return _DirectEffectResult("modeled") if operation == "include_router": @@ -3854,6 +4061,8 @@ def classify_direct_call( # noqa: PLR0911, PLR0912, PLR0915 cutoff, (module.name, effect_order), "copy", + "include_router", + _native_span(module.path, call), ) ) return _DirectEffectResult("modeled") @@ -3871,7 +4080,19 @@ def classify_direct_call( # noqa: PLR0911, PLR0912, PLR0915 inventory_only=True, ) return _DirectEffectResult("limited") - edges.append(_Edge(owner.key, child.key, path, effect_order, None, None, "live")) + edges.append( + _Edge( + owner.key, + child.key, + path, + effect_order, + None, + None, + "live", + "mount", + _native_span(module.path, call), + ) + ) return _DirectEffectResult("modeled") def visit( # noqa: PLR0912, PLR0915 - explicit executed-effect taxonomy diff --git a/tests/unit/test_formatters.py b/tests/unit/test_formatters.py index bdef16d..677bd18 100644 --- a/tests/unit/test_formatters.py +++ b/tests/unit/test_formatters.py @@ -41,6 +41,7 @@ from fastapi_endpoint_detector.output.markdown_output import MarkdownFormatter from fastapi_endpoint_detector.output.text_output import TextFormatter from fastapi_endpoint_detector.output.yaml_output import YamlFormatter +from fastapi_endpoint_detector.parser.secure_ast_extractor import SecureASTExtractor def test_inventory_strength_is_structured_and_visible() -> None: @@ -57,7 +58,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"] == 3 + assert json_result["schema_version"] == 4 assert json_result["inventory_status"] == "conditional" assert json_result["inventory_limitations"][0]["reason"] == limitation.reason assert json_result["route_conditions"][0]["reason"] == limitation.reason @@ -129,6 +130,44 @@ def test_json_and_yaml_preserve_optional_dependency_graph() -> None: } +def test_schema_v4_json_and_yaml_preserve_nested_native_provenance( + tmp_path: Path, +) -> None: + app_file = tmp_path / "main.py" + app_file.write_text( + "from fastapi import APIRouter, FastAPI\n" + "app = FastAPI()\n" + "mounted = FastAPI()\n" + "router = APIRouter()\n" + "@router.get('/items')\n" + "def items(): pass\n" + "mounted.include_router(router)\n" + "app.mount('/service', mounted)\n", + encoding="utf-8", + ) + inventory = SecureASTExtractor(app_file).extract_inventory() + + json_result = json.loads(JsonFormatter().format_inventory(inventory)) + yaml_result = yaml.safe_load(YamlFormatter().format_inventory(inventory)) + assert json_result["schema_version"] == yaml_result["schema_version"] == 4 + json_provenance = json_result["endpoints"][0]["native_provenance"] + yaml_provenance = yaml_result["endpoints"][0]["native_provenance"] + assert json_provenance == yaml_provenance + assert [edge["operation"] for edge in json_provenance["assembly_chain"]] == [ + "mount", + "include_router", + ] + assert [edge["mode"] for edge in json_provenance["assembly_chain"]] == [ + "live", + "copy", + ] + assert [item["object_kind"] for item in json_provenance["object_chain"]] == [ + "app", + "app", + "router", + ] + + def test_json_and_yaml_preserve_startup_activation_evidence() -> None: condition = EndpointDiscoveryCondition( source_path=Path("/app/main.py"), diff --git a/tests/unit/test_route_provenance.py b/tests/unit/test_route_provenance.py new file mode 100644 index 0000000..ba30067 --- /dev/null +++ b/tests/unit/test_route_provenance.py @@ -0,0 +1,372 @@ +"""Secure-AST native route assembly provenance and ownership tests.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest +from pydantic import ValidationError + +from fastapi_endpoint_detector.analyzer.change_mapper import ChangeMapper +from fastapi_endpoint_detector.analyzer.endpoint_registry import EndpointRegistry +from fastapi_endpoint_detector.models.diff import ChangeType, DiffFile, DiffHunk +from fastapi_endpoint_detector.models.endpoint import ( + NativeRegistrationKind, + NativeRootSelectionKind, + NativeRouteProvenance, + SnapshotSide, +) +from fastapi_endpoint_detector.models.report import ( + ChangeEffectKind, + EvidenceProducer, +) +from fastapi_endpoint_detector.parser.secure_ast_extractor import SecureASTExtractor + + +class _EmptySCIP: + def ensure_index(self, *, force: bool = False) -> None: + del force + + def definitions_at(self, _file_path: Path, _lines: set[int]) -> list[object]: + return [] + + +def _write_composed_app(path: Path, *, prefix: str = "/new") -> None: + path.write_text( + "from fastapi import APIRouter, FastAPI\n" + "app = FastAPI()\n" + "first = APIRouter(prefix='/first')\n" + "second = APIRouter(prefix='/second')\n" + "@first.get(\n" + " '/items',\n" + ")\n" + "def first_items(): pass\n" + "@second.get('/items')\n" + "def second_items(): pass\n" + f"app.include_router(first, prefix={prefix!r})\n" + "app.include_router(second, prefix='/other')\n", + encoding="utf-8", + ) + + +def _nested_provenance_payload(tmp_path: Path) -> dict[str, Any]: + app_file = tmp_path / "nested.py" + app_file.write_text( + "from fastapi import APIRouter, FastAPI\n" + "app = FastAPI()\n" + "mounted = FastAPI()\n" + "router = APIRouter()\n" + "@router.get('/items')\n" + "def items(): pass\n" + "mounted.include_router(router)\n" + "app.mount('/service', mounted)\n", + encoding="utf-8", + ) + endpoint = SecureASTExtractor(app_file).extract_endpoints()[0] + assert endpoint.native_provenance is not None + return endpoint.native_provenance.model_dump(mode="python") + + +def test_secure_provenance_preserves_multiline_registration_and_exact_chain( + tmp_path: Path, +) -> None: + app_file = tmp_path / "main.py" + _write_composed_app(app_file) + + endpoints = SecureASTExtractor(app_file).extract_endpoints() + endpoint = next(item for item in endpoints if item.identifier == "GET /new/first/items") + provenance = endpoint.native_provenance + + assert provenance is not None + assert provenance.side == SnapshotSide.TARGET + assert provenance.root.selection_kind == NativeRootSelectionKind.APP_VARIABLE + assert provenance.registration.kind == NativeRegistrationKind.DECORATOR + assert provenance.registration.operation == "get" + assert ( + provenance.registration.source_span.start_line, + provenance.registration.source_span.end_line, + ) == (5, 7) + assert [item.operation for item in provenance.assembly_chain] == ["include_router"] + assert [item.resolved_prefix for item in provenance.assembly_chain] == ["/new"] + assert provenance.assembly_chain[0].source_span.start_line == 11 + assert [item.object_kind for item in provenance.object_chain] == ["app", "router"] + assert [item.resolved_prefix for item in provenance.object_chain] == ["", "/first"] + + +def test_repeated_include_occurrences_remain_distinct(tmp_path: Path) -> None: + app_file = tmp_path / "main.py" + app_file.write_text( + "from fastapi import APIRouter, FastAPI\n" + "app = FastAPI()\n" + "router = APIRouter()\n" + "@router.get('/items')\n" + "def items(): pass\n" + "app.include_router(router, prefix='/one')\n" + "app.include_router(router, prefix='/two')\n", + encoding="utf-8", + ) + + endpoints = SecureASTExtractor(app_file).extract_endpoints() + evidence = { + item.identifier: item.native_provenance.assembly_chain[0] # type: ignore[union-attr] + for item in endpoints + } + + assert set(evidence) == {"GET /one/items", "GET /two/items"} + assert {item.source_span.start_line for item in evidence.values()} == {6, 7} + assert len({item.occurrence_order for item in evidence.values()}) == 2 + + +def test_snapshot_side_and_factory_bootstrap_roots_are_explicit(tmp_path: Path) -> None: + (tmp_path / "factory.py").write_text( + "from fastapi import FastAPI\n" + "def create():\n" + " app = FastAPI()\n" + " @app.get('/factory')\n" + " def route(): pass\n" + " return app\n", + encoding="utf-8", + ) + factory_endpoint = SecureASTExtractor( + tmp_path, + app_entry="factory:create", + snapshot_side=SnapshotSide.BASELINE, + ).extract_endpoints()[0] + + assert factory_endpoint.native_provenance is not None + assert factory_endpoint.native_provenance.side == SnapshotSide.BASELINE + assert ( + factory_endpoint.native_provenance.root.selection_kind + == NativeRootSelectionKind.APP_ENTRY_FACTORY + ) + + (tmp_path / "bootstrap.py").write_text( + "from fastapi import FastAPI\n" + "app = FastAPI()\n" + "def route(): pass\n" + "def run():\n" + " app.add_api_route('/bootstrap', route)\n", + encoding="utf-8", + ) + bootstrap_endpoint = SecureASTExtractor( + tmp_path, + app_entry="bootstrap:app", + bootstrap_entry="bootstrap:run", + ).extract_endpoints()[0] + + assert bootstrap_endpoint.native_provenance is not None + assert bootstrap_endpoint.native_provenance.root.bootstrap_span is not None + assert ( + bootstrap_endpoint.native_provenance.registration.kind == NativeRegistrationKind.IMPERATIVE + ) + assert bootstrap_endpoint.native_provenance.registration.source_span.start_line == 5 + + +def test_native_provenance_rejects_root_disconnected_from_object_chain( + tmp_path: Path, +) -> None: + payload = _nested_provenance_payload(tmp_path) + payload["root"]["symbol"] = "not_the_selected_root" + + with pytest.raises(ValidationError, match="root must be the first object"): + NativeRouteProvenance.model_validate(payload) + + +@pytest.mark.parametrize( + ("selection_kind", "incompatible_kind"), + [ + (NativeRootSelectionKind.APP_VARIABLE, "router"), + (NativeRootSelectionKind.APP_ENTRY_OBJECT, "router"), + (NativeRootSelectionKind.APP_ENTRY_FACTORY, "router"), + (NativeRootSelectionKind.ROUTER_VARIABLE, "app"), + ], +) +def test_native_provenance_rejects_incompatible_root_object_role( + tmp_path: Path, + selection_kind: NativeRootSelectionKind, + incompatible_kind: str, +) -> None: + payload = _nested_provenance_payload(tmp_path) + payload["root"]["selection_kind"] = selection_kind + payload["object_chain"][0]["object_kind"] = incompatible_kind + + with pytest.raises(ValidationError, match="root selection must match"): + NativeRouteProvenance.model_validate(payload) + + +@pytest.mark.parametrize( + ("edge_index", "incompatible_mode"), + [(0, "copy"), (1, "live")], + ids=["mount-copy", "include-router-live"], +) +def test_native_provenance_rejects_incompatible_assembly_mode( + tmp_path: Path, + edge_index: int, + incompatible_mode: str, +) -> None: + payload = _nested_provenance_payload(tmp_path) + payload["assembly_chain"][edge_index]["mode"] = incompatible_mode + + with pytest.raises(ValidationError, match="operation must use its declared composition mode"): + NativeRouteProvenance.model_validate(payload) + + +@pytest.mark.parametrize( + ("object_index", "incompatible_kind"), + [(1, "router"), (2, "app")], + ids=["mount-router", "include-router-app"], +) +def test_native_provenance_rejects_incompatible_assembly_object_role( + tmp_path: Path, + object_index: int, + incompatible_kind: str, +) -> None: + payload = _nested_provenance_payload(tmp_path) + payload["object_chain"][object_index]["object_kind"] = incompatible_kind + + with pytest.raises(ValidationError, match="operation must target a compatible object role"): + NativeRouteProvenance.model_validate(payload) + + +def test_registry_maps_include_change_only_to_exact_descendants(tmp_path: Path) -> None: + app_file = tmp_path / "main.py" + _write_composed_app(app_file) + endpoints = SecureASTExtractor(app_file).extract_endpoints() + registry = EndpointRegistry() + registry.register_many(endpoints) + + matches = registry.get_structural_overlaps(Path("main.py"), {11}) + + assert [(item.identifier, kinds, lines) for item, kinds, lines in matches] == [ + ("GET /new/first/items", ("assembly_include_router",), {11}) + ] + assert registry.get_structural_overlaps(Path("ambiguous/main.py"), {11}) == [] + + +def test_change_mapper_emits_structural_evidence_for_target_prefix_change( + tmp_path: Path, + monkeypatch, +) -> None: + app_file = tmp_path / "main.py" + _write_composed_app(app_file) + mapper = ChangeMapper(app_file, secure_ast=True, use_cache=False) + monkeypatch.setattr(mapper, "_check_mypy_dependency", lambda *_args, **_kwargs: None) + diff_file = DiffFile( + path=Path("main.py"), + change_type=ChangeType.MODIFIED, + hunks=[ + DiffHunk( + source_start=11, + source_length=1, + target_start=11, + target_length=1, + added_lines=[11], + removed_lines=[11], + ) + ], + added_lines=1, + removed_lines=1, + ) + + affected, processed_added, processed_removed = mapper._analyze_diff_file(diff_file) + + assert [item.endpoint.identifier for item in affected] == ["GET /new/first/items"] + assert processed_added == {11} + assert processed_removed == set() + assert affected[0].effect_evidence[0].producer == EvidenceProducer.STRUCTURAL + assert affected[0].effect_evidence[0].effect == ChangeEffectKind.ROUTE_ASSEMBLY + + +def test_duplicate_public_routes_keep_distinct_assembly_occurrences( + tmp_path: Path, + monkeypatch, +) -> None: + app_file = tmp_path / "main.py" + app_file.write_text( + "from fastapi import APIRouter, FastAPI\n" + "app = FastAPI()\n" + "router = APIRouter()\n" + "@router.get('/items')\n" + "def items(): pass\n" + "app.include_router(router)\n" + "app.include_router(router)\n", + encoding="utf-8", + ) + mapper = ChangeMapper(app_file, secure_ast=True, use_cache=False) + monkeypatch.setattr(mapper, "_check_mypy_dependency", lambda *_args, **_kwargs: None) + diff_file = DiffFile( + path=Path("main.py"), + change_type=ChangeType.MODIFIED, + hunks=[ + DiffHunk( + source_start=6, + source_length=2, + target_start=6, + target_length=2, + added_lines=[6, 7], + removed_lines=[], + ) + ], + added_lines=2, + ) + + affected, processed_added, _processed_removed = mapper._analyze_diff_file(diff_file) + + assert [item.endpoint.identifier for item in affected] == ["GET /items", "GET /items"] + assert { + item.endpoint.native_provenance.assembly_chain[0].source_span.start_line # type: ignore[union-attr] + for item in affected + } == {6, 7} + assert processed_added == {6, 7} + + +def test_scip_mode_keeps_target_and_baseline_assembly_sides_distinct(tmp_path: Path) -> None: + target = tmp_path / "target" + baseline = tmp_path / "baseline" + target.mkdir() + baseline.mkdir() + _write_composed_app(target / "main.py", prefix="/new") + _write_composed_app(baseline / "main.py", prefix="/old") + mapper = ChangeMapper( + target / "main.py", + secure_ast=True, + use_scip=True, + use_cache=False, + baseline_app_path=baseline / "main.py", + ) + mapper._scip_analyzer = _EmptySCIP() # type: ignore[assignment] + mapper._baseline_scip_analyzer = _EmptySCIP() # type: ignore[assignment] + diff_file = DiffFile( + path=Path("main.py"), + source_path=Path("main.py"), + change_type=ChangeType.MODIFIED, + hunks=[ + DiffHunk( + source_start=11, + source_length=1, + target_start=11, + target_length=1, + added_lines=[11], + removed_lines=[11], + ) + ], + added_lines=1, + removed_lines=1, + ) + + affected, orphans = mapper._analyze_with_scip([diff_file], [], None) + + assert {item.endpoint.identifier for item in affected} == { + "GET /new/first/items", + "GET /old/first/items", + } + sides = { + item.endpoint.identifier: item.endpoint.native_provenance.side # type: ignore[union-attr] + for item in affected + } + assert sides == { + "GET /new/first/items": SnapshotSide.TARGET, + "GET /old/first/items": SnapshotSide.BASELINE, + } + assert orphans == []