Merge pull request #2957 from markparonyan/api-issue-08-67pip0

feat(api): add machine-readable error codes and align OpenAPI error schemas

gns3server/api/errors.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python
+#
+# Copyright (C) 2026 GNS3 Technologies Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""
+Error response helpers shared by the exception handlers and the OpenAPI schema.
+"""
+
+from collections.abc import Mapping
+from http import HTTPStatus
+from typing import Any
+
+from fastapi.responses import JSONResponse
+
+from gns3server.schemas.common import ErrorMessage
+
+AUTHENTICATED_ERROR_RESPONSES = {
+    401: "Authentication is required or the credentials are invalid",
+    403: "Not enough privileges to perform this operation",
+    422: "Request validation error",
+}
+
+
+def code_for_status(status_code: int) -> str:
+    try:
+        phrase = HTTPStatus(status_code).phrase
+    except ValueError:
+        return "error"
+    return phrase.lower().replace(" ", "_").replace("-", "_")
+
+
+def error_response(
+    status_code: int,
+    message: Any,
+    code: str | None = None,
+    details: dict[str, Any] | None = None,
+    headers: Mapping[str, str] | None = None,
+) -> JSONResponse:
+    content: dict[str, Any] = {"message": message, "code": code or code_for_status(status_code)}
+    if details:
+        content["details"] = details
+    return JSONResponse(status_code=status_code, content=content, headers=headers)
+
+
+def customize_openapi_errors(schema: dict[str, Any]) -> dict[str, Any]:
+    components = schema.setdefault("components", {}).setdefault("schemas", {})
+    components.pop("HTTPValidationError", None)
+    components.pop("ValidationError", None)
+    if "ErrorMessage" not in components:
+        error_schema = ErrorMessage.model_json_schema(ref_template="#/components/schemas/{model}")
+        components["ErrorMessage"] = {k: v for k, v in error_schema.items() if k != "$defs"}
+
+    for path_item in schema.get("paths", {}).values():
+        for operation in path_item.values():
+            if not isinstance(operation, dict) or "responses" not in operation:
+                continue
+            responses = operation["responses"]
+            wanted = ["422"] if "422" in responses else []
+            if operation.get("security"):
+                wanted = ["401", "403", "422"]
+            for status_code in wanted:
+                responses[status_code] = {
+                    "description": AUTHENTICATED_ERROR_RESPONSES[int(status_code)],
+                    "content": {"application/json": {"schema": {"$ref": "#/components/schemas/ErrorMessage"}}},
+                }
+    return schema

gns3server/api/server.py

@@ -19,7 +19,7 @@
 FastAPI app
 """
 
-from typing import cast
+from typing import Any, cast
 
 from fastapi import FastAPI, HTTPException, Request, status
 from fastapi.exceptions import RequestValidationError
@@ -29,13 +29,14 @@ from fastapi.openapi.docs import (
     get_swagger_ui_html,
     get_swagger_ui_oauth2_redirect_html,
 )
-from fastapi.responses import JSONResponse
 from fastapi.staticfiles import StaticFiles
 from sqlalchemy.exc import SQLAlchemyError
+from starlette.exceptions import HTTPException as StarletteHTTPException
 from uvicorn.main import Server as UvicornServer
 
 # MCP is an optional feature — import only if dependencies are installed
 from gns3server.agent import MCP_AVAILABLE
+from gns3server.api.errors import customize_openapi_errors, error_response
 from gns3server.api.routes import controller, index
 from gns3server.api.routes.compute import compute_api
 from gns3server.controller.controller_error import (
@@ -71,9 +72,16 @@ import logging
 log = logging.getLogger(__name__)
 
 
+class ControllerAPI(FastAPI):
+    def openapi(self) -> dict[str, Any]:
+        if self.openapi_schema is None:
+            customize_openapi_errors(super().openapi())
+        return cast(dict[str, Any], self.openapi_schema)
+
+
 def get_application() -> FastAPI:
 
-    application = FastAPI(
+    application = ControllerAPI(
         lifespan=tasks.lifespan,
         title="GNS3 controller API",
         description="This page describes the public controller API for GNS3.\n\n"
@@ -158,82 +166,65 @@ async def redoc_html():
 @app.exception_handler(ControllerError)
 async def controller_error_handler(request: Request, exc: ControllerError):
     log.error(f"Controller error in {request.url.path} ({request.method}): {exc}")
-    return JSONResponse(
-        status_code=status.HTTP_409_CONFLICT,
-        content={"message": str(exc)},
-    )
+    return error_response(status.HTTP_409_CONFLICT, str(exc), exc.code, exc.details)
 
 
 @app.exception_handler(ControllerTimeoutError)
 async def controller_timeout_error_handler(request: Request, exc: ControllerTimeoutError):
     log.error(f"Controller timeout error in {request.url.path} ({request.method}): {exc}")
-    return JSONResponse(
-        status_code=status.HTTP_408_REQUEST_TIMEOUT,
-        content={"message": str(exc)},
-    )
+    return error_response(status.HTTP_408_REQUEST_TIMEOUT, str(exc), exc.code, exc.details)
 
 
 @app.exception_handler(ControllerUnauthorizedError)
 async def controller_unauthorized_error_handler(request: Request, exc: ControllerUnauthorizedError):
     log.error(f"Controller unauthorized error in {request.url.path} ({request.method}): {exc}")
-    return JSONResponse(
-        status_code=status.HTTP_401_UNAUTHORIZED,
-        content={"message": str(exc)},
-    )
+    return error_response(status.HTTP_401_UNAUTHORIZED, str(exc), exc.code, exc.details)
 
 
 @app.exception_handler(ControllerForbiddenError)
 async def controller_forbidden_error_handler(request: Request, exc: ControllerForbiddenError):
     log.error(f"Controller forbidden error in {request.url.path} ({request.method}): {exc}")
-    return JSONResponse(
-        status_code=status.HTTP_403_FORBIDDEN,
-        content={"message": str(exc)},
-    )
+    return error_response(status.HTTP_403_FORBIDDEN, str(exc), exc.code, exc.details)
 
 
 @app.exception_handler(ControllerNotFoundError)
 async def controller_not_found_error_handler(request: Request, exc: ControllerNotFoundError):
     log.error(f"Controller not found error in {request.url.path} ({request.method}): {exc}")
-    return JSONResponse(
-        status_code=status.HTTP_404_NOT_FOUND,
-        content={"message": str(exc)},
-    )
+    return error_response(status.HTTP_404_NOT_FOUND, str(exc), exc.code, exc.details)
 
 
 @app.exception_handler(ControllerBadRequestError)
 async def controller_bad_request_error_handler(request: Request, exc: ControllerBadRequestError):
     log.error(f"Controller bad request error in {request.url.path} ({request.method}): {exc}")
-    return JSONResponse(
-        status_code=status.HTTP_400_BAD_REQUEST,
-        content={"message": str(exc)},
-    )
+    return error_response(status.HTTP_400_BAD_REQUEST, str(exc), exc.code, exc.details)
 
 
 @app.exception_handler(ComputeConflictError)
 async def compute_conflict_error_handler(request: Request, exc: ComputeConflictError):
     log.error(f"Controller received error from compute for request '{exc.url()}': {exc}")
-    return JSONResponse(
-        status_code=status.HTTP_409_CONFLICT,
-        content={"message": str(exc)},
-    )
+    return error_response(status.HTTP_409_CONFLICT, str(exc), exc.code, exc.details)
 
 
-# make sure the content key is "message", not "detail" per default
-@app.exception_handler(HTTPException)
-async def http_exception_handler(request: Request, exc: HTTPException):
-    return JSONResponse(status_code=exc.status_code, content={"message": exc.detail}, headers=exc.headers)
+@app.exception_handler(StarletteHTTPException)
+async def http_exception_handler(request: Request, exc: StarletteHTTPException):
+    return error_response(exc.status_code, exc.detail, headers=exc.headers)
 
 
 @app.exception_handler(SQLAlchemyError)
 async def sqlalchemy_error_handler(request: Request, exc: SQLAlchemyError):
     log.error(f"Controller database error in {request.url.path} ({request.method}): {exc}")
-    return JSONResponse(
-        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-        content={"message": "Database error detected, please check logs to find details"},
+    return error_response(
+        status.HTTP_500_INTERNAL_SERVER_ERROR,
+        "Database error detected, please check logs to find details",
+        "database_error",
     )
 
 
 @app.exception_handler(RequestValidationError)
 async def validation_exception_handler(request: Request, exc: RequestValidationError):
     log.error(f"Request validation error in {request.url.path} ({request.method}): {exc}")
-    return JSONResponse(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, content={"message": str(exc)})
+    errors = [
+        {"loc": list(error.get("loc", ())), "msg": error.get("msg"), "type": error.get("type")}
+        for error in exc.errors()
+    ]
+    return error_response(status.HTTP_422_UNPROCESSABLE_CONTENT, str(exc), "validation_error", {"errors": errors})

gns3server/controller/__init__.py

@@ -710,7 +710,9 @@ class Controller:
         except KeyError:
             if compute_id == "vm":
                 raise ControllerNotFoundError("Cannot use a node on the GNS3 VM server with the GNS3 VM not configured")
-            raise ControllerNotFoundError(f"Compute ID {compute_id} doesn't exist")
+            raise ControllerNotFoundError(
+                f"Compute ID {compute_id} doesn't exist", code="compute_not_found", details={"compute_id": compute_id}
+            )
 
     def has_compute(self, compute_id):
         """
@@ -748,7 +750,9 @@ class Controller:
         try:
             return self._projects[project_id]
         except KeyError:
-            raise ControllerNotFoundError(f"Project ID {project_id} doesn't exist")
+            raise ControllerNotFoundError(
+                f"Project ID {project_id} doesn't exist", code="project_not_found", details={"project_id": project_id}
+            )
 
     async def get_loaded_project(self, project_id):
         """

gns3server/controller/controller_error.py

@@ -15,11 +15,17 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+from typing import Any
+
 
 class ControllerError(Exception):
-    def __init__(self, message: str):
+    default_code = "conflict"
+
+    def __init__(self, message: str, code: str | None = None, details: dict[str, Any] | None = None):
         super().__init__()
         self._message = message
+        self._code = code or self.default_code
+        self._details = details
 
     def __repr__(self):
         return self._message
@@ -27,30 +33,33 @@ class ControllerError(Exception):
     def __str__(self):
         return self._message
 
+    @property
+    def code(self) -> str:
+        return self._code
+
+    @property
+    def details(self) -> dict[str, Any] | None:
+        return self._details
+
 
 class ControllerNotFoundError(ControllerError):
-    def __init__(self, message: str):
-        super().__init__(message)
+    default_code = "not_found"
 
 
 class ControllerBadRequestError(ControllerError):
-    def __init__(self, message: str):
-        super().__init__(message)
+    default_code = "bad_request"
 
 
 class ControllerUnauthorizedError(ControllerError):
-    def __init__(self, message: str):
-        super().__init__(message)
+    default_code = "unauthorized"
 
 
 class ControllerForbiddenError(ControllerError):
-    def __init__(self, message: str):
-        super().__init__(message)
+    default_code = "forbidden"
 
 
 class ControllerTimeoutError(ControllerError):
-    def __init__(self, message: str):
-        super().__init__(message)
+    default_code = "timeout"
 
 
 class ComputeError(ControllerError):
@@ -65,8 +74,10 @@ class ComputeConflictError(ComputeError):
     :param response: compute JSON response
     """
 
+    default_code = "compute_conflict"
+
     def __init__(self, url, response):
-        super().__init__(response["message"])
+        super().__init__(response["message"], code=response.get("code"), details=response.get("details"))
         self._url = url
         self._response = response
 

gns3server/controller/link.py

@@ -274,9 +274,22 @@ class Link:
 
         port = node.get_port(adapter_number, port_number)
         if port is None:
-            raise ControllerNotFoundError(f"Port {adapter_number}/{port_number} for {node.name} not found")
+            raise ControllerNotFoundError(
+                f"Port {adapter_number}/{port_number} for {node.name} not found",
+                code="port_not_found",
+                details={"node_id": node.id, "adapter_number": adapter_number, "port_number": port_number},
+            )
         if port.link is not None:
-            raise ControllerError("Port is already used")
+            raise ControllerError(
+                "Port is already used",
+                code="port_in_use",
+                details={
+                    "node_id": node.id,
+                    "adapter_number": adapter_number,
+                    "port_number": port_number,
+                    "link_id": port.link.id,
+                },
+            )
 
         self._link_type = port.link_type
 

gns3server/controller/project.py

@@ -748,7 +748,9 @@ class Project:
         try:
             return self._nodes[node_id]
         except KeyError:
-            raise ControllerNotFoundError(f"Node ID {node_id} doesn't exist")
+            raise ControllerNotFoundError(
+                f"Node ID {node_id} doesn't exist", code="node_not_found", details={"node_id": node_id}
+            )
 
     def _get_closed_data(self, section, id_key):
         """
@@ -817,7 +819,9 @@ class Project:
         try:
             return self._drawings[drawing_id]
         except KeyError:
-            raise ControllerNotFoundError(f"Drawing ID {drawing_id} doesn't exist")
+            raise ControllerNotFoundError(
+                f"Drawing ID {drawing_id} doesn't exist", code="drawing_not_found", details={"drawing_id": drawing_id}
+            )
 
     @open_required
     async def delete_drawing(self, drawing_id):
@@ -1101,7 +1105,9 @@ class Project:
         try:
             return self._links[link_id]
         except KeyError:
-            raise ControllerNotFoundError(f"Link ID {link_id} doesn't exist")
+            raise ControllerNotFoundError(
+                f"Link ID {link_id} doesn't exist", code="link_not_found", details={"link_id": link_id}
+            )
 
     @property
     def links(self):
@@ -1462,7 +1468,11 @@ class Project:
         try:
             return self._snapshots[snapshot_id]
         except KeyError:
-            raise ControllerNotFoundError(f"Snapshot ID {snapshot_id} doesn't exist")
+            raise ControllerNotFoundError(
+                f"Snapshot ID {snapshot_id} doesn't exist",
+                code="snapshot_not_found",
+                details={"snapshot_id": snapshot_id},
+            )
 
     def _load_snapshot_config(self):
 

gns3server/schemas/common.py

@@ -15,7 +15,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from enum import Enum
-from typing import Optional
+from typing import Any, Optional
 
 from pydantic import BaseModel, Field, field_validator
 
@@ -26,6 +26,11 @@ class ErrorMessage(BaseModel):
     """
 
     message: str
+    code: Optional[str] = Field(
+        None,
+        description="Stable machine-readable error code, for example port_in_use or node_not_found",
+    )
+    details: Optional[dict[str, Any]] = Field(None, description="Optional structured data about the error")
 
 
 class NodeStatus(str, Enum):

tests/api/routes/controller/test_version.py

@@ -42,7 +42,10 @@ class TestVersionRoutes:
         params = {"version": "0.4.2"}
         response = await client.post(app.url_path_for("check_version"), json=params)
         assert response.status_code == status.HTTP_409_CONFLICT
-        assert response.json() == {"message": f"Client version 0.4.2 is not the same as server version {__version__}"}
+        assert response.json() == {
+            "message": f"Client version 0.4.2 is not the same as server version {__version__}",
+            "code": "conflict",
+        }
 
     async def test_version_invalid_input_schema(self, app: FastAPI, client: AsyncClient) -> None:
 

tests/api/routes/test_error_responses.py

@@ -0,0 +1,221 @@
+#
+# Copyright (C) 2026 GNS3 Technologies Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import uuid
+from typing import Tuple
+from unittest.mock import MagicMock
+
+import pytest
+import pytest_asyncio
+from fastapi import FastAPI, status
+from httpx import AsyncClient
+
+from gns3server.api.errors import code_for_status
+from gns3server.controller.compute import Compute
+from gns3server.controller.controller_error import ComputeConflictError, ControllerError
+from gns3server.controller.node import Node
+from gns3server.controller.ports.ethernet_port import EthernetPort
+from gns3server.controller.project import Project
+from gns3server.db.models import User
+from gns3server.schemas import ErrorMessage
+from gns3server.services import auth_service
+from gns3server.services.authentication import DEFAULT_JWT_SECRET_KEY
+from tests.utils import AsyncioMagicMock, asyncio_patch
+
+pytestmark = pytest.mark.asyncio
+
+
+class TestErrorResponses:
+    @pytest_asyncio.fixture
+    async def nodes(self, compute: Compute, project: Project) -> Tuple[Node, Node]:
+
+        response = MagicMock()
+        response.json = {"console": 2048}
+        compute.post = AsyncioMagicMock(return_value=response)
+
+        node1 = await project.add_node(compute, "node1", None, node_type="qemu")
+        node1._ports = [EthernetPort("E0", 0, 0, 3)]
+        node2 = await project.add_node(compute, "node2", None, node_type="qemu")
+        node2._ports = [EthernetPort("E0", 0, 2, 4)]
+        return node1, node2
+
+    async def test_unauthorized(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
+
+        response = await client.get(
+            app.url_path_for("get_project", project_id=project.id),
+            headers={"Authorization": "Bearer invalid_token"},
+        )
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
+        body = response.json()
+        assert isinstance(body["message"], str)
+        assert body["code"] == "unauthorized"
+        assert "detail" not in body
+
+    async def test_forbidden(self, app: FastAPI, client: AsyncClient, test_user: User, project: Project) -> None:
+
+        token = auth_service.create_access_token(test_user.username, secret_key=DEFAULT_JWT_SECRET_KEY)
+        response = await client.delete(
+            app.url_path_for("delete_project", project_id=project.id),
+            headers={"Authorization": f"Bearer {token}"},
+        )
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+        body = response.json()
+        assert isinstance(body["message"], str)
+        assert body["code"] == "forbidden"
+
+    async def test_unknown_node(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
+
+        node_id = str(uuid.uuid4())
+        response = await client.get(app.url_path_for("get_node", project_id=project.id, node_id=node_id))
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+        body = response.json()
+        assert body["message"] == f"Node ID {node_id} doesn't exist"
+        assert body["code"] == "node_not_found"
+        assert body["details"] == {"node_id": node_id}
+
+    async def test_unknown_project(self, app: FastAPI, client: AsyncClient) -> None:
+
+        project_id = str(uuid.uuid4())
+        response = await client.get(app.url_path_for("get_project", project_id=project_id))
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+        body = response.json()
+        assert body["code"] == "project_not_found"
+        assert body["details"] == {"project_id": project_id}
+
+    async def test_unknown_route(self, client: AsyncClient) -> None:
+
+        response = await client.get("/v3/no/such/route")
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+        body = response.json()
+        assert body["message"] == "Not Found"
+        assert body["code"] == "not_found"
+        assert "detail" not in body
+
+    async def test_port_in_use(
+        self, app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]
+    ) -> None:
+
+        node1, node2 = nodes
+        payload = {
+            "nodes": [
+                {"node_id": node1.id, "adapter_number": 0, "port_number": 3},
+                {"node_id": node2.id, "adapter_number": 2, "port_number": 4},
+            ]
+        }
+        with asyncio_patch("gns3server.controller.udp_link.UDPLink.create"):
+            first = await client.post(app.url_path_for("create_link", project_id=project.id), json=payload)
+            assert first.status_code == status.HTTP_201_CREATED
+            second = await client.post(app.url_path_for("create_link", project_id=project.id), json=payload)
+
+        assert second.status_code == status.HTTP_409_CONFLICT
+        body = second.json()
+        assert body["message"] == "Port is already used"
+        assert body["code"] == "port_in_use"
+        assert body["details"]["node_id"] == node1.id
+        assert body["details"]["adapter_number"] == 0
+        assert body["details"]["port_number"] == 3
+        assert body["details"]["link_id"] == first.json()["link_id"]
+
+    async def test_generic_conflict(
+        self, app: FastAPI, client: AsyncClient, project: Project, nodes: Tuple[Node, Node]
+    ) -> None:
+
+        node1, _ = nodes
+        node1._ports = [EthernetPort("E0", 0, 0, 3), EthernetPort("E1", 0, 0, 4)]
+        response = await client.post(
+            app.url_path_for("create_link", project_id=project.id),
+            json={
+                "nodes": [
+                    {"node_id": node1.id, "adapter_number": 0, "port_number": 3},
+                    {"node_id": node1.id, "adapter_number": 0, "port_number": 4},
+                ]
+            },
+        )
+        assert response.status_code == status.HTTP_409_CONFLICT
+        body = response.json()
+        assert body["message"] == "Cannot connect to itself"
+        assert body["code"] == "conflict"
+        assert "details" not in body
+
+    async def test_validation_error(self, app: FastAPI, client: AsyncClient, project: Project) -> None:
+
+        response = await client.post(app.url_path_for("create_link", project_id=project.id), json={"nodes": "bad"})
+        assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
+        body = response.json()
+        assert isinstance(body["message"], str)
+        assert body["code"] == "validation_error"
+        assert body["details"]["errors"]
+        assert {"loc", "msg", "type"} <= set(body["details"]["errors"][0])
+        assert "detail" not in body
+
+    async def test_message_only_clients_keep_working(self, app: FastAPI, client: AsyncClient) -> None:
+
+        response = await client.get(app.url_path_for("get_project", project_id=str(uuid.uuid4())))
+        assert ErrorMessage.model_validate({"message": response.json()["message"]}).code is None
+        assert isinstance(response.json()["message"], str)
+
+
+class TestErrorCodes:
+    async def test_default_codes(self) -> None:
+
+        assert ControllerError("x").code == "conflict"
+        assert ControllerError("x", code="custom", details={"a": 1}).details == {"a": 1}
+
+    async def test_compute_conflict_code(self) -> None:
+
+        error = ComputeConflictError("http://c/x", {"message": "boom", "code": "port_in_use"})
+        assert error.code == "port_in_use"
+        assert ComputeConflictError("http://c/x", {"message": "boom"}).code == "compute_conflict"
+
+    @pytest.mark.parametrize(
+        "status_code,code",
+        [(401, "unauthorized"), (403, "forbidden"), (404, "not_found"), (409, "conflict"), (501, "not_implemented")],
+    )
+    async def test_code_for_status(self, status_code: int, code: str) -> None:
+
+        assert code_for_status(status_code) == code
+
+
+class TestOpenAPIErrorSchemas:
+    async def test_error_schemas(self, client: AsyncClient) -> None:
+
+        response = await client.get("/openapi.json")
+        assert response.status_code == status.HTTP_200_OK
+        spec = response.json()
+
+        schemas = spec["components"]["schemas"]
+        assert "HTTPValidationError" not in schemas
+        assert "ValidationError" not in schemas
+        assert set(schemas["ErrorMessage"]["properties"]) == {"message", "code", "details"}
+        assert schemas["ErrorMessage"]["required"] == ["message"]
+
+        error_ref = {"$ref": "#/components/schemas/ErrorMessage"}
+        authenticated = 0
+        for path, path_item in spec["paths"].items():
+            for method, operation in path_item.items():
+                responses = operation["responses"]
+                if "422" in responses:
+                    assert responses["422"]["content"]["application/json"]["schema"] == error_ref, (method, path)
+                if operation.get("security"):
+                    authenticated += 1
+                    for code in ("401", "403", "422"):
+                        assert responses[code]["content"]["application/json"]["schema"] == error_ref, (
+                            method,
+                            path,
+                            code,
+                        )
+        assert authenticated > 0
+        assert "HTTPValidationError" not in response.text