Add tests for Pydantic models

Author: yugokato

src/openapi_test_client/libraries/api/api_functions/utils/param_type.py

@@ -327,13 +327,18 @@ def is_optional_type(tp: Any) -> bool:
             return False
 
 
-def is_union_type(tp: Any) -> bool:
+def is_union_type(tp: Any, exclude_optional: bool = False) -> bool:
     """Check if the type annotation is a Union type
 
     :param tp: Type annotation
+    :param exclude_optional: Exclude Optional[] type
     """
     origin_type = get_origin(tp)
-    return origin_type in [Union, UnionType]
+    is_union = origin_type in [Union, UnionType]
+    if exclude_optional:
+        return is_union and NoneType not in get_args(tp)
+    else:
+        return is_union
 
 
 def is_deprecated_param(tp: Any) -> bool:

src/openapi_test_client/libraries/api/api_functions/utils/pydantic_model.py

@@ -1,20 +1,18 @@
+import datetime
 import inspect
 import ipaddress
 import os
 from collections.abc import Generator
 from contextlib import contextmanager
 from dataclasses import Field as DataclassField
-from datetime import date, datetime, time, timedelta
 from pathlib import Path
 from types import EllipsisType
 from typing import Any, TypeVar, get_args
 from uuid import UUID
 
 from pydantic import (
     AnyHttpUrl,
-    Base64Bytes,
     Base64Str,
-    Base64UrlBytes,
     Base64UrlStr,
     EmailStr,
     Field,
@@ -27,23 +25,23 @@
 
 import openapi_test_client.libraries.api.api_functions.utils.param_model as param_model_util
 import openapi_test_client.libraries.api.api_functions.utils.param_type as param_type_util
-from openapi_test_client.libraries.api.types import Constraint, DataclassModel, EndpointModel, Format, ParamModel
+from openapi_test_client.libraries.api.types import Constraint, DataclassModel, EndpointModel, File, Format, ParamModel
 
 T = TypeVar("T")
 
 
 # TODO: Update this if needed
 PARAM_FORMAT_AND_TYPE_MAP = {
     "uuid": UUID,
-    "date-time": datetime,
-    "date": date,
-    "time": time,
-    "duration": timedelta,
-    "binary": bytes,
+    "date-time": datetime.datetime,
+    "date": datetime.date,
+    "time": datetime.time,
+    "duration": datetime.timedelta,
+    "binary": bytes | File,
     "byte": bytes,
     "path": Path,
-    "base64": Base64Str | Base64Bytes,
-    "base64url": Base64UrlStr | Base64UrlBytes,
+    "base64": Base64Str,
+    "base64url": Base64UrlStr,
     "email": EmailStr,
     "name-email": NameEmail,
     "uri": AnyHttpUrl,

src/openapi_test_client/libraries/common/json_encoder.py

@@ -1,3 +1,4 @@
+import base64
 import json
 from dataclasses import asdict, is_dataclass
 from datetime import datetime
@@ -12,10 +13,12 @@ class CustomJsonEncoder(json.JSONEncoder):
     def default(self, obj: Any) -> Any:
         if isinstance(obj, UUID | Decimal):
             return str(obj)
+        elif isinstance(obj, bytes):
+            return base64.b64encode(obj).decode("utf-8")
         elif isinstance(obj, datetime):
             return obj.isoformat()
         elif is_dataclass(obj) and not isinstance(obj, type):
-            return asdict(obj)
+            return {k: self.default(v) for k, v in asdict(obj).items()}
         elif isinstance(obj, BaseModel):
             return obj.model_dump(mode="json", exclude_unset=True)
         else:

tests/conftest.py

@@ -8,6 +8,8 @@
 from pytest import Item, TempPathFactory
 from pytest_mock import MockerFixture
 
+from openapi_test_client.libraries.api.types import File
+
 
 def pytest_make_parametrize_id(val: Any, argname: str) -> str:
     return f"{argname}={val!r}"
@@ -34,3 +36,17 @@ def _mock_sys_path_and_modules(mocker: MockerFixture) -> None:
 def temp_dir(tmp_path_factory: TempPathFactory) -> Path:
     current_test_name = os.environ["PYTEST_CURRENT_TEST"].rsplit(" ", 1)[0]
     return tmp_path_factory.mktemp(clean_obj_name(current_test_name))
+
+
+@pytest.fixture(scope="session")
+def image_data() -> bytes:
+    return (  # https://evanhahn.com/worlds-smallest-png/
+        b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR"
+        b"\x00\x00\x00\x01\x00\x00\x00\x01\x01\x00\x00\x00\x007n\xf9$\x00\x00\x00\nIDATx"
+        b"\x01c`\x00\x00\x00\x02\x00\x01su\x01\x18\x00\x00\x00\x00IEND\xaeB`\x82"
+    )
+
+
+@pytest.fixture(scope="session")
+def image_file(image_data: bytes) -> File:
+    return File(filename="test_image.png", content=image_data, content_type="image/png")

tests/integration/test_api_users.py

@@ -44,14 +44,8 @@ def test_get_users(api_client: DemoAppAPIClient, validation_mode: bool) -> None:
 
 
 @pytest.mark.parametrize("validation_mode", [False, True])
-def test_upload_image(api_client: DemoAppAPIClient, validation_mode: bool) -> None:
+def test_upload_image(api_client: DemoAppAPIClient, validation_mode: bool, image_data: bytes) -> None:
     """Check basic client/server functionality of upload user image API"""
-    image_data = (
-        # https://evanhahn.com/worlds-smallest-png/
-        b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR"
-        b"\x00\x00\x00\x01\x00\x00\x00\x01\x01\x00\x00\x00\x007n\xf9$\x00\x00\x00\nIDATx"
-        b"\x01c`\x00\x00\x00\x02\x00\x01su\x01\x18\x00\x00\x00\x00IEND\xaeB`\x82"
-    )
     file = File(filename="test_image.png", content=image_data, content_type="image/png")
     r = api_client.Users.upload_image(file=file, description="test image", validate=validation_mode)
     assert r.status_code == 201

tests/unit/conftest.py

@@ -2,23 +2,24 @@
 import os
 import shutil
 from collections.abc import Generator
-from dataclasses import dataclass
+from dataclasses import dataclass, make_dataclass
 from pathlib import Path
-from typing import Any
+from typing import Annotated, Any, cast
 
 import pytest
+from _pytest.fixtures import SubRequest
 from pytest import FixtureRequest
 from pytest_mock import MockerFixture
 
+import openapi_test_client.libraries.api.api_functions.utils.param_type as param_type_util
 from openapi_test_client import ENV_VAR_PACKAGE_DIR
 from openapi_test_client.clients.base import OpenAPIClient
 from openapi_test_client.clients.demo_app import DemoAppAPIClient
 from openapi_test_client.clients.demo_app.api.auth import AuthAPI
-from openapi_test_client.libraries.api.api_client_generator import (
-    setup_external_directory,
-)
+from openapi_test_client.libraries.api.api_client_generator import setup_external_directory
+from openapi_test_client.libraries.api.api_functions.utils.pydantic_model import PARAM_FORMAT_AND_TYPE_MAP
 from openapi_test_client.libraries.api.api_spec import OpenAPISpec
-from openapi_test_client.libraries.api.types import ParamModel, Unset
+from openapi_test_client.libraries.api.types import File, Format, Optional, ParamModel, Unset
 from tests.unit import helper
 
 
@@ -96,3 +97,74 @@ class Model(ParamModel):
         inner_param2: str = Unset
 
     return Model
+
+
+@pytest.fixture
+def NewParamModel(request: SubRequest) -> type[ParamModel]:
+    """A new dataclass param model generated with requested field data via indirect parametrization
+
+    The fixture can be take the field data in various shapes as follows:
+    - Just one field:
+        - Only field type (field name and the default value will be automatically set)
+        - As tuple (field name, field type) or (field name, field type, default value)
+    - Multiple fields: List of above
+    """
+    if not hasattr(request, "param"):
+        raise ValueError(f"{NewParamModel.__name__} fixture must be used as an indirect parametrization")
+
+    def add_field(field_data: Any | tuple[str, Any] | tuple[str, Any, Any], idx: int = 1) -> None:
+        if isinstance(field_data, tuple):
+            assert len(field_data) <= 3, f"Invalid field: {field_data}. Each field must be given as 2 or 3 items"
+            if len(field_data) == 1:
+                fields.append((f"field_{idx}", field_data, Unset))
+            elif len(field_data) >= 2:
+                fields.append(field_data)
+        else:
+            fields.append((f"field{idx}", field_data, Unset))
+
+    requested_field_data = request.param
+    fields: list[Any | tuple[str, Any] | tuple[str, Any, Any]] = []
+    if isinstance(requested_field_data, list):
+        for i, requested_field in enumerate(requested_field_data, start=1):
+            add_field(requested_field, idx=i)
+    else:
+        add_field(requested_field_data)
+
+    param_model = cast(type[ParamModel], make_dataclass("Model", fields, bases=(ParamModel,)))
+    return param_model
+
+
+@pytest.fixture(scope="session")
+def ParamModelWithParamFormats() -> type[ParamModel]:
+    """A a dataclass param model that has fields with various param formats we support"""
+    fields = [
+        ("uuid", Optional[Annotated[str, Format("uuid")]], Unset),
+        ("date_time", Optional[Annotated[str, Format("date-time")]], Unset),
+        ("date", Optional[Annotated[str, Format("date")]], Unset),
+        ("time", Optional[Annotated[str, Format("time")]], Unset),
+        ("duration", Optional[Annotated[str, Format("duration")]], Unset),
+        ("binary", Optional[Annotated[str, Format("binary")]], Unset),
+        ("file", Optional[Annotated[File, Format("binary")]], Unset),
+        ("byte", Optional[Annotated[str, Format("byte")]], Unset),
+        ("path", Optional[Annotated[str, Format("path")]], Unset),
+        ("base64", Optional[Annotated[str, Format("base64")]], Unset),
+        ("base64url", Optional[Annotated[str, Format("base64url")]], Unset),
+        ("email", Optional[Annotated[str, Format("email")]], Unset),
+        ("name_email", Optional[Annotated[str, Format("name-email")]], Unset),
+        ("uri", Optional[Annotated[str, Format("uri")]], Unset),
+        ("ipv4", Optional[Annotated[str, Format("ipv4")]], Unset),
+        ("ipv6", Optional[Annotated[str, Format("ipv6")]], Unset),
+        ("ipvanyaddress", Optional[Annotated[str, Format("ipvanyaddress")]], Unset),
+        ("ipvanyinterface", Optional[Annotated[str, Format("ipvanyinterface")]], Unset),
+        ("ipvanynetwork", Optional[Annotated[str, Format("ipvanynetwork")]], Unset),
+        ("phone", Optional[Annotated[str, Format("phone")]], Unset),
+    ]
+    param_model = cast(type[ParamModel], make_dataclass("Model", fields, bases=(ParamModel,)))
+
+    # Make sure the model covers all Pydantic specific types we support
+    annotated_types = [param_type_util.get_annotated_type(t) for _, t, _ in fields]
+    param_formats = [x.__metadata__[0].value for x in annotated_types]
+    undefined_formats = set(PARAM_FORMAT_AND_TYPE_MAP.keys()).difference(set(param_formats))
+    assert not undefined_formats, f"Missing test coverage for these formats: {undefined_formats}"
+
+    return param_model

tests/unit/test_param_type.py

@@ -109,6 +109,7 @@ def test_get_type_annotation_as_str(tp: Any, expected_tp_str: str, is_optional:
         (str | int, str | int),
         (str | None, str),
         (str | int | None, str | int),
+        (Annotated[str, Constraint(min_len=5)] | Annotated[int, Constraint(min=5)], str | int),
         (list[str], str),
         (list[dict[str, Any]], dict[str, Any]),
         (Annotated[str, "meta"], str),
@@ -225,6 +226,7 @@ def test_is_optional_type(tp: Any, is_optional_type: bool) -> None:
     assert param_type_util.is_optional_type(tp) is is_optional_type
 
 
+@pytest.mark.parametrize("exclude_optional", [False, True])
 @pytest.mark.parametrize(
     ("tp", "is_union_type"),
     [
@@ -240,12 +242,16 @@ def test_is_optional_type(tp: Any, is_optional_type: bool) -> None:
         (Optional[Annotated[str | int, "meta"]], True),
     ],
 )
-def test_is_union_type(tp: Any, is_union_type: bool) -> None:
+def test_is_union_type(tp: Any, is_union_type: bool, exclude_optional: bool) -> None:
     """Verify that we can check whether a given type annotation itself is a union type or not
 
     Note: Optional[] is also considered as union
     """
-    assert param_type_util.is_union_type(tp) is is_union_type
+    if exclude_optional:
+        is_union = is_union_type and NoneType not in get_args(tp)
+    else:
+        is_union = is_union_type
+    assert param_type_util.is_union_type(tp, exclude_optional=exclude_optional) is is_union
 
 
 @pytest.mark.parametrize(