fix(warnings): emit DUPLICATE_OPERATION as a structured warning

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.

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))

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