feat(hot-reload): Add HotModule dataclass and HotReloader registry
This commit is contained in:
@@ -0,0 +1,32 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
from dataclasses import dataclass, field
|
||||||
|
from typing import Any
|
||||||
|
import copy
|
||||||
|
import traceback
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class HotModule:
|
||||||
|
name: str
|
||||||
|
file_path: str
|
||||||
|
state_keys: list[str] = field(default_factory=list)
|
||||||
|
delegation_targets: list[str] = field(default_factory=list)
|
||||||
|
|
||||||
|
class HotReloader:
|
||||||
|
HOT_MODULES: dict[str, HotModule] = {}
|
||||||
|
last_error: str | None = None
|
||||||
|
is_error_state: bool = False
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def register(cls, module: HotModule) -> None:
|
||||||
|
if module.name in cls.HOT_MODULES:
|
||||||
|
raise ValueError(f"Module {module.name} already registered")
|
||||||
|
cls.HOT_MODULES[module.name] = module
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def capture_state(cls, app: Any, state_keys: list[str]) -> dict[str, Any]:
|
||||||
|
return {key: copy.deepcopy(getattr(app, key, None)) for key in state_keys if hasattr(app, key)}
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def restore_state(cls, app: Any, state: dict[str, Any]) -> None:
|
||||||
|
for key, value in state.items():
|
||||||
|
setattr(app, key, value)
|
||||||
@@ -0,0 +1,35 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
import pytest
|
||||||
|
from src.hot_reloader import HotModule, HotReloader
|
||||||
|
|
||||||
|
def test_hot_module_dataclass_fields():
|
||||||
|
hm = HotModule(
|
||||||
|
name="test_module",
|
||||||
|
file_path="/path/to/test_module.py",
|
||||||
|
state_keys=["attr1", "attr2"],
|
||||||
|
delegation_targets=["method_a", "method_b"],
|
||||||
|
)
|
||||||
|
assert hm.name == "test_module"
|
||||||
|
assert hm.file_path == "/path/to/test_module.py"
|
||||||
|
assert hm.state_keys == ["attr1", "attr2"]
|
||||||
|
assert hm.delegation_targets == ["method_a", "method_b"]
|
||||||
|
|
||||||
|
def test_hot_reloader_register_and_get():
|
||||||
|
HotReloader.HOT_MODULES.clear()
|
||||||
|
hm = HotModule(name="test_mod", file_path="/fake/path.py", state_keys=[], delegation_targets=[])
|
||||||
|
HotReloader.register(hm)
|
||||||
|
assert "test_mod" in HotReloader.HOT_MODULES
|
||||||
|
assert HotReloader.HOT_MODULES["test_mod"] is hm
|
||||||
|
|
||||||
|
def test_hot_reloader_register_duplicate_raises():
|
||||||
|
HotReloader.HOT_MODULES.clear()
|
||||||
|
hm1 = HotModule(name="dup", file_path="/a.py", state_keys=[], delegation_targets=[])
|
||||||
|
HotReloader.register(hm1)
|
||||||
|
with pytest.raises(ValueError, match="already registered"):
|
||||||
|
HotReloader.register(hm1)
|
||||||
|
|
||||||
|
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
|
||||||
Reference in New Issue
Block a user