Guard mutable generation inputs

Author: koxudaxi

src/datamodel_code_generator/parser/jsonschema.py

@@ -23,6 +23,7 @@
 from pydantic import (
     ConfigDict,
     Field,
+    ValidationError,
     field_validator,
     model_validator,
 )
@@ -5100,6 +5101,19 @@ def _validate_schema_object(
             return self.SCHEMA_OBJECT_TYPE(is_boolean_schema_false=True)
         try:
             return self.SCHEMA_OBJECT_TYPE.model_validate(raw)
+        except TypeError as e:
+            if isinstance(raw, dict) and "self" in raw and "multiple values for argument 'self'" in str(e):
+                raw_without_self = raw.copy()
+                raw_without_self.pop("self", None)
+                try:
+                    return self.SCHEMA_OBJECT_TYPE.model_validate(raw_without_self)
+                except (TypeError, ValidationError) as nested_e:
+                    e = nested_e
+            raise SchemaParseError(
+                message=f"{type(e).__name__}: {e}",
+                path=path,
+                original_error=e,
+            ) from e
         except SchemaParseError:
             raise
         except Exception as e:
@@ -5459,8 +5473,6 @@ def _parse_file(  # noqa: PLR0912, PLR0913, PLR0914, PLR0915
                 preserve_class_name=preserve_root_class_name,
             ).name
             with self.root_id_context(raw):
-                # Some jsonschema docs include attribute self to have include version details
-                raw.pop("self", None)
                 # parse $id before parsing $ref
                 root_obj = self._validate_schema_object(raw, path_parts or ["#"])
                 self.parse_id(root_obj, [*path_parts, "#"] if path_parts else ["#"])

tests/conftest.py

@@ -12,7 +12,9 @@
 import sys
 import time
 from collections import Counter
-from collections.abc import Callable, Mapping, Sequence
+from collections.abc import Callable, Generator, Mapping, Sequence
+from contextlib import contextmanager
+from copy import deepcopy
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from itertools import starmap
@@ -702,6 +704,28 @@ def assert_output(
     _assert_with_external_file(output, expected_path)
 
 
+def _tracks_mutation(value: object) -> bool:
+    match value:
+        case dict() | list():
+            return True
+    return False
+
+
+@contextmanager
+def assert_inputs_not_mutated(inputs: Mapping[str, object] | None) -> Generator[None, None, None]:
+    """Assert guarded mutable inputs are unchanged after the wrapped operation."""
+    __tracebackhide__ = True
+    if inputs is None:
+        yield
+        return
+
+    snapshots = {label: deepcopy(value) for label, value in inputs.items() if _tracks_mutation(value)}
+    yield
+    for label, before in snapshots.items():
+        if (current := inputs[label]) != before:  # pragma: no cover - exercised by helper tests
+            pytest.fail(f"{label} was mutated: before={before!r}, after={current!r}")
+
+
 def assert_directory_content(
     output_dir: Path,
     expected_dir: Path,

tests/main/conftest.py

@@ -11,7 +11,7 @@
 import time
 import warnings
 from argparse import Namespace
-from collections.abc import Callable, Generator, Sequence
+from collections.abc import Callable, Generator, Mapping, Sequence
 from contextlib import contextmanager
 from functools import cache
 from pathlib import Path
@@ -31,6 +31,7 @@
     _infer_expected_file,
     _validation_stats,
     assert_directory_content,
+    assert_inputs_not_mutated,
     assert_output,
     assert_warnings_contain,
     freeze_time,
@@ -611,6 +612,7 @@ def run_generate_file_and_assert(
     expected_file: str | Path | None = None,
     transform: Callable[[str], str] | None = None,
     expected_warnings: Sequence[str] | None = None,
+    unchanged_inputs: Mapping[str, object] | None = None,
     **generate_kwargs: Any,
 ) -> None:
     """Execute generate() for a file input and assert the generated output."""
@@ -632,19 +634,20 @@ def run_generate_file_and_assert(
     if input_file_type is not None:
         generate_options["input_file_type"] = input_file_type
 
-    if expected_warnings is None:
-        generate(
-            input_=input_,
-            **generate_options,
-        )
-    else:
-        with warnings.catch_warnings(record=True) as warning_records:
-            warnings.simplefilter("always")
+    with assert_inputs_not_mutated(unchanged_inputs):
+        if expected_warnings is None:
             generate(
                 input_=input_,
                 **generate_options,
             )
-        assert_warnings_contain(warning_records, *expected_warnings)
+        else:
+            with warnings.catch_warnings(record=True) as warning_records:
+                warnings.simplefilter("always")
+                generate(
+                    input_=input_,
+                    **generate_options,
+                )
+            assert_warnings_contain(warning_records, *expected_warnings)
 
     if expected_file is None:
         frame = inspect.currentframe()
@@ -654,24 +657,32 @@ def run_generate_file_and_assert(
         del frame
 
     assert_func(output_path, expected_file, transform=transform)
-    _assert_builtin_generate_formatter_parity(
-        input_=input_,
-        output_path=output_path,
-        generate_options=generate_options,
-        expected_warnings=expected_warnings,
-    )
+    with assert_inputs_not_mutated(unchanged_inputs):
+        _assert_builtin_generate_formatter_parity(
+            input_=input_,
+            output_path=output_path,
+            generate_options=generate_options,
+            expected_warnings=expected_warnings,
+        )
 
 
 def run_generate_and_assert(
     *,
     input_: Any,
     expected_file: Path,
+    assert_input_unchanged: bool = False,
+    unchanged_inputs: Mapping[str, object] | None = None,
     **generate_kwargs: Any,
 ) -> None:
     """Execute generate(output=None) and assert the returned text output."""
     __tracebackhide__ = True
 
-    result = generate(input_=input_, **_default_formatter_generate_options(generate_kwargs))
+    guarded_inputs = dict(unchanged_inputs or {})
+    if assert_input_unchanged:
+        guarded_inputs["input_"] = input_
+
+    with assert_inputs_not_mutated(guarded_inputs or None):
+        result = generate(input_=input_, **_default_formatter_generate_options(generate_kwargs))
     if not isinstance(result, str):  # pragma: no cover
         pytest.fail(f"Expected generate() to return str, got {type(result).__name__}")
     assert_output(result, expected_file)

tests/main/jsonschema/test_main_jsonschema.py

@@ -23,6 +23,7 @@
     TargetPydanticVersion,
     chdir,
     generate,
+    load_data_from_path,
 )
 from datamodel_code_generator.__main__ import Exit
 from datamodel_code_generator.format import is_supported_in_black
@@ -33,6 +34,7 @@
     MockHttpxResponse,
     assert_directory_content,
     assert_httpx_get_kwargs,
+    create_assert_file_content,
     freeze_time,
     validate_generated_code,
 )
@@ -2247,6 +2249,65 @@ def test_main_generate_without_input_file_type(output_file: Path) -> None:
     )
 
 
+def test_generate_keeps_cached_yaml_ref_original_unchanged(tmp_path: Path) -> None:
+    """Keep cached parsed YAML references unchanged while resolving JSON Schema refs."""
+    schema_path = tmp_path / "schema.yaml"
+    child_path = tmp_path / "child.yaml"
+    output_path = tmp_path / "output.py"
+    expected_path = tmp_path / "expected.py"
+    schema_path.write_text(
+        """
+title: Root
+type: object
+properties:
+  child:
+    $ref: child.yaml#
+""".lstrip(),
+        encoding="utf-8",
+    )
+    child_path.write_text(
+        """
+self:
+  version: metadata
+title: Child
+type: object
+properties:
+  name:
+    type: string
+""".lstrip(),
+        encoding="utf-8",
+    )
+    cached_child_schema = load_data_from_path(child_path.resolve(), "utf-8")
+    expected_path.write_text(
+        """# generated by datamodel-codegen:
+#   filename:  schema.yaml
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class Child(BaseModel):
+    name: str | None = None
+
+
+class Root(BaseModel):
+    child: Child | None = None
+""",
+        encoding="utf-8",
+    )
+
+    run_generate_file_and_assert(
+        input_path=schema_path,
+        output_path=output_path,
+        input_file_type=InputFileType.JsonSchema,
+        assert_func=create_assert_file_content(tmp_path),
+        expected_file=expected_path.name,
+        disable_timestamp=True,
+        unchanged_inputs={"cached child.yaml": cached_child_schema},
+    )
+
+
 def test_main_generate_relative_input_path(output_file: Path) -> None:
     """Test helper with a relative input path."""
     run_generate_file_and_assert(

tests/main/protobuf/test_main_protobuf.py

@@ -14,6 +14,7 @@
     EXPECTED_PROTOBUF_PATH,
     PROTOBUF_DATA_PATH,
     assert_input_file_type,
+    run_generate_and_assert,
     run_generate_file_and_assert,
     run_main_and_assert,
 )
@@ -163,14 +164,14 @@ def test_generate_api_protobuf_definition_key_collision() -> None:
 
 def test_generate_api_protobuf_from_path_list() -> None:
     """Generate Protocol Buffers models from a list of .proto file paths."""
-    result = generate(
-        cast(Any, [(PROTOBUF_DATA_PATH / "spec_proto3.proto").resolve()]),  # noqa: TC006
+    run_generate_and_assert(
+        input_=cast(Any, [(PROTOBUF_DATA_PATH / "spec_proto3.proto").resolve()]),  # noqa: TC006
         input_file_type=InputFileType.Protobuf,
         disable_timestamp=True,
+        expected_file=EXPECTED_PROTOBUF_PATH / "spec_proto3_list_input.py",
+        assert_input_unchanged=True,
     )
 
-    assert_output(result, EXPECTED_PROTOBUF_PATH / "spec_proto3_list_input.py")
-
 
 def test_generate_api_protobuf_rejects_dict_input() -> None:
     """Reject mapping input because Protocol Buffers requires .proto text."""

tests/main/test_main_general.py

@@ -2711,6 +2711,7 @@ def test_generate_with_dict_jsonschema() -> None:
         input_file_type=InputFileType.JsonSchema,
         disable_timestamp=True,
         expected_file=EXPECTED_MAIN_PATH / "dict_input" / "jsonschema.py",
+        assert_input_unchanged=True,
     )
 
 
@@ -2723,6 +2724,7 @@ def test_generate_with_dict_openapi() -> None:
         input_file_type=InputFileType.OpenAPI,
         disable_timestamp=True,
         expected_file=EXPECTED_MAIN_PATH / "dict_input" / "openapi.py",
+        assert_input_unchanged=True,
     )
 
 

tests/test_conftest_helpers.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 from types import SimpleNamespace
 from typing import TYPE_CHECKING
-from unittest.mock import call
+from unittest.mock import Mock, call
 
 import pytest
 
@@ -15,6 +15,7 @@
 from tests.conftest import (
     _infer_expected_file,
     assert_exact_directory_content,
+    assert_inputs_not_mutated,
     assert_parser_modules,
     assert_parser_results,
 )
@@ -94,6 +95,80 @@ def test_assert_parser_modules_rejects_unexpected_module(tmp_path: Path) -> None
         assert_parser_modules({("sample.py",): "value = 1\n"}, expected_dir)
 
 
+def test_assert_inputs_not_mutated_allows_unchanged_nested_values() -> None:
+    """Mutation guard accepts unchanged dict/list inputs and ignores immutable labels."""
+    schema = {"properties": {"name": {"type": "string"}}, "required": ["name"]}
+
+    with assert_inputs_not_mutated({"schema": schema, "description": "ignored"}):
+        assert schema["properties"]["name"]["type"] == "string"
+
+
+def test_assert_inputs_not_mutated_reports_nested_mutation() -> None:
+    """Mutation guard reports the label for nested dict/list mutations."""
+    schema = {"properties": {"name": {"type": "string"}}, "required": ["name"]}
+
+    with (
+        pytest.raises(pytest.fail.Exception, match="schema was mutated"),
+        assert_inputs_not_mutated({"schema": schema}),
+    ):
+        schema["required"].append("age")
+
+
+def test_run_generate_and_assert_detects_mutated_input(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """run_generate_and_assert can guard caller-provided dict/list inputs."""
+    schema = {"properties": {"name": {"type": "string"}}, "required": ["name"]}
+
+    def fake_generate(*, input_: object, **_: object) -> str:
+        assert isinstance(input_, dict)
+        input_["required"].append("age")
+        return "value = 1\n"
+
+    monkeypatch.setattr(main_conftest, "generate", fake_generate)
+    assert_output_noop = Mock()
+    monkeypatch.setattr(main_conftest, "assert_output", assert_output_noop)
+
+    with pytest.raises(pytest.fail.Exception, match="input_ was mutated"):
+        main_conftest.run_generate_and_assert(
+            input_=schema,
+            expected_file=tmp_path / "expected.py",
+            assert_input_unchanged=True,
+        )
+    assert_output_noop.assert_not_called()
+
+
+def test_run_generate_file_and_assert_detects_mutated_unchanged_input(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    """run_generate_file_and_assert can guard cached parsed original objects."""
+    cached_schema = {"properties": {"name": {"type": "string"}}, "required": ["name"]}
+    input_path = tmp_path / "schema.json"
+    output_path = tmp_path / "output.py"
+
+    def fake_generate(*, input_: object, **options: object) -> None:
+        assert input_ == input_path
+        cached_schema["required"].append("age")
+        output = options["output"]
+        assert isinstance(output, Path)
+        output.write_text("value = 1\n", encoding="utf-8")
+
+    monkeypatch.setattr(main_conftest, "generate", fake_generate)
+    assert_file_noop = Mock()
+
+    with pytest.raises(pytest.fail.Exception, match="cached schema was mutated"):
+        main_conftest.run_generate_file_and_assert(
+            input_path=input_path,
+            output_path=output_path,
+            assert_func=assert_file_noop,
+            expected_file="unused.py",
+            unchanged_inputs={"cached schema": cached_schema},
+        )
+    assert_file_noop.assert_not_called()
+
+
 def test_builtin_parity_mock_call_preservation(mocker: MockerFixture) -> None:
     """Mock call history is restored after parity-only calls."""
     mocked_callable = mocker.Mock()