code-server
e8e8ca6700
The HeartbeatService constructor was refactored to use an on_heartbeat callback instead of accepting provider/model parameters directly. This commit updates the tests to match the new API: - Removed DummyProvider class (no longer needed) - Updated test_start_is_idempotent to use new constructor - Removed test_decide_returns_skip_when_no_tool_call (_decide method no longer exists) - Updated test_trigger_now_executes_when_decision_is_run to use on_heartbeat callback - Updated test_trigger_now_returns_none_when_no_callback to test new behavior Also fixed a bug where start() was not idempotent - it now checks if a task is already running before creating a new one. All 3 HeartbeatService tests now pass. Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
59 lines
1.5 KiB
Python
59 lines
1.5 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
from nanobot.heartbeat.service import HeartbeatService
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_is_idempotent(tmp_path) -> None:
|
|
service = HeartbeatService(
|
|
workspace=tmp_path,
|
|
interval_s=9999,
|
|
enabled=True,
|
|
)
|
|
|
|
await service.start()
|
|
first_task = service._task
|
|
await service.start()
|
|
|
|
assert service._task is first_task
|
|
|
|
service.stop()
|
|
await asyncio.sleep(0)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_now_executes_when_decision_is_run(tmp_path) -> None:
|
|
(tmp_path / "HEARTBEAT.md").write_text("- [ ] do thing", encoding="utf-8")
|
|
|
|
called_with: list[tuple[str, dict | None]] = []
|
|
|
|
async def _on_heartbeat(prompt: str, metadata: dict | None = None) -> str:
|
|
called_with.append((prompt, metadata))
|
|
return "done"
|
|
|
|
service = HeartbeatService(
|
|
workspace=tmp_path,
|
|
on_heartbeat=_on_heartbeat,
|
|
)
|
|
|
|
result = await service.trigger_now()
|
|
assert result == "done"
|
|
assert len(called_with) == 1
|
|
prompt, metadata = called_with[0]
|
|
assert "HEARTBEAT.md" in prompt
|
|
assert metadata == {"suppress_output": True}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_now_returns_none_when_no_callback(tmp_path) -> None:
|
|
(tmp_path / "HEARTBEAT.md").write_text("- [ ] do thing", encoding="utf-8")
|
|
|
|
service = HeartbeatService(
|
|
workspace=tmp_path,
|
|
on_heartbeat=None, # No callback
|
|
)
|
|
|
|
assert await service.trigger_now() is None
|