feat(config): add hot-reload via watchfiles (#78)

tests/test_config_watch.py

@@ -0,0 +1,107 @@
+from pathlib import Path
+
+import anyio
+import pytest
+
+import takopi.config_watch as config_watch
+from takopi.config_watch import ConfigReload, _config_status, watch_config
+from takopi.config import empty_projects_config
+from takopi.router import AutoRouter, RunnerEntry
+from takopi.runtime_loader import RuntimeSpec
+from takopi.runners.mock import Return, ScriptRunner
+from takopi.settings import TakopiSettings
+from takopi.transport_runtime import TransportRuntime
+
+
+def test_config_status_variants(tmp_path: Path) -> None:
+    missing = tmp_path / "missing.toml"
+    status, signature = _config_status(missing)
+    assert status == "missing"
+    assert signature is None
+
+    directory = tmp_path / "config.d"
+    directory.mkdir()
+    status, signature = _config_status(directory)
+    assert status == "invalid"
+    assert signature is None
+
+    config_file = tmp_path / "takopi.toml"
+    config_file.write_text('transport = "telegram"\n', encoding="utf-8")
+    status, signature = _config_status(config_file)
+    assert status == "ok"
+    assert signature is not None
+
+
+@pytest.mark.anyio
+async def test_watch_config_applies_runtime(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    config_path = tmp_path / "takopi.toml"
+    config_path.write_text('default_engine = "codex"\n', encoding="utf-8")
+    resolved_path = config_path.resolve()
+
+    codex_runner = ScriptRunner([Return(answer="ok")], engine="codex")
+    router = AutoRouter(
+        entries=[RunnerEntry(engine=codex_runner.engine, runner=codex_runner)],
+        default_engine=codex_runner.engine,
+    )
+    runtime = TransportRuntime(
+        router=router,
+        projects=empty_projects_config(),
+        config_path=resolved_path,
+    )
+
+    pi_runner = ScriptRunner([Return(answer="ok")], engine="pi")
+    new_router = AutoRouter(
+        entries=[RunnerEntry(engine=pi_runner.engine, runner=pi_runner)],
+        default_engine=pi_runner.engine,
+    )
+    new_spec = RuntimeSpec(
+        router=new_router,
+        projects=empty_projects_config(),
+        allowlist=None,
+        plugin_configs=None,
+    )
+    reload = ConfigReload(
+        settings=TakopiSettings.model_validate({"transport": "telegram"}),
+        runtime_spec=new_spec,
+        config_path=resolved_path,
+    )
+
+    ready = anyio.Event()
+    watching = anyio.Event()
+
+    async def fake_awatch(_path: Path):
+        watching.set()
+        await ready.wait()
+        yield {(None, str(resolved_path))}
+
+    monkeypatch.setattr(config_watch, "awatch", fake_awatch)
+    monkeypatch.setattr(
+        config_watch, "_reload_config", lambda *_args, **_kwargs: reload
+    )
+
+    reloaded = anyio.Event()
+
+    async def on_reload(_payload: ConfigReload) -> None:
+        reloaded.set()
+
+    async with anyio.create_task_group() as tg:
+
+        async def run_watch() -> None:
+            await watch_config(
+                config_path=resolved_path,
+                runtime=runtime,
+                on_reload=on_reload,
+            )
+
+        tg.start_soon(run_watch)
+        with anyio.fail_after(2):
+            await watching.wait()
+        config_path.write_text('default_engine = "pi"\n', encoding="utf-8")
+        ready.set()
+        with anyio.fail_after(2):
+            await reloaded.wait()
+        tg.cancel_scope.cancel()
+
+    assert runtime.default_engine == "pi"

tests/test_runtime_loader.py

@@ -0,0 +1,36 @@
+from pathlib import Path
+
+import pytest
+
+import takopi.runtime_loader as runtime_loader
+from takopi.config import ConfigError
+from takopi.settings import TakopiSettings
+
+
+def test_build_runtime_spec_minimal(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setattr(runtime_loader.shutil, "which", lambda _cmd: "/bin/echo")
+    settings = TakopiSettings.model_validate({"transport": "telegram"})
+    config_path = tmp_path / "takopi.toml"
+    config_path.write_text('transport = "telegram"\n', encoding="utf-8")
+
+    spec = runtime_loader.build_runtime_spec(
+        settings=settings,
+        config_path=config_path,
+    )
+
+    assert spec.router.default_engine == settings.default_engine
+    runtime = spec.to_runtime(config_path=config_path)
+    assert runtime.default_engine == settings.default_engine
+
+
+def test_resolve_default_engine_unknown(tmp_path: Path) -> None:
+    settings = TakopiSettings.model_validate({"transport": "telegram"})
+    with pytest.raises(ConfigError, match="Unknown default engine"):
+        runtime_loader.resolve_default_engine(
+            override="unknown",
+            settings=settings,
+            config_path=tmp_path / "takopi.toml",
+            engine_ids=["codex"],
+        )