Private
Public Access
0
0

feat(hot-reload): Implement HotReloader.reload and reload_all

This commit is contained in:
2026-05-16 01:33:11 -04:00
parent 14d45e9dd0
commit 8260c4a6b9
2 changed files with 102 additions and 2 deletions
+36 -1
View File
@@ -29,4 +29,39 @@ class HotReloader:
@classmethod
def restore_state(cls, app: Any, state: dict[str, Any]) -> None:
for key, value in state.items():
setattr(app, key, value)
setattr(app, key, value)
@classmethod
def reload(cls, module_name: str, app: Any) -> bool:
if module_name not in cls.HOT_MODULES:
cls.last_error = f"Module {module_name} not registered"
cls.is_error_state = True
return False
hm = cls.HOT_MODULES[module_name]
state = cls.capture_state(app, hm.state_keys)
try:
import importlib
import sys
if module_name in sys.modules:
old_module = sys.modules[module_name]
importlib.reload(old_module)
else:
importlib.import_module(module_name)
cls.last_error = None
cls.is_error_state = False
return True
except Exception:
cls.restore_state(app, state)
cls.last_error = traceback.format_exc()
cls.is_error_state = True
return False
@classmethod
def reload_all(cls, app: Any) -> bool:
success = True
for name in cls.HOT_MODULES:
if not cls.reload(name, app):
success = False
return success
+66 -1
View File
@@ -1,6 +1,9 @@
from __future__ import annotations
import pytest
from src.hot_reloader import HotModule, HotReloader
from unittest.mock import MagicMock, patch
import sys
import types
def test_hot_module_dataclass_fields():
hm = HotModule(
@@ -32,4 +35,66 @@ def test_hot_reloader_is_error_state():
HotReloader.HOT_MODULES.clear()
HotReloader.last_error = None
HotReloader.is_error_state = False
assert HotReloader.is_error_state is False
assert HotReloader.is_error_state is False
def test_reload_unknown_module_returns_false():
HotReloader.HOT_MODULES.clear()
HotReloader.register(HotModule(name="nonexistent_mod", file_path="/nonexistent.py", state_keys=[], delegation_targets=[]))
app = MagicMock()
result = HotReloader.reload("nonexistent_mod", app)
assert result is False
assert HotReloader.is_error_state is True
assert HotReloader.last_error is not None
def test_reload_success_clears_error_state():
HotReloader.HOT_MODULES.clear()
test_mod = types.ModuleType("src._test_reload_mod_src")
sys.modules["src._test_reload_mod_src"] = test_mod
HotReloader.register(HotModule(name="src._test_reload_mod_src", file_path="/fake.py", state_keys=[], delegation_targets=[]))
app = MagicMock()
HotReloader.is_error_state = True
HotReloader.last_error = "previous error"
with patch("importlib.reload", return_value=test_mod):
result = HotReloader.reload("src._test_reload_mod_src", app)
assert result is True
assert HotReloader.is_error_state is False
assert HotReloader.last_error is None
del sys.modules["src._test_reload_mod_src"]
def test_reload_captures_and_restores_state_on_failure():
HotReloader.HOT_MODULES.clear()
HotReloader.register(HotModule(name="bad_mod", file_path="/bad.py", state_keys=["_test_attr"], delegation_targets=[]))
app = MagicMock()
app._test_attr = "preserved_value"
result = HotReloader.reload("bad_mod", app)
assert result is False
assert HotReloader.is_error_state is True
assert app._test_attr == "preserved_value"
def test_reload_all_success():
HotReloader.HOT_MODULES.clear()
mod1 = types.ModuleType("hr_test_mod1")
mod2 = types.ModuleType("hr_test_mod2")
sys.modules["hr_test_mod1"] = mod1
sys.modules["hr_test_mod2"] = mod2
HotReloader.register(HotModule(name="hr_test_mod1", file_path="/fake1.py", state_keys=[], delegation_targets=[]))
HotReloader.register(HotModule(name="hr_test_mod2", file_path="/fake2.py", state_keys=[], delegation_targets=[]))
app = MagicMock()
with patch("importlib.reload", return_value=mod1):
result = HotReloader.reload_all(app)
assert result is True
assert HotReloader.is_error_state is False
del sys.modules["hr_test_mod1"]
del sys.modules["hr_test_mod2"]
def test_reload_all_partial_failure():
HotReloader.HOT_MODULES.clear()
mod1 = types.ModuleType("hr_test_mod1")
sys.modules["hr_test_mod1"] = mod1
HotReloader.register(HotModule(name="hr_test_mod1", file_path="/fake1.py", state_keys=[], delegation_targets=[]))
HotReloader.register(HotModule(name="hr_nonexistent", file_path="/nonexistent.py", state_keys=[], delegation_targets=[]))
app = MagicMock()
result = HotReloader.reload_all(app)
assert result is False
assert HotReloader.is_error_state is True
del sys.modules["hr_test_mod1"]