From 120a7d94b56745f64ab28be916307f5d92129374 Mon Sep 17 00:00:00 2001
From: SSOBHY2 <mohamedsobhy672005@gmail.com>
Date: Wed, 3 Dec 2025 22:11:43 -0500
Subject: [PATCH] openapi: add test for mixed Pydantic v1/v2 models

---
 tests/test_openapi_mixed_pydantic_versions.py | 122 ++++++++++++++++++
 1 file changed, 122 insertions(+)
 create mode 100644 tests/test_openapi_mixed_pydantic_versions.py

diff --git a/tests/test_openapi_mixed_pydantic_versions.py b/tests/test_openapi_mixed_pydantic_versions.py
new file mode 100644
index 000000000..dc79083b1
--- /dev/null
+++ b/tests/test_openapi_mixed_pydantic_versions.py
@@ -0,0 +1,122 @@
+from typing import List
+
+from fastapi import FastAPI
+from fastapi._compat import may_v1
+from fastapi.testclient import TestClient
+from pydantic import BaseModel
+
+from .utils import PYDANTIC_V2, needs_py_lt_314, needs_pydanticv2
+
+
+class AddressV1(may_v1.BaseModel):
+    street: str
+
+
+class UserV1(may_v1.BaseModel):
+    name: str
+    addresses: List[AddressV1] = []
+
+
+class AddressV2(BaseModel):
+    street: str
+
+
+class UserV2(BaseModel):
+    name: str
+    primary_address: AddressV1
+    secondary_address: AddressV2 | None = None
+
+    if PYDANTIC_V2:
+        # Match the pattern used in other tests to force separate input/output schemas
+        model_config = {"json_schema_serialization_defaults_required": True}
+
+
+def _collect_refs(obj):
+    if isinstance(obj, dict):
+        for key, value in obj.items():
+            if key == "$ref" and isinstance(value, str):
+                yield value
+            else:
+                yield from _collect_refs(value)
+    elif isinstance(obj, list):
+        for item in obj:
+            yield from _collect_refs(item)
+
+
+@needs_pydanticv2
+@needs_py_lt_314
+def test_openapi_mixed_pydantic_models_with_separate_input_output_schemas() -> None:
+    app = FastAPI(separate_input_output_schemas=True)
+
+    @app.post("/v1-users/", response_model=UserV1)
+    def create_v1_user(user: UserV1) -> UserV1:  # pragma: no cover - behavior tested via OpenAPI
+        return user
+
+    @app.post("/v2-users/", response_model=UserV2)
+    def create_v2_user(user: UserV2) -> UserV2:  # pragma: no cover - behavior tested via OpenAPI
+        return user
+
+    client = TestClient(app)
+    response = client.get("/openapi.json")
+    assert response.status_code == 200, response.text
+
+    openapi_schema = response.json()
+    schemas = openapi_schema["components"]["schemas"]
+
+    # Ensure both Pydantic v1 and v2 models are present in the components
+    user_v1_keys = [name for name in schemas if "UserV1" in name]
+    user_v2_keys = [name for name in schemas if "UserV2" in name]
+    address_v1_keys = [name for name in schemas if "AddressV1" in name]
+    address_v2_keys = [name for name in schemas if "AddressV2" in name]
+
+    assert user_v1_keys, "Expected at least one schema for UserV1 (Pydantic v1 model)."
+    assert user_v2_keys, "Expected at least one schema for UserV2 (Pydantic v2 model)."
+    assert address_v1_keys, "Expected at least one schema for AddressV1 (Pydantic v1 model)."
+    assert address_v2_keys, "Expected at least one schema for AddressV2 (Pydantic v2 model)."
+
+    # Ensure that references in the OpenAPI document point to schemas for both versions
+    all_refs = list(_collect_refs(openapi_schema))
+    assert any("UserV1" in ref for ref in all_refs), "Expected at least one $ref to a UserV1 schema."
+    assert any("UserV2" in ref for ref in all_refs), "Expected at least one $ref to a UserV2 schema."
+    assert any(
+        "AddressV1" in ref for ref in all_refs
+    ), "Expected at least one $ref to an AddressV1 schema."
+    assert any(
+        "AddressV2" in ref for ref in all_refs
+    ), "Expected at least one $ref to an AddressV2 schema."
+
+
+@needs_pydanticv2
+@needs_py_lt_314
+def test_openapi_mixed_pydantic_models_without_separate_input_output_schemas() -> None:
+    app = FastAPI(separate_input_output_schemas=False)
+
+    @app.post("/v1-users/", response_model=UserV1)
+    def create_v1_user(user: UserV1) -> UserV1:  # pragma: no cover - behavior tested via OpenAPI
+        return user
+
+    @app.post("/v2-users/", response_model=UserV2)
+    def create_v2_user(user: UserV2) -> UserV2:  # pragma: no cover - behavior tested via OpenAPI
+        return user
+
+    client = TestClient(app)
+    response = client.get("/openapi.json")
+    assert response.status_code == 200, response.text
+
+    openapi_schema = response.json()
+    schemas = openapi_schema["components"]["schemas"]
+
+    # Even without separate_input_output_schemas, both versions should be represented
+    user_v1_keys = [name for name in schemas if "UserV1" in name]
+    user_v2_keys = [name for name in schemas if "UserV2" in name]
+    address_v1_keys = [name for name in schemas if "AddressV1" in name]
+    address_v2_keys = [name for name in schemas if "AddressV2" in name]
+
+    assert user_v1_keys, "Expected at least one schema for UserV1 (Pydantic v1 model)."
+    assert user_v2_keys, "Expected at least one schema for UserV2 (Pydantic v2 model)."
+    assert address_v1_keys, "Expected at least one schema for AddressV1 (Pydantic v1 model)."
+    assert address_v2_keys, "Expected at least one schema for AddressV2 (Pydantic v2 model)."
+
+    # Check that there are no obviously broken references (all $ref values should target components/schemas)
+    all_refs = list(_collect_refs(openapi_schema))
+    assert all(ref.startswith("#/components/schemas/") for ref in all_refs), "Found a $ref outside components/schemas."