Guard mutable generation inputs

src/datamodel_code_generator/parser/jsonschema.py

@@ -281,19 +281,39 @@ def validate_exclusive_maximum_and_exclusive_minimum(cls, values: Any) -> Any:
             return values
         exclusive_maximum: float | bool | None = values.get("exclusiveMaximum")
         exclusive_minimum: float | bool | None = values.get("exclusiveMinimum")
+        if not isinstance(exclusive_maximum, bool) and not isinstance(exclusive_minimum, bool):
+            return values
 
-        if exclusive_maximum is True:
-            values["exclusiveMaximum"] = values["maximum"]
-            del values["maximum"]
-        elif exclusive_maximum is False:
-            del values["exclusiveMaximum"]
-        if exclusive_minimum is True:
-            values["exclusiveMinimum"] = values["minimum"]
-            del values["minimum"]
-        elif exclusive_minimum is False:
-            del values["exclusiveMinimum"]
+        values = dict(values)
+        match exclusive_maximum:
+            case True:
+                values["exclusiveMaximum"] = values["maximum"]
+                del values["maximum"]
+            case False:
+                del values["exclusiveMaximum"]
+        match exclusive_minimum:
+            case True:
+                values["exclusiveMinimum"] = values["minimum"]
+                del values["minimum"]
+            case False:
+                del values["exclusiveMinimum"]
         return values
 
+    @model_validator(mode="before")
+    @classmethod
+    def collect_extra_fields(cls, values: Any) -> Any:
+        """Collect raw schema extension fields without overriding known schema fields."""
+        if not isinstance(values, dict):
+            return values
+        alias_extras = values.get(cls.__extra_key__, {})
+        raw_extras = {k: v for k, v in values.items() if k not in EXCLUDE_FIELD_KEYS}
+        if not alias_extras and not raw_extras:
+            return values
+        extras = {**alias_extras, **raw_extras}
+        if "const" in alias_extras:  # pragma: no cover
+            extras["const"] = alias_extras["const"]
+        return {**values, cls.__extra_key__: extras}
+
     @field_validator("ref")
     def validate_ref(cls, value: Any) -> Any:  # noqa: N805
         """Validate and normalize $ref values."""
@@ -402,23 +422,8 @@ def validate_null_type(cls, value: Any) -> Any:  # noqa: N805
         ignored_types=(cached_property,),
     )
 
-    def __init__(self, **data: Any) -> None:
-        """Initialize JsonSchemaObject with extra fields handling."""
-        super().__init__(**data)
-        items = data.get("items")
-        if items is False:
-            self.items = False
-        elif items == []:
-            self.items = []
-        # Restore extras from alias key (for dict -> parse_obj round-trip)
-        alias_extras = data.get(self.__extra_key__, {})
-        # Collect custom keys from raw data
-        raw_extras = {k: v for k, v in data.items() if k not in EXCLUDE_FIELD_KEYS}
-        if alias_extras or raw_extras:
-            # Merge: raw_extras takes precedence (original data is the source of truth)
-            self.extras = {**alias_extras, **raw_extras}
-            if "const" in alias_extras:  # pragma: no cover
-                self.extras["const"] = alias_extras["const"]
+    def model_post_init(self, __context: Any, /) -> None:
+        """Apply post-validation compatibility handling for extension metadata."""
         # Support x-propertyNames extension for OpenAPI 3.0
         if "x-propertyNames" in self.extras and self.propertyNames is None:
             x_prop_names = self.extras.pop("x-propertyNames")
@@ -448,7 +453,7 @@ def ref_object_name(self) -> str:  # pragma: no cover
     def validate_items(cls, values: Any) -> Any:  # noqa: N805
         """Validate items field, converting empty dicts to None."""
         # this condition expects empty dict
-        return values or None
+        return None if values == {} else values
 
     @cached_property
     def has_default(self) -> bool:
@@ -619,6 +624,7 @@ def _validate_schema_python_import_path(value: Any, field_name: str) -> str:
     "$recursiveAnchor",
     "$dynamicRef",
     "$dynamicAnchor",
+    "self",
     JsonSchemaObject.__extra_key__,
 }
 
@@ -5478,8 +5484,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,25 @@ def assert_output(
     _assert_with_external_file(output, expected_path)
 
 
+def _tracks_mutation(value: object) -> bool:
+    return isinstance(value, (dict, list))
+
+
+@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,109 @@ def test_main_generate_without_input_file_type(output_file: Path) -> None:
     )
 
 
+def test_generate_keeps_cached_yaml_root_original_unchanged(tmp_path: Path) -> None:
+    """Keep cached parsed YAML root unchanged while ignoring root self metadata."""
+    schema_path = tmp_path / "schema.yaml"
+    output_path = tmp_path / "output.py"
+    expected_path = tmp_path / "expected.py"
+    schema_path.write_text(
+        """
+self:
+  version: metadata
+title: Root
+type: object
+properties:
+  name:
+    type: string
+""".lstrip(),
+        encoding="utf-8",
+    )
+    cached_schema = load_data_from_path(schema_path.resolve(), "utf-8")
+    expected_path.write_text(
+        """# generated by datamodel-codegen:
+#   filename:  schema.yaml
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class Root(BaseModel):
+    name: str | 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 schema.yaml": cached_schema},
+    )
+
+
+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."""