From 289f9394b60bb54a46ddeaed728265e0991e5a11 Mon Sep 17 00:00:00 2001 From: Mark Paronyan Date: Wed, 7 Oct 2026 10:43:30 +0000 Subject: [PATCH] feat(api): add machine-readable error codes and align OpenAPI error schemas --- gns3server/api/errors.py | 79 +++++++ gns3server/api/server.py | 69 +++--- gns3server/controller/__init__.py | 8 +- gns3server/controller/controller_error.py | 35 ++-- gns3server/controller/link.py | 17 +- gns3server/controller/project.py | 18 +- gns3server/schemas/common.py | 7 +- tests/api/routes/controller/test_version.py | 5 +- tests/api/routes/test_error_responses.py | 221 ++++++++++++++++++++ 9 files changed, 398 insertions(+), 61 deletions(-) create mode 100644 gns3server/api/errors.py create mode 100644 tests/api/routes/test_error_responses.py diff --git a/gns3server/api/errors.py b/gns3server/api/errors.py new file mode 100644 index 000000000..adf9d35a9 --- /dev/null +++ b/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 . + +""" +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 diff --git a/gns3server/api/server.py b/gns3server/api/server.py index 4ea037e83..2edadf895 100644 --- a/gns3server/api/server.py +++ b/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", @@ -151,82 +159,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}) diff --git a/gns3server/controller/__init__.py b/gns3server/controller/__init__.py index 9a65812a8..075256d7e 100644 --- a/gns3server/controller/__init__.py +++ b/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): """ diff --git a/gns3server/controller/controller_error.py b/gns3server/controller/controller_error.py index c6f531498..ac6a747aa 100644 --- a/gns3server/controller/controller_error.py +++ b/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 . +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 diff --git a/gns3server/controller/link.py b/gns3server/controller/link.py index ff20f7c6b..3ac5c7317 100644 --- a/gns3server/controller/link.py +++ b/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 diff --git a/gns3server/controller/project.py b/gns3server/controller/project.py index c5aedf731..09e750040 100644 --- a/gns3server/controller/project.py +++ b/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): diff --git a/gns3server/schemas/common.py b/gns3server/schemas/common.py index 3b700ee5e..262bbc9cf 100644 --- a/gns3server/schemas/common.py +++ b/gns3server/schemas/common.py @@ -15,7 +15,7 @@ # along with this program. If not, see . 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): diff --git a/tests/api/routes/controller/test_version.py b/tests/api/routes/controller/test_version.py index 39fd1ee17..a56e1ff26 100644 --- a/tests/api/routes/controller/test_version.py +++ b/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: diff --git a/tests/api/routes/test_error_responses.py b/tests/api/routes/test_error_responses.py new file mode 100644 index 000000000..db3577494 --- /dev/null +++ b/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 . + +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