fix(core): never write unparseable entity.py from fetch_schema (#167)

- strip datamodel-code-generator sentinel defaults before writing
- validate the generated source with ast.parse before opening the file
- roll back the previous entity.py when the new model cannot import
- log black/isort failures instead of discarding them

Author: LukasGold

src/osw/core.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import ast
 import importlib
 import json
 import logging
@@ -113,6 +114,93 @@ def collect_messages(
     return target
 
 
+def remove_unserializable_default_sentinels(content: str) -> str:
+    """Replaces defaults that repr a datamodel-code-generator sentinel object
+
+    datamodel-code-generator uses a bare `UNDEFINED = object()` sentinel with
+    `is` identity checks. oold's merge_deep deep-copies schema dicts during
+    allOf composition, which clones that sentinel into a look-alike object,
+    defeating the identity guard, so the generator reprs it into source as
+    `default_factory=lambda :Foo.parse_obj(<object object at 0x...>)`, which
+    is not valid Python.
+
+    Based on oold.generator.Generator.generate(), but matching the factory
+    expression itself rather than everything up to the next closing paren:
+    oold's pattern consumes the paren belonging to `parse_obj(`, which leaves
+    a dangling `)` behind whenever the sentinel is wrapped in a call.
+    """
+    return re.sub(
+        r"default_factory=lambda\s*:\s*"
+        r"(?:[\w.]+\(<object object at 0x[0-9a-fA-F]+>\)"
+        r"|<object object at 0x[0-9a-fA-F]+>)",
+        "default=None",
+        content,
+    )
+
+
+def ensure_valid_python_source(content: str, path: str) -> None:
+    """Raises a descriptive SyntaxError if content is not valid Python
+
+    _fetch_schema writes the generated model to a file that is imported right
+    after (or, for non-final calls, on the next process start). A corrupt
+    write poisons every later import of osw.model.entity, and there is
+    previously-valid content sitting at `path` that would otherwise still
+    work. Validating before opening the file for writing means a bad
+    generation leaves that previous, valid content in place instead.
+    """
+    try:
+        ast.parse(content)
+    except SyntaxError as e:
+        offending_line = ""
+        if e.lineno is not None:
+            lines = content.splitlines()
+            if 0 < e.lineno <= len(lines):
+                offending_line = lines[e.lineno - 1]
+        message = (
+            f"Generated model for '{path}' is not valid Python: "
+            f"{e.msg} (line {e.lineno}): {offending_line!r}"
+        )
+        _logger.error(message)
+        raise SyntaxError(message) from e
+
+
+def reload_module_or_restore(module, path: str, previous_content: str = None) -> None:
+    """Reloads module, putting previous_content back if the import fails
+
+    ast.parse only proves the generated model is syntactically valid. It can
+    still fail at import time, e.g. on an undefined name or an error raised
+    while a class body is executed. Restoring the previous file content keeps
+    later imports of osw.model.entity working instead of leaving a module on
+    disk that raises for the rest of the installation's lifetime.
+    """
+    try:
+        importlib.reload(module)
+    except Exception as e:
+        _logger.error(f"Generated model at '{path}' failed to import: {e}")
+        if previous_content is not None:
+            _logger.error(f"Restoring the previous content of '{path}'")
+            with open(path, "w", encoding="utf-8") as f:
+                f.write(previous_content)
+            try:
+                importlib.reload(module)
+            except Exception as restore_error:
+                # do not mask the original failure, but make it obvious that
+                # the module is now broken in memory as well
+                _logger.error(
+                    f"Restoring '{path}' did not make it importable again: "
+                    f"{restore_error}"
+                )
+        raise
+
+
+def read_file_if_exists(path: str) -> str:
+    """Returns the content of path, or None if it does not exist yet"""
+    if not os.path.exists(path):
+        return None
+    with open(path, encoding="utf-8") as f:
+        return f.read()
+
+
 # Reusable type definitions
 class OverwriteOptions(Enum):
     """Options for overwriting properties"""
@@ -845,6 +933,9 @@ def _fetch_schema(
             # are not v1 compatible mainly by using update_model()
             content = re.sub(r"(,?\s*unique_items=True\s*)", "", content)
 
+            # fix unserializable defaults from datamodel-code-generator (#125)
+            content = remove_unserializable_default_sentinels(content)
+
             # Detect empty subclasses, replaces their occurrences with base classes,
             # and removes the empty class definitions.
             # Only processes subclasses that follow naming patterns:
@@ -1053,14 +1144,27 @@ def _fetch_schema(
                     content = black.format_str(content, mode=black.Mode())
                     # run isort to sort imports using Vertical Hanging Indent style
                     content = isort.code(content, profile="black")
-                except Exception:
-                    pass  # black is optional, continue without formatting
+                except Exception as e:
+                    # black/isort are optional, continue without formatting, but
+                    # do not hide a signal that the generated content is broken
+                    _logger.warning(f"Failed to format generated model content: {e}")
+
+            # validate before writing: a corrupt write poisons every later
+            # import of this file, so leaving the previous valid content in
+            # place is strictly better than writing invalid syntax (#125)
+            ensure_valid_python_source(content, result_model_path)
+
+            # keep the current file so that a model that parses but does not
+            # import can be rolled back below (#125)
+            previous_content = read_file_if_exists(result_model_path)
 
             with open(result_model_path, "w", encoding="utf-8") as f:
                 f.write(content)
 
             if fetchSchemaParam.final:
-                importlib.reload(model)  # reload the updated module
+                # reload the updated module, restoring the previous content if
+                # the generated model turns out not to be importable
+                reload_module_or_restore(model, result_model_path, previous_content)
                 if not site_cache_state:
                     self.site.disable_cache()  # restore original state
 

tests/test_fetch_schema_write_safety.py

@@ -0,0 +1,157 @@
+"""Unit tests for the sentinel-cleanup and syntax-validation guards in osw.core.
+
+Regression guard for #125: datamodel-code-generator uses a bare
+`UNDEFINED = object()` sentinel with `is` identity checks. oold's merge_deep
+deep-copies schema dicts during allOf composition, which clones that sentinel
+into a look-alike object, defeating the identity guard, so the generator
+repr's it into source (see oold.generator.Generator.generate() for oold's own
+workaround). _fetch_schema used to write that content straight to entity.py
+and reload it, raising SyntaxError and poisoning every later `import
+osw.core` in the process.
+
+These tests exercise the extracted helpers directly, fully offline, and
+never call the real (wiki- and network-backed) `_fetch_schema`.
+"""
+
+import ast
+import importlib
+import sys
+
+import pytest
+
+from osw.core import (
+    ensure_valid_python_source,
+    reload_module_or_restore,
+    remove_unserializable_default_sentinels,
+)
+
+
+@pytest.fixture
+def throwaway_module(tmp_path):
+    """An importable module on disk, cleaned out of sys.modules afterwards"""
+    name = "osw_test_throwaway_model"
+    path = tmp_path / f"{name}.py"
+    path.write_text("VALUE = 1\n", encoding="utf-8")
+    sys.path.insert(0, str(tmp_path))
+    try:
+        yield importlib.import_module(name), path
+    finally:
+        sys.path.remove(str(tmp_path))
+        sys.modules.pop(name, None)
+
+
+def test_sentinel_default_is_rewritten_to_valid_python():
+    """A repr'd sentinel object breaks ast.parse until the substitution runs."""
+    bad = (
+        "risk_assessment: RiskAssessmentProcess | None = Field("
+        "default_factory=lambda :<object object at 0x000001A2B3C4D5E6>)\n"
+    )
+    with pytest.raises(SyntaxError):
+        ast.parse(bad)
+
+    fixed = remove_unserializable_default_sentinels(bad)
+
+    assert "<object object at" not in fixed
+    ast.parse(fixed)  # must not raise
+
+
+def test_sentinel_wrapped_in_a_parse_obj_call_is_rewritten():
+    """The shape actually reported in #125, where the sentinel sits inside a
+    `parse_obj(...)` call and the field carries further keyword arguments.
+
+    oold's own regex stops at the first `)` after the address, which is the
+    one belonging to `parse_obj(`, and so leaves a dangling `)` behind.
+    """
+    bad = (
+        "risk_assessment: RiskAssessmentProcess | None = Field("
+        "default_factory=lambda :RiskAssessmentProcess.parse_obj("
+        "<object object at 0x000001A2B3C4D5E6>), options={'a': 1})\n"
+    )
+    with pytest.raises(SyntaxError):
+        ast.parse(bad)
+
+    fixed = remove_unserializable_default_sentinels(bad)
+
+    assert "<object object at" not in fixed
+    assert "options={'a': 1}" in fixed  # trailing kwargs are preserved
+    ast.parse(fixed)  # must not raise
+
+
+def test_legitimate_default_factory_lambda_is_not_mangled():
+    """A normal `default_factory=lambda: uuid4()` must survive untouched."""
+    legit = "id: UUID = Field(default_factory=lambda: uuid4())\n"
+
+    assert remove_unserializable_default_sentinels(legit) == legit
+
+
+def test_ensure_valid_python_source_accepts_valid_source():
+    ensure_valid_python_source("class Foo:\n    pass\n", "entity.py")  # no raise
+
+
+def test_ensure_valid_python_source_raises_with_a_useful_message():
+    bad = "class Foo(:\n    pass\n"
+
+    with pytest.raises(SyntaxError) as exc_info:
+        ensure_valid_python_source(bad, "entity.py")
+
+    message = str(exc_info.value)
+    assert "entity.py" in message
+    assert "line 1" in message
+    assert "class Foo(:" in message
+
+
+def test_validation_failure_leaves_an_existing_target_untouched(tmp_path):
+    """Mirrors the guarded write in _fetch_schema: validation runs before the
+    file is ever opened for writing, so a bad generation never touches the
+    previous, valid content sitting at the target path.
+    """
+    target = tmp_path / "entity.py"
+    target.write_text("previous_valid_content = 1\n", encoding="utf-8")
+
+    bad = "class Foo(:\n    pass\n"
+
+    with pytest.raises(SyntaxError):
+        ensure_valid_python_source(bad, str(target))
+        target.write_text("corrupted", encoding="utf-8")  # never reached
+
+    assert target.read_text(encoding="utf-8") == "previous_valid_content = 1\n"
+
+
+def test_reload_restores_previous_content_when_the_new_model_cannot_import(
+    throwaway_module,
+):
+    """Syntactically valid content can still fail at import time. The file must
+    be rolled back so later imports keep working.
+    """
+    module, path = throwaway_module
+    previous_content = path.read_text(encoding="utf-8")
+    broken = "raise RuntimeError('not importable')\n"
+    ast.parse(broken)  # passes the syntax guard, so only the import catches it
+    path.write_text(broken, encoding="utf-8")
+
+    with pytest.raises(RuntimeError):
+        reload_module_or_restore(module, str(path), previous_content)
+
+    assert path.read_text(encoding="utf-8") == previous_content
+    assert module.VALUE == 1  # the in-memory module works again too
+
+
+def test_reload_keeps_the_new_model_when_it_imports(throwaway_module):
+    module, path = throwaway_module
+    # the length has to differ from the original source, otherwise the pyc
+    # cache (keyed on mtime and size) can survive the reload
+    path.write_text("VALUE = 222\n", encoding="utf-8")
+
+    reload_module_or_restore(module, str(path), "VALUE = 1\n")
+
+    assert path.read_text(encoding="utf-8") == "VALUE = 222\n"
+    assert module.VALUE == 222
+
+
+def test_reload_without_previous_content_still_raises(throwaway_module):
+    """First-ever write has nothing to roll back to, but must not fail silently."""
+    module, path = throwaway_module
+    path.write_text("raise RuntimeError('not importable')\n", encoding="utf-8")
+
+    with pytest.raises(RuntimeError):
+        reload_module_or_restore(module, str(path), None)