romfast/echo-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(heartbeat): smart calendar with daily summary and dedup reminders

Browse Source 
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>
This commit is contained in:
MoltBot Service
2026-02-15 11:09:59 +00:00
parent 9c1f9f94e7
commit f8ff971627
3 changed files with 185 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
tests/test_heartbeat.py
View File

@@ -2,14 +2,16 @@
import json
import time
from datetime import datetime, timedelta, timezone
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from src.heartbeat import (
_check_calendar,
_calendar_daily_summary,
_calendar_next_reminder,
_check_calendar_smart,
_check_email,
_check_git,
_check_kb_index,
@@ -207,52 +209,113 @@ class TestCheckEmail:
# ---------------------------------------------------------------------------
class TestCheckCalendar:
"""Test calendar check via tools/calendar_check.py with JSON parsing."""
class TestCheckCalendarSmart:
"""Test smart calendar: daily summary + next-event reminders with dedup."""
def test_no_script(self, tmp_env):
assert _check_calendar({}) is None
assert _check_calendar_smart({}, (23, 8)) is None
def test_json_with_upcoming(self, tmp_env):
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": "NLP Session", "minutes_until": 45, "time": "15:00"},
{"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({})
assert result == "Calendar: in 45 min — NLP Session (15:00)"
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_json_no_upcoming(self, tmp_env):
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": []})
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):
assert _check_calendar({}) is None
result = _check_calendar_smart(state, (23, 8))
assert result is None
def test_plaintext_fallback(self, tmp_env):
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")
mock_result = MagicMock(returncode=0, stdout="Meeting at 3pm\n")
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):
assert _check_calendar({}) == "Calendar: Meeting at 3pm"
result = _check_calendar_smart(state, (23, 8))
assert result is None
def test_empty_output(self, tmp_env):
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")
mock_result = MagicMock(returncode=0, stdout="\n")
output = json.dumps({"today": [], "tomorrow": []})
mock_result = MagicMock(returncode=0, stdout=output)
state = {}
with patch("src.heartbeat.subprocess.run", return_value=mock_result):
assert _check_calendar({}) is None
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({}) is None
assert _check_calendar_smart({}, (23, 8)) is None
# ---------------------------------------------------------------------------
@@ -401,7 +464,7 @@ class TestRunHeartbeat:
def test_all_ok(self, tmp_env):
with patch("src.heartbeat._check_email", return_value=None), \
patch("src.heartbeat._check_calendar", 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):
@@ -410,7 +473,7 @@ class TestRunHeartbeat:
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", return_value=None), \
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), \
@@ -418,29 +481,22 @@ class TestRunHeartbeat:
result = run_heartbeat()
assert result == "Email: 2 necitite (X, Y) | KB: 1 fișiere reindexate"
def test_quiet_hours_suppresses_normal(self, tmp_env):
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", return_value=None), \
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"
def test_quiet_hours_allows_critical_calendar(self, tmp_env):
with patch("src.heartbeat._check_email", return_value="Email: 5 necitite (A)"), \
patch("src.heartbeat._check_calendar", return_value="Calendar: in 30 min — Meeting (15:00)"), \
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 == "Calendar: in 30 min — Meeting (15:00)"
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", 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):
@@ -452,7 +508,7 @@ class TestRunHeartbeat:
"""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", 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), \
@@ -462,7 +518,7 @@ class TestRunHeartbeat:
def test_saves_state_after_run(self, tmp_env):
with patch("src.heartbeat._check_email", return_value=None), \
patch("src.heartbeat._check_calendar", 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):
@@ -473,7 +529,7 @@ class TestRunHeartbeat:
def test_saves_check_timestamps(self, tmp_env):
with patch("src.heartbeat._check_email", return_value=None), \
patch("src.heartbeat._check_calendar", 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):