romfast/echo-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
echo-core/tests/test_heartbeat.py
MoltBot Service f8ff971627 refactor(heartbeat): smart calendar with daily summary and dedup reminders 
Calendar no longer bypasses quiet hours. First run after quiet hours
sends full daily summary, subsequent runs only remind for next event
within 45 min with deduplication. Calendar cooldown set to 30 min.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-15 11:09:59 +00:00

540 lines
22 KiB
Python
Raw Permalink Blame History

"""Tests for src/heartbeat.py — Periodic health checks."""
import json
import time
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from src.heartbeat import (
_calendar_daily_summary,
_calendar_next_reminder,
_check_calendar_smart,
_check_email,
_check_git,
_check_kb_index,
_get_extra_instructions,
_is_quiet_hour,
_load_state,
_save_state,
_should_run,
run_heartbeat,
DEFAULT_COOLDOWNS,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def tmp_env(tmp_path, monkeypatch):
"""Redirect PROJECT_ROOT, STATE_FILE, TOOLS_DIR, HEARTBEAT_MD to tmp_path."""
root = tmp_path / "project"
root.mkdir()
tools = root / "tools"
tools.mkdir()
mem = root / "memory"
mem.mkdir()
personality = root / "personality"
personality.mkdir()
state_file = mem / "heartbeat-state.json"
monkeypatch.setattr("src.heartbeat.PROJECT_ROOT", root)
monkeypatch.setattr("src.heartbeat.STATE_FILE", state_file)
monkeypatch.setattr("src.heartbeat.TOOLS_DIR", tools)
monkeypatch.setattr("src.heartbeat.HEARTBEAT_MD", personality / "HEARTBEAT.md")
return {
"root": root, "tools": tools, "memory": mem,
"state_file": state_file, "personality": personality,
}
# ---------------------------------------------------------------------------
# _is_quiet_hour
# ---------------------------------------------------------------------------
class TestIsQuietHour:
"""Test quiet hour detection with overnight and daytime ranges."""
def test_overnight_range_before_midnight(self):
assert _is_quiet_hour(23, (23, 8)) is True
def test_overnight_range_after_midnight(self):
assert _is_quiet_hour(3, (23, 8)) is True
def test_overnight_range_outside(self):
assert _is_quiet_hour(12, (23, 8)) is False
def test_overnight_range_at_end_boundary(self):
assert _is_quiet_hour(8, (23, 8)) is False
def test_daytime_range_inside(self):
assert _is_quiet_hour(12, (9, 17)) is True
def test_daytime_range_at_start(self):
assert _is_quiet_hour(9, (9, 17)) is True
def test_daytime_range_at_end(self):
assert _is_quiet_hour(17, (9, 17)) is False
def test_daytime_range_outside(self):
assert _is_quiet_hour(20, (9, 17)) is False
# ---------------------------------------------------------------------------
# _should_run (cooldowns)
# ---------------------------------------------------------------------------
class TestShouldRun:
"""Test cooldown logic for checks."""
def test_no_previous_run(self):
assert _should_run("email", {}, datetime.now(timezone.utc)) is True
def test_within_cooldown(self):
now = datetime.now(timezone.utc)
checks = {"email": (now - timedelta(minutes=10)).isoformat()}
assert _should_run("email", checks, now) is False
def test_past_cooldown(self):
now = datetime.now(timezone.utc)
checks = {"email": (now - timedelta(minutes=35)).isoformat()}
assert _should_run("email", checks, now) is True
def test_zero_cooldown_always_runs(self):
now = datetime.now(timezone.utc)
checks = {"calendar": now.isoformat()}
assert _should_run("calendar", checks, now) is True
def test_corrupt_timestamp(self):
now = datetime.now(timezone.utc)
checks = {"email": "not-a-date"}
assert _should_run("email", checks, now) is True
def test_custom_cooldowns(self):
"""Accepts custom cooldowns dict."""
now = datetime.now(timezone.utc)
checks = {"email": (now - timedelta(minutes=5)).isoformat()}
# Default 1800s (30min) — should NOT run
assert _should_run("email", checks, now) is False
# Custom 60s — should run (5 min > 60s)
assert _should_run("email", checks, now, cooldowns={"email": 60}) is True
# ---------------------------------------------------------------------------
# _check_email
# ---------------------------------------------------------------------------
class TestCheckEmail:
"""Test email check via tools/email_check.py with JSON parsing."""
def test_no_script(self, tmp_env):
assert _check_email({}) is None
def test_json_with_unread(self, tmp_env):
script = tmp_env["tools"] / "email_check.py"
script.write_text("pass")
output = json.dumps({
"ok": True, "unread_count": 2,
"emails": [
{"subject": "Meeting azi", "from": "boss@work.com"},
{"subject": "Factura", "from": "billing@x.com"},
]
})
mock_result = MagicMock(returncode=0, stdout=output)
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
result = _check_email({})
assert result == "Email: 2 necitite (Meeting azi, Factura)"
def test_json_zero_unread(self, tmp_env):
script = tmp_env["tools"] / "email_check.py"
script.write_text("pass")
output = json.dumps({"ok": True, "unread_count": 0, "emails": []})
mock_result = MagicMock(returncode=0, stdout=output)
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
assert _check_email({}) is None
def test_json_not_ok(self, tmp_env):
script = tmp_env["tools"] / "email_check.py"
script.write_text("pass")
output = json.dumps({"ok": False, "error": "auth failed"})
mock_result = MagicMock(returncode=0, stdout=output)
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
assert _check_email({}) is None
def test_plaintext_fallback(self, tmp_env):
script = tmp_env["tools"] / "email_check.py"
script.write_text("pass")
mock_result = MagicMock(returncode=0, stdout="3 new messages\n")
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
assert _check_email({}) == "Email: 3 new messages"
def test_plaintext_zero(self, tmp_env):
script = tmp_env["tools"] / "email_check.py"
script.write_text("pass")
mock_result = MagicMock(returncode=0, stdout="0\n")
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
assert _check_email({}) is None
def test_empty_output(self, tmp_env):
script = tmp_env["tools"] / "email_check.py"
script.write_text("pass")
mock_result = MagicMock(returncode=0, stdout="\n")
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
assert _check_email({}) is None
def test_nonzero_returncode(self, tmp_env):
script = tmp_env["tools"] / "email_check.py"
script.write_text("pass")
mock_result = MagicMock(returncode=1, stdout="error")
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
assert _check_email({}) is None
def test_subprocess_exception(self, tmp_env):
script = tmp_env["tools"] / "email_check.py"
script.write_text("pass")
with patch("src.heartbeat.subprocess.run", side_effect=TimeoutError):
assert _check_email({}) is None
# ---------------------------------------------------------------------------
# _check_calendar
# ---------------------------------------------------------------------------
class TestCheckCalendarSmart:
"""Test smart calendar: daily summary + next-event reminders with dedup."""
def test_no_script(self, tmp_env):
assert _check_calendar_smart({}, (23, 8)) is None
def test_first_run_sends_daily_summary(self, tmp_env):
"""First run after quiet hours should return full day summary."""
script = tmp_env["tools"] / "calendar_check.py"
script.write_text("pass")
output = json.dumps({
"today": [
{"summary": "Standup", "time": "09:00"},
{"summary": "Lunch", "time": "12:00"},
],
"tomorrow": []
})
mock_result = MagicMock(returncode=0, stdout=output)
state = {}
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
result = _check_calendar_smart(state, (23, 8))
assert "Program azi" in result
assert "09:00 — Standup" in result
assert "12:00 — Lunch" in result
assert state["calendar"]["daily_summary_date"] == date.today().isoformat()
def test_second_run_sends_next_event_only(self, tmp_env):
"""After daily summary, only remind for next event within 45 min."""
script = tmp_env["tools"] / "calendar_check.py"
script.write_text("pass")
output = json.dumps({
"upcoming": [
{"summary": "Meeting", "minutes_until": 30, "time": "10:00"},
]
})
mock_result = MagicMock(returncode=0, stdout=output)
state = {"calendar": {"daily_summary_date": date.today().isoformat(), "reminded_events": {}}}
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
result = _check_calendar_smart(state, (23, 8))
assert "in 30 min — Meeting (10:00)" in result
assert "Meeting@10:00" in state["calendar"]["reminded_events"]
def test_dedup_prevents_repeat_reminder(self, tmp_env):
"""Same event should not be reminded twice."""
script = tmp_env["tools"] / "calendar_check.py"
script.write_text("pass")
output = json.dumps({
"upcoming": [
{"summary": "Meeting", "minutes_until": 25, "time": "10:00"},
]
})
mock_result = MagicMock(returncode=0, stdout=output)
today_str = date.today().isoformat()
state = {"calendar": {
"daily_summary_date": today_str,
"reminded_events": {"Meeting@10:00": today_str}
}}
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
result = _check_calendar_smart(state, (23, 8))
assert result is None
def test_event_beyond_45_min_not_reminded(self, tmp_env):
"""Events more than 45 min away should not trigger a reminder."""
script = tmp_env["tools"] / "calendar_check.py"
script.write_text("pass")
output = json.dumps({
"upcoming": [
{"summary": "Later", "minutes_until": 90, "time": "14:00"},
]
})
mock_result = MagicMock(returncode=0, stdout=output)
state = {"calendar": {"daily_summary_date": date.today().isoformat(), "reminded_events": {}}}
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
result = _check_calendar_smart(state, (23, 8))
assert result is None
def test_no_events_today(self, tmp_env):
"""No events returns None and still marks summary as sent."""
script = tmp_env["tools"] / "calendar_check.py"
script.write_text("pass")
output = json.dumps({"today": [], "tomorrow": []})
mock_result = MagicMock(returncode=0, stdout=output)
state = {}
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
result = _check_calendar_smart(state, (23, 8))
assert result is None
assert state["calendar"]["daily_summary_date"] == date.today().isoformat()
def test_old_reminded_events_cleaned(self, tmp_env):
"""Reminded events from previous days are cleaned up."""
script = tmp_env["tools"] / "calendar_check.py"
script.write_text("pass")
output = json.dumps({"today": [{"summary": "X", "time": "09:00"}], "tomorrow": []})
mock_result = MagicMock(returncode=0, stdout=output)
state = {"calendar": {
"daily_summary_date": "2020-01-01",
"reminded_events": {"OldEvent@09:00": "2020-01-01"}
}}
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
_check_calendar_smart(state, (23, 8))
assert "OldEvent@09:00" not in state["calendar"]["reminded_events"]
def test_subprocess_exception(self, tmp_env):
script = tmp_env["tools"] / "calendar_check.py"
script.write_text("pass")
with patch("src.heartbeat.subprocess.run", side_effect=OSError("fail")):
assert _check_calendar_smart({}, (23, 8)) is None
# ---------------------------------------------------------------------------
# _check_kb_index
# ---------------------------------------------------------------------------
class TestCheckKbIndex:
"""Test KB index freshness check with auto-reindex."""
def test_missing_index(self, tmp_env):
with patch("src.heartbeat._run_reindex") as mock_reindex:
result = _check_kb_index()
assert result == "KB: index regenerat"
mock_reindex.assert_called_once()
def test_up_to_date(self, tmp_env):
kb_dir = tmp_env["root"] / "memory" / "kb"
kb_dir.mkdir(parents=True)
md_file = kb_dir / "notes.md"
md_file.write_text("old notes")
time.sleep(0.05)
index = kb_dir / "index.json"
index.write_text("{}")
assert _check_kb_index() is None
def test_needs_reindex(self, tmp_env):
kb_dir = tmp_env["root"] / "memory" / "kb"
kb_dir.mkdir(parents=True)
index = kb_dir / "index.json"
index.write_text("{}")
time.sleep(0.05)
md1 = kb_dir / "a.md"
md1.write_text("new")
md2 = kb_dir / "b.md"
md2.write_text("also new")
with patch("src.heartbeat._run_reindex") as mock_reindex:
result = _check_kb_index()
assert result == "KB: 2 fișiere reindexate"
mock_reindex.assert_called_once()
# ---------------------------------------------------------------------------
# _check_git
# ---------------------------------------------------------------------------
class TestCheckGit:
"""Test git status check."""
def test_clean(self, tmp_env):
mock_result = MagicMock(returncode=0, stdout="\n")
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
assert _check_git() is None
def test_dirty(self, tmp_env):
mock_result = MagicMock(
returncode=0,
stdout=" M file1.py\n?? file2.py\n M file3.py\n",
)
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
assert _check_git() == "Git: 3 uncommitted"
def test_subprocess_exception(self, tmp_env):
with patch("src.heartbeat.subprocess.run", side_effect=OSError):
assert _check_git() is None
# ---------------------------------------------------------------------------
# _get_extra_instructions
# ---------------------------------------------------------------------------
class TestGetExtraInstructions:
"""Test HEARTBEAT.md parsing for extra instructions."""
def test_no_file(self, tmp_env):
"""Returns None when HEARTBEAT.md doesn't exist."""
assert _get_extra_instructions() is None
def test_only_boilerplate(self, tmp_env):
"""Returns None when HEARTBEAT.md has only standard rules."""
hb = tmp_env["personality"] / "HEARTBEAT.md"
hb.write_text(
"# HEARTBEAT.md\n\n"
"## Reguli\n\n"
"- **Noapte (23:00-08:00):** Doar HEARTBEAT_OK, nu deranja\n"
"- **Nu spama:** Dacă nu e nimic, HEARTBEAT_OK\n"
)
assert _get_extra_instructions() is None
def test_with_extra(self, tmp_env):
"""Returns extra instructions when present."""
hb = tmp_env["personality"] / "HEARTBEAT.md"
hb.write_text(
"# HEARTBEAT.md\n\n"
"## Reguli\n\n"
"- **Noapte (23:00-08:00):** Doar HEARTBEAT_OK, nu deranja\n\n"
"## Extra\n\n"
"- Verifică dacă backup-ul s-a făcut\n"
"- Raportează uptime-ul serverului\n"
)
result = _get_extra_instructions()
assert result is not None
assert "backup" in result
assert "uptime" in result
# ---------------------------------------------------------------------------
# _load_state / _save_state
# ---------------------------------------------------------------------------
class TestState:
"""Test state persistence."""
def test_load_missing_file(self, tmp_env):
state = _load_state()
assert state == {"last_run": None, "checks": {}}
def test_round_trip(self, tmp_env):
original = {"last_run": "2025-01-01T00:00:00", "checks": {"email": "2025-01-01T00:00:00"}}
_save_state(original)
loaded = _load_state()
assert loaded == original
def test_load_corrupt_json(self, tmp_env):
tmp_env["state_file"].write_text("not valid json {{{")
state = _load_state()
assert state == {"last_run": None, "checks": {}}
def test_save_creates_parent_dir(self, tmp_path, monkeypatch):
state_file = tmp_path / "deep" / "nested" / "state.json"
monkeypatch.setattr("src.heartbeat.STATE_FILE", state_file)
_save_state({"last_run": None, "checks": {}})
assert state_file.exists()
# ---------------------------------------------------------------------------
# run_heartbeat (integration)
# ---------------------------------------------------------------------------
class TestRunHeartbeat:
"""Test the top-level run_heartbeat orchestrator."""
def test_all_ok(self, tmp_env):
with patch("src.heartbeat._check_email", return_value=None), \
patch("src.heartbeat._check_calendar_smart", return_value=None), \
patch("src.heartbeat._check_kb_index", return_value=None), \
patch("src.heartbeat._check_git", return_value=None), \
patch("src.heartbeat._run_claude_extra", return_value=None):
result = run_heartbeat()
assert result == "HEARTBEAT_OK"
def test_with_results(self, tmp_env):
with patch("src.heartbeat._check_email", return_value="Email: 2 necitite (X, Y)"), \
patch("src.heartbeat._check_calendar_smart", return_value=None), \
patch("src.heartbeat._check_kb_index", return_value="KB: 1 fișiere reindexate"), \
patch("src.heartbeat._check_git", return_value=None), \
patch("src.heartbeat._is_quiet_hour", return_value=False), \
patch("src.heartbeat._run_claude_extra", return_value=None):
result = run_heartbeat()
assert result == "Email: 2 necitite (X, Y) | KB: 1 fișiere reindexate"
def test_quiet_hours_suppresses_all(self, tmp_env):
"""During quiet hours, calendar is also suppressed (no longer critical)."""
with patch("src.heartbeat._check_email", return_value="Email: 5 necitite (A)"), \
patch("src.heartbeat._check_calendar_smart", return_value="Calendar stuff") as mock_cal, \
patch("src.heartbeat._check_kb_index", return_value=None), \
patch("src.heartbeat._check_git", return_value="Git: 2 uncommitted"), \
patch("src.heartbeat._is_quiet_hour", return_value=True):
result = run_heartbeat()
assert result == "HEARTBEAT_OK"
mock_cal.assert_not_called()
def test_config_disables_check(self, tmp_env):
"""Checks can be disabled via config."""
config = {"heartbeat": {"checks": {"git": False}}}
with patch("src.heartbeat._check_email", return_value=None), \
patch("src.heartbeat._check_calendar_smart", return_value=None), \
patch("src.heartbeat._check_kb_index", return_value=None), \
patch("src.heartbeat._check_git", return_value="Git: 5 uncommitted") as mock_git, \
patch("src.heartbeat._run_claude_extra", return_value=None):
result = run_heartbeat(config)
mock_git.assert_not_called()
assert result == "HEARTBEAT_OK"
def test_config_custom_quiet_hours(self, tmp_env):
"""Quiet hours can be overridden via config."""
config = {"heartbeat": {"quiet_hours": [0, 1]}} # only 0-1 is quiet
with patch("src.heartbeat._check_email", return_value=None), \
patch("src.heartbeat._check_calendar_smart", return_value=None), \
patch("src.heartbeat._check_kb_index", return_value=None), \
patch("src.heartbeat._check_git", return_value="Git: 3 uncommitted"), \
patch("src.heartbeat._is_quiet_hour", return_value=False), \
patch("src.heartbeat._run_claude_extra", return_value=None):
result = run_heartbeat(config)
assert "Git: 3 uncommitted" in result
def test_saves_state_after_run(self, tmp_env):
with patch("src.heartbeat._check_email", return_value=None), \
patch("src.heartbeat._check_calendar_smart", return_value=None), \
patch("src.heartbeat._check_kb_index", return_value=None), \
patch("src.heartbeat._check_git", return_value=None), \
patch("src.heartbeat._run_claude_extra", return_value=None):
run_heartbeat()
state = json.loads(tmp_env["state_file"].read_text())
assert "last_run" in state
assert state["last_run"] is not None
def test_saves_check_timestamps(self, tmp_env):
with patch("src.heartbeat._check_email", return_value=None), \
patch("src.heartbeat._check_calendar_smart", return_value=None), \
patch("src.heartbeat._check_kb_index", return_value=None), \
patch("src.heartbeat._check_git", return_value=None), \
patch("src.heartbeat._run_claude_extra", return_value=None):
run_heartbeat()
state = json.loads(tmp_env["state_file"].read_text())
assert "checks" in state
assert "calendar" in state["checks"]
Reference in New Issue View Git Blame Copy Permalink