"""Tests for src/heartbeat.py — Periodic health checks.""" import json import time from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import MagicMock, patch import pytest from src.heartbeat import ( _check_calendar, _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 TestCheckCalendar: """Test calendar check via tools/calendar_check.py with JSON parsing.""" def test_no_script(self, tmp_env): assert _check_calendar({}) is None def test_json_with_upcoming(self, tmp_env): script = tmp_env["tools"] / "calendar_check.py" script.write_text("pass") output = json.dumps({ "upcoming": [ {"summary": "NLP Session", "minutes_until": 45, "time": "15:00"}, ] }) mock_result = MagicMock(returncode=0, stdout=output) with patch("src.heartbeat.subprocess.run", return_value=mock_result): result = _check_calendar({}) assert result == "Calendar: in 45 min — NLP Session (15:00)" def test_json_no_upcoming(self, tmp_env): script = tmp_env["tools"] / "calendar_check.py" script.write_text("pass") output = json.dumps({"upcoming": []}) mock_result = MagicMock(returncode=0, stdout=output) with patch("src.heartbeat.subprocess.run", return_value=mock_result): assert _check_calendar({}) is None def test_plaintext_fallback(self, tmp_env): script = tmp_env["tools"] / "calendar_check.py" script.write_text("pass") mock_result = MagicMock(returncode=0, stdout="Meeting at 3pm\n") with patch("src.heartbeat.subprocess.run", return_value=mock_result): assert _check_calendar({}) == "Calendar: Meeting at 3pm" def test_empty_output(self, tmp_env): script = tmp_env["tools"] / "calendar_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_calendar({}) is None 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 # --------------------------------------------------------------------------- # _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", 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", 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_normal(self, tmp_env): with patch("src.heartbeat._check_email", return_value="Email: 5 necitite (A)"), \ patch("src.heartbeat._check_calendar", return_value=None), \ 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)" 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_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", 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", 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", 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"]