fix(core): roll back entity.py when the generated model cannot import

- ast.parse only proves syntax, not that the module imports
- restore the previous file content if importlib.reload raises
- reload again after the rollback so the module works in memory
- re-raise the original import error instead of masking it

Author: LukasGold

src/osw/core.py

@@ -162,6 +162,43 @@ def ensure_valid_python_source(content: str, path: str) -> None:
         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"""
@@ -1112,11 +1149,17 @@ def _fetch_schema(
             # 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

@@ -14,10 +14,30 @@
 """
 
 import ast
+import importlib
+import sys
 
 import pytest
 
-from osw.core import ensure_valid_python_source, remove_unserializable_default_sentinels
+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():
@@ -95,3 +115,43 @@ def test_validation_failure_leaves_an_existing_target_untouched(tmp_path):
         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)