Add compile/exec validation for generated Python code in tests (#2665)

* Add code validation for generated files and introduce pytest options for execution

* Add validation for generated code execution and improve target version checks

* Add option to skip code validation in OpenAPI tests

* Add coverage exclusions for code validation functions in conftest.py

* Add coverage exclusions for conditional checks in conftest.py

* Refactor output model argument naming to use '--output-model-type' across tests and validation functions

* Refactor output model argument naming to use '--output-model-type' across tests and validation functions

* Add target Python version argument to Pydantic v2 collision test

* Refactor Pydantic v2 collision test to use current Python version for target argument

* Refactor generated code validation to improve exception handling and remove unused imports

* Refactor generated code validation to handle directory inputs correctly

* Refactor validation logic to use DataModelType for Pydantic version checks

Author: koxudaxi

pyproject.toml

@@ -80,6 +80,7 @@ dev = [
 test = [
   "freezegun; python_version<'3.10'",
   "inline-snapshot>=0.31.1",
+  "msgspec>=0.18",
   "pytest>=6.1",
   "pytest>=8.3.4",
   "pytest-benchmark",
@@ -144,6 +145,7 @@ conflicts = [
   [
     { group = "pydantic1" },
     { group = "pkg-meta" },
+    { group = "dev" },
   ],
 ]
 

tests/conftest.py

@@ -5,6 +5,7 @@
 import difflib
 import inspect
 import sys
+import time
 from typing import TYPE_CHECKING, Any, Protocol
 
 import pytest
@@ -16,6 +17,55 @@
     from collections.abc import Callable
     from pathlib import Path
 
+
+class CodeValidationStats:
+    """Track code validation statistics."""
+
+    def __init__(self) -> None:
+        """Initialize statistics counters."""
+        self.compile_count = 0
+        self.compile_time = 0.0
+        self.exec_count = 0
+        self.exec_time = 0.0
+        self.errors: list[tuple[str, str]] = []
+
+    def record_compile(self, elapsed: float) -> None:
+        """Record a compile operation."""
+        self.compile_count += 1
+        self.compile_time += elapsed
+
+    def record_exec(self, elapsed: float) -> None:
+        """Record an exec operation."""
+        self.exec_count += 1
+        self.exec_time += elapsed
+
+    def record_error(self, file_path: str, error: str) -> None:  # pragma: no cover
+        """Record a validation error."""
+        self.errors.append((file_path, error))
+
+
+_validation_stats = CodeValidationStats()
+
+
+def pytest_terminal_summary(terminalreporter: Any, exitstatus: int, config: pytest.Config) -> None:  # noqa: ARG001  # pragma: no cover
+    """Print code validation summary at the end of test run."""
+    if _validation_stats.compile_count > 0:
+        terminalreporter.write_sep("=", "Code Validation Summary")
+        terminalreporter.write_line(
+            f"Compiled {_validation_stats.compile_count} files in {_validation_stats.compile_time:.3f}s "
+            f"(avg: {_validation_stats.compile_time / _validation_stats.compile_count * 1000:.2f}ms)"
+        )
+        if _validation_stats.exec_count > 0:
+            terminalreporter.write_line(
+                f"Executed {_validation_stats.exec_count} files in {_validation_stats.exec_time:.3f}s "
+                f"(avg: {_validation_stats.exec_time / _validation_stats.exec_count * 1000:.2f}ms)"
+            )
+        if _validation_stats.errors:
+            terminalreporter.write_line(f"\nValidation errors: {len(_validation_stats.errors)}")
+            for file_path, error in _validation_stats.errors:
+                terminalreporter.write_line(f"  {file_path}: {error}")
+
+
 if sys.version_info >= (3, 10):
     from datetime import datetime, timezone
 
@@ -358,3 +408,33 @@ def _preload_heavy_modules() -> None:
     import isort  # noqa: PLC0415, F401
 
     import datamodel_code_generator  # noqa: PLC0415, F401
+
+
+def validate_generated_code(
+    code: str,
+    file_path: str,
+    *,
+    do_exec: bool = False,
+) -> None:
+    """Validate generated code by compiling and optionally executing it.
+
+    Args:
+        code: The generated Python code to validate.
+        file_path: Path to the file (for error reporting).
+        do_exec: Whether to execute the code after compiling (default: False).
+    """
+    try:
+        start = time.perf_counter()
+        compiled = compile(code, file_path, "exec")
+        _validation_stats.record_compile(time.perf_counter() - start)
+
+        if do_exec:
+            start = time.perf_counter()
+            exec(compiled, {})
+            _validation_stats.record_exec(time.perf_counter() - start)
+    except SyntaxError as e:  # pragma: no cover
+        _validation_stats.record_error(file_path, f"SyntaxError: {e}")
+        raise
+    except Exception as e:  # pragma: no cover
+        _validation_stats.record_error(file_path, f"{type(e).__name__}: {e}")
+        raise

tests/main/conftest.py

@@ -2,9 +2,11 @@
 
 from __future__ import annotations
 
+import importlib.util
 import inspect
 import shutil
 import sys
+import time
 from argparse import Namespace
 from collections.abc import Callable, Generator, Sequence
 from pathlib import Path
@@ -14,8 +16,17 @@
 import pytest
 from packaging import version
 
+from datamodel_code_generator import DataModelType
 from datamodel_code_generator.__main__ import Exit, main
-from tests.conftest import AssertFileContent, assert_directory_content, assert_output, freeze_time
+from datamodel_code_generator.util import PYDANTIC_V2
+from tests.conftest import (
+    AssertFileContent,
+    _validation_stats,
+    assert_directory_content,
+    assert_output,
+    freeze_time,
+    validate_generated_code,
+)
 
 InputFileTypeLiteral = Literal["auto", "openapi", "jsonschema", "json", "yaml", "dict", "csv", "graphql"]
 CopyFilesMapping = Sequence[tuple[Path, Path]]
@@ -210,6 +221,8 @@ def run_main_and_assert(  # noqa: PLR0912
     # stdin options
     stdin_path: Path | None = None,
     monkeypatch: pytest.MonkeyPatch | None = None,
+    # Code validation options
+    skip_code_validation: bool = False,
 ) -> None:
     """Execute main() and assert output.
 
@@ -335,6 +348,119 @@ def run_main_and_assert(  # noqa: PLR0912
             expected_file = f"{func_name}.py"
         assert_func(output_path, expected_file, transform=transform)
 
+    if output_path is not None and not skip_code_validation:
+        _validate_output_files(output_path, extra_args)
+
+
+def _get_argument_value(arguments: Sequence[str] | None, argument_name: str) -> str | None:
+    """Extract argument value from arguments."""
+    if arguments is None:
+        return None
+    argument_list = list(arguments)
+    for index, argument in enumerate(argument_list):
+        if argument == argument_name and index + 1 < len(argument_list):
+            return argument_list[index + 1]
+    return None
+
+
+def _parse_target_version(extra_arguments: Sequence[str] | None) -> tuple[int, int] | None:
+    """Parse target Python version from arguments."""
+    if (target_version := _get_argument_value(extra_arguments, "--target-python-version")) is None:
+        return None
+    try:
+        return tuple(int(part) for part in target_version.split("."))  # type: ignore[return-value]
+    except ValueError:  # pragma: no cover
+        return None
+
+
+def _should_skip_compile(extra_arguments: Sequence[str] | None) -> bool:
+    """Check if compile should be skipped when target version > runtime version."""
+    if (target_version := _parse_target_version(extra_arguments)) is None:
+        return False
+    return target_version > sys.version_info[:2]
+
+
+def _should_skip_exec(extra_arguments: Sequence[str] | None) -> bool:
+    """Check if exec should be skipped based on model type, pydantic version, and Python version."""
+    output_model_type = _get_argument_value(extra_arguments, "--output-model-type")
+    is_pydantic_v1 = output_model_type is None or output_model_type == DataModelType.PydanticBaseModel.value
+    if (is_pydantic_v1 and PYDANTIC_V2) or (
+        output_model_type == DataModelType.PydanticV2BaseModel.value and not PYDANTIC_V2
+    ):
+        return True
+    if (target_version := _parse_target_version(extra_arguments)) is None:
+        return True
+    if target_version != sys.version_info[:2]:
+        return True
+    return _get_argument_value(extra_arguments, "--base-class") is not None
+
+
+def _validate_output_files(output_path: Path, extra_arguments: Sequence[str] | None = None) -> None:
+    """Validate generated Python files by compiling/executing them."""
+    if _should_skip_compile(extra_arguments):
+        return
+    should_exec = not _should_skip_exec(extra_arguments)
+    if output_path.is_file() and output_path.suffix == ".py":
+        validate_generated_code(output_path.read_text(encoding="utf-8"), str(output_path), do_exec=should_exec)
+    elif output_path.is_dir():  # pragma: no cover
+        for python_file in output_path.rglob("*.py"):
+            validate_generated_code(python_file.read_text(encoding="utf-8"), str(python_file), do_exec=False)
+        if should_exec:  # pragma: no cover
+            _import_package(output_path)
+
+
+def _import_package(output_path: Path) -> None:  # pragma: no cover  # noqa: PLR0912
+    """Import generated packages to validate they can be loaded."""
+    if (output_path / "__init__.py").exists():
+        packages = [(output_path.parent, output_path.name)]
+    else:
+        packages = [
+            (output_path, directory.name)
+            for directory in output_path.iterdir()
+            if directory.is_dir() and (directory / "__init__.py").exists()
+        ]
+    if not packages:
+        return
+
+    imported_modules: list[str] = []
+    start_time = time.perf_counter()
+    try:
+        for parent_directory, package_name in packages:
+            package_path = parent_directory / package_name
+            sys.path.insert(0, str(parent_directory))
+            spec = importlib.util.spec_from_file_location(
+                package_name, package_path / "__init__.py", submodule_search_locations=[str(package_path)]
+            )
+            if spec is None or spec.loader is None:
+                continue
+            module = importlib.util.module_from_spec(spec)
+            sys.modules[package_name] = module
+            imported_modules.append(package_name)
+            spec.loader.exec_module(module)
+
+            for python_file in package_path.rglob("*.py"):
+                if python_file.name == "__init__.py":
+                    continue
+                relative_path = python_file.relative_to(package_path)
+                module_name = f"{package_name}.{'.'.join(relative_path.with_suffix('').parts)}"
+                submodule_spec = importlib.util.spec_from_file_location(module_name, python_file)
+                if submodule_spec is None or submodule_spec.loader is None:
+                    continue
+                submodule = importlib.util.module_from_spec(submodule_spec)
+                sys.modules[module_name] = submodule
+                imported_modules.append(module_name)
+                submodule_spec.loader.exec_module(submodule)
+        _validation_stats.record_exec(time.perf_counter() - start_time)
+    except Exception as exception:
+        _validation_stats.record_error(str(output_path), f"{type(exception).__name__}: {exception}")
+        raise
+    finally:
+        for parent_directory, _ in packages:
+            if str(parent_directory) in sys.path:
+                sys.path.remove(str(parent_directory))
+        for module_name in imported_modules:
+            sys.modules.pop(module_name, None)
+
 
 def run_main_url_and_assert(
     *,
@@ -361,3 +487,5 @@ def run_main_url_and_assert(
     return_code = _run_main_url(url, output_path, input_file_type, extra_args=extra_args)
     _assert_exit_code(return_code, Exit.OK, f"URL: {url}")
     assert_func(output_path, expected_file, transform=transform)
+
+    _validate_output_files(output_path, extra_args)

tests/main/graphql/test_main_graphql.py

@@ -39,7 +39,7 @@ def test_main_graphql_simple_star_wars(output_model: str, expected_output: str,
         input_file_type="graphql",
         assert_func=assert_file_content,
         expected_file=expected_output,
-        extra_args=["--output-model", output_model],
+        extra_args=["--output-model-type", output_model],
     )
 
 
@@ -323,7 +323,7 @@ def test_main_graphql_dataclass_arguments(output_file: Path) -> None:
         assert_func=assert_file_content,
         expected_file="simple_star_wars_dataclass_arguments.py",
         extra_args=[
-            "--output-model",
+            "--output-model-type",
             "dataclasses.dataclass",
             "--dataclass-arguments",
             '{"slots": true, "order": true}',
@@ -347,7 +347,7 @@ def test_main_graphql_dataclass_arguments_with_pydantic(output_file: Path) -> No
         assert_func=assert_file_content,
         expected_file="simple_star_wars.py",
         extra_args=[
-            "--output-model",
+            "--output-model-type",
             "pydantic.BaseModel",
             "--dataclass-arguments",
             '{"slots": true, "order": true}',
@@ -372,7 +372,7 @@ def test_main_graphql_dataclass_frozen_keyword_only(output_file: Path) -> None:
         assert_func=assert_file_content,
         expected_file="simple_star_wars_dataclass_frozen_kw_only.py",
         extra_args=[
-            "--output-model",
+            "--output-model-type",
             "dataclasses.dataclass",
             "--frozen",
             "--keyword-only",

tests/main/jsonschema/test_main_jsonschema.py

@@ -270,7 +270,7 @@ def test_main_jsonschema_dataclass_arguments_with_pydantic(output_file: Path) ->
         assert_func=assert_file_content,
         expected_file="general.py",
         extra_args=[
-            "--output-model",
+            "--output-model-type",
             "pydantic.BaseModel",
             "--dataclass-arguments",
             '{"slots": true, "order": true}',
@@ -291,7 +291,7 @@ def test_main_jsonschema_dataclass_frozen_keyword_only(output_file: Path) -> Non
         assert_func=assert_file_content,
         expected_file="general_dataclass_frozen_kw_only.py",
         extra_args=[
-            "--output-model",
+            "--output-model-type",
             "dataclasses.dataclass",
             "--frozen",
             "--keyword-only",
@@ -403,7 +403,7 @@ def test_main_null_and_array(output_model: str, expected_output: str, output_fil
         input_file_type="jsonschema",
         assert_func=assert_file_content,
         expected_file=expected_output,
-        extra_args=["--output-model", output_model],
+        extra_args=["--output-model-type", output_model],
     )
 
 
@@ -443,7 +443,7 @@ def test_main_complicated_enum_default_member(
     output_model: str, expected_output: str, option: str | None, output_file: Path
 ) -> None:
     """Test complicated enum with default member."""
-    extra_args = [a for a in [option, "--output-model", output_model] if a]
+    extra_args = [a for a in [option, "--output-model-type", output_model] if a]
     run_main_and_assert(
         input_path=JSON_SCHEMA_DATA_PATH / "complicated_enum.json",
         output_path=output_file,
@@ -1672,7 +1672,7 @@ def test_main_jsonschema_combine_any_of_object(
     union_mode: str | None, output_model: str, expected_output: str, output_file: Path
 ) -> None:
     """Test combining anyOf with objects."""
-    extra_args = ["--output-model", output_model]
+    extra_args = ["--output-model-type", output_model]
     if union_mode is not None:
         extra_args.extend(["--union-mode", union_mode])
     run_main_and_assert(
@@ -1689,9 +1689,9 @@ def test_main_jsonschema_combine_any_of_object(
 @pytest.mark.parametrize(
     ("extra_args", "expected_file"),
     [
-        (["--output-model", "pydantic_v2.BaseModel"], "jsonschema_root_model_ordering.py"),
+        (["--output-model-type", "pydantic_v2.BaseModel"], "jsonschema_root_model_ordering.py"),
         (
-            ["--output-model", "pydantic_v2.BaseModel", "--keep-model-order"],
+            ["--output-model-type", "pydantic_v2.BaseModel", "--keep-model-order"],
             "jsonschema_root_model_ordering_keep_model_order.py",
         ),
     ],
@@ -1745,7 +1745,7 @@ def test_main_jsonschema_field_extras_field_include_all_keys(
         assert_func=assert_file_content,
         expected_file=expected_output,
         extra_args=[
-            "--output-model",
+            "--output-model-type",
             output_model,
             "--field-include-all-keys",
             "--field-extra-keys-without-x-prefix",
@@ -1778,7 +1778,7 @@ def test_main_jsonschema_field_extras_field_extra_keys(
         assert_func=assert_file_content,
         expected_file=expected_output,
         extra_args=[
-            "--output-model",
+            "--output-model-type",
             output_model,
             "--field-extra-keys",
             "key2",
@@ -1810,7 +1810,7 @@ def test_main_jsonschema_field_extras(output_model: str, expected_output: str, o
         input_file_type="jsonschema",
         assert_func=assert_file_content,
         expected_file=expected_output,
-        extra_args=["--output-model", output_model],
+        extra_args=["--output-model-type", output_model],
     )