diff --git a/src/azure_functions_openapi/registry.py b/src/azure_functions_openapi/registry.py index 9787562e..346a682b 100644 --- a/src/azure_functions_openapi/registry.py +++ b/src/azure_functions_openapi/registry.py @@ -31,6 +31,7 @@ def __init__(self) -> None: self._entries: dict[str, dict[str, Any]] = {} self._discovery_warnings: list[tuple[str | None, str]] = [] self._empty_discoveries: list[str] = [] + self._duplicate_operations: list[str] = [] self._lock = threading.RLock() @property @@ -97,6 +98,7 @@ def clear(self) -> None: self._entries.clear() self._discovery_warnings.clear() self._empty_discoveries.clear() + self._duplicate_operations.clear() def find_by_function_id( self, function_id: str, method: str | None = None @@ -200,6 +202,27 @@ def empty_discoveries(self) -> list[str]: """Return the recorded empty-app type names, deduplicated and sorted.""" with self._lock: return sorted(self._empty_discoveries) + def add_duplicate_operation(self, method: str, path: str) -> None: + """Record that two registrations collided on the same ``METHOD path``. + + When two ``@openapi`` registrations resolve to the same HTTP method and + path, the spec generator keeps only the last operation and drops the + earlier one (non-strict mode). Historically that drop was only logged, + so ``--fail-on-warnings`` could not observe a silently vanished + operation. Recording it here lets the generator surface a structured + ``duplicate-operation`` warning. Identical ``METHOD path`` collisions are + deduplicated, mirroring :meth:`add_discovery_warning`. + """ + with self._lock: + record = f"{method.upper()} {path}" + if record not in self._duplicate_operations: + self._duplicate_operations.append(record) + + @property + def duplicate_operations(self) -> list[str]: + """Return the recorded ``METHOD path`` collisions, deduplicated and sorted.""" + with self._lock: + return sorted(self._duplicate_operations) # Process-wide singleton. The ``@openapi`` decorator records metadata at import # time — before any application object exists — so a shared instance is required. diff --git a/src/azure_functions_openapi/spec.py b/src/azure_functions_openapi/spec.py index 018eefbd..ad5a3065 100644 --- a/src/azure_functions_openapi/spec.py +++ b/src/azure_functions_openapi/spec.py @@ -479,6 +479,8 @@ def generate_openapi_spec( if strict: raise OpenAPISpecConfigError(_dup_msg) logger.warning("OpenAPI spec: %s", _dup_msg) + _dup_registry = registry if registry is not None else _default_registry + _dup_registry.add_duplicate_operation(method, path) path_item[method] = op except (KeyError, TypeError, ValueError): @@ -814,6 +816,15 @@ def get_openapi_yaml( ) +# Duplicate-operation is a merge-time collision, not a skew signal: two +# registrations resolve to the same METHOD path and the last one silently +# overwrites the earlier operation. The recorded ``METHOD path`` is appended by +# :func:`_collect_duplicate_operation_warnings` for attribution. +_DUPLICATE_OPERATION_MESSAGE = ( + "A duplicate METHOD path collision dropped an operation; only the last " + "@openapi registration appears in the spec" +) + # Method/binding mismatch is authored disagreement, not a skew signal: an # explicit ``@openapi(method=...)`` names a verb the HTTP binding does not serve, # so the generated operation cannot be reached at runtime. @@ -909,6 +920,31 @@ def _collect_empty_discovery_warnings( ] +def _collect_duplicate_operation_warnings( + registry: OpenAPIRegistry | None = None, +) -> list[SpecWarning]: + """Derive duplicate-operation warnings from the registry's recorded collisions. + + When two ``@openapi`` registrations resolve to the same ``METHOD path`` the + spec generator keeps only the last operation and drops the earlier one (#386); + :meth:`OpenAPIRegistry.add_duplicate_operation` records each such collision + during generation. This turns each recorded ``METHOD path`` into a structured + :class:`WarningCode.DUPLICATE_OPERATION` so ``--fail-on-warnings`` can observe + a silently dropped operation instead of it only appearing in the logs. When + ``registry`` is provided its records are used, keeping warnings isolated to the + same registry the spec was built from. + """ + reg = registry if registry is not None else _default_registry + return [ + SpecWarning( + code=WarningCode.DUPLICATE_OPERATION, + message=f"{_DUPLICATE_OPERATION_MESSAGE}: {operation}", + function_name=None, + ) + for operation in reg.duplicate_operations + ] + + def _collect_binding_mismatch_warnings( registry: OpenAPIRegistry | None = None, ) -> list[SpecWarning]: @@ -1005,6 +1041,7 @@ def collect_spec_warnings( warnings_list: list[SpecWarning] = _collect_skew_warnings(registry) warnings_list.extend(_collect_discovery_warnings(registry)) warnings_list.extend(_collect_empty_discovery_warnings(registry)) + warnings_list.extend(_collect_duplicate_operation_warnings(registry)) warnings_list.extend(_collect_binding_mismatch_warnings(registry)) for message in _validate_spec(spec): warnings_list.append(SpecWarning(code=WarningCode.SPEC_VALIDATION, message=message)) diff --git a/tests/test_spec_warnings.py b/tests/test_spec_warnings.py index e547e5eb..1f02047e 100644 --- a/tests/test_spec_warnings.py +++ b/tests/test_spec_warnings.py @@ -19,7 +19,9 @@ from azure_functions_openapi.bridge import scan_endpoint_metadata from azure_functions_openapi.cli import handle_generate from azure_functions_openapi.decorator import clear_openapi_registry +from azure_functions_openapi.exceptions import OpenAPISpecConfigError from azure_functions_openapi.registry import OpenAPIRegistry +from azure_functions_openapi.registry import registry as default_registry from azure_functions_openapi.spec import ( collect_spec_warnings, generate_openapi_report, @@ -419,3 +421,73 @@ def test_empty_paths_hint_adapts_when_app_was_provided( err = capsys.readouterr().err assert "Hint: use --app" not in err assert "no" in err and "@openapi-decorated routes" in err + + +# --------------------------------------------------------------------------- +# #386: duplicate-operation warnings for METHOD path collisions +# --------------------------------------------------------------------------- + + +def _dup_entry(function_name: str) -> dict[str, Any]: + return { + "function_name": function_name, + "route": "dup", + "method": "post", + "response": {"200": {"description": "OK"}}, + } + + +class TestDuplicateOperationWarnings: + """Two registrations colliding on the same METHOD path must surface a + structured duplicate-operation warning; only the last operation wins, and + ``--fail-on-warnings`` must observe the silently dropped operation (#386).""" + + @staticmethod + def _colliding_registry() -> OpenAPIRegistry: + reg = OpenAPIRegistry() + # Distinct registry keys that both resolve to POST /api/dup, so the + # spec merge (not the registry) is what collapses them. + reg.set("first", _dup_entry("first")) + reg.set("second", _dup_entry("second")) + return reg + + def test_duplicate_yields_single_structured_warning(self) -> None: + reg = self._colliding_registry() + spec = generate_openapi_spec(registry=reg) + dups = [ + w + for w in collect_spec_warnings(spec, registry=reg) + if w.code == WarningCode.DUPLICATE_OPERATION + ] + assert len(dups) == 1 + assert "POST /api/dup" in dups[0].message + + def test_last_operation_wins_in_spec(self) -> None: + reg = self._colliding_registry() + spec = generate_openapi_spec(registry=reg) + # The merge keeps exactly one POST operation for the shared path. + path_item = spec["paths"]["/api/dup"] + assert list(path_item.keys()) == ["post"] + # "Last wins" must be verified, not just "one survives": the surviving + # operation must be the second registration, so its operationId reflects + # ``second`` rather than the overwritten ``first``. + assert path_item["post"]["operationId"] == "post_second" + + def test_strict_mode_still_raises(self) -> None: + reg = self._colliding_registry() + with pytest.raises(OpenAPISpecConfigError): + generate_openapi_spec(registry=reg, strict=True) + + def test_no_duplicate_warning_without_collision(self) -> None: + reg = OpenAPIRegistry() + reg.set("only", _dup_entry("only")) + spec = generate_openapi_spec(registry=reg) + codes = {w.code for w in collect_spec_warnings(spec, registry=reg)} + assert WarningCode.DUPLICATE_OPERATION not in codes + + def test_fail_on_warnings_catches_dropped_operation(self) -> None: + # The global CLI path must exit non-zero: a silently dropped operation + # is exactly what --fail-on-warnings exists to catch. + default_registry.set("first", _dup_entry("first")) + default_registry.set("second", _dup_entry("second")) + assert handle_generate(_args(fail_on_warnings=True)) == 2