"""The heartbeat that brings reminders and delivered files in — with fakes for the network. Covers the report_status POST, the outbox list, the download rules (.part first, never overwrite, clean up on failure) and the loop's stop/disable behaviour.""" import sys import threading from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import pytest from bolt_assistant import config, heartbeat class FakeResponse: def __init__(self, payload=None, data=b"", status_code=200): self._payload = payload self._data = data self.status_code = status_code def json(self): return self._payload def iter_content(self, chunk_size=65536): yield self._data @pytest.fixture(autouse=True) def _quiet(monkeypatch): monkeypatch.setattr(heartbeat, "log", lambda msg: None) def _files_get(data=b"hello", files=None, status_code=200): def get(url, params=None, headers=None, timeout=None, stream=None): if url.endswith("/desk/files"): return FakeResponse(payload={"files": files or []}) return FakeResponse(data=data, status_code=status_code) return get # ── one beat ───────────────────────────────────────────────────────────────── def test_heartbeat_once_reports_in_and_returns_ok(tmp_path): seen = {} def post(url, json=None, headers=None, timeout=None): seen["url"], seen["json"], seen["headers"] = url, json, headers return FakeResponse(payload={"ok": True}) result = heartbeat.heartbeat_once(tmp_path, post=post, get=_files_get()) assert result == {"ok": True, "announcement": "", "saved": []} assert seen["url"].endswith("/desk/report_status") assert seen["json"] == {"session_id": config.SESSION_ID, "download_dir": str(tmp_path)} assert seen["headers"] == {"X-Desk-Api-Key": config.API_KEY} def test_heartbeat_once_carries_the_announcement(tmp_path): def post(url, json=None, headers=None, timeout=None): return FakeResponse(payload={"reply": " You have a reminder. "}) result = heartbeat.heartbeat_once(tmp_path, post=post, get=_files_get()) assert result["announcement"] == "You have a reminder." def test_heartbeat_once_downloads_the_outbox(tmp_path): seen = {} def get(url, params=None, headers=None, timeout=None, stream=None): if url.endswith("/desk/files"): seen["list"] = params return FakeResponse(payload={"files": [{"id": "abc", "name": "note.txt", "size": 5}]}) seen["download"] = params return FakeResponse(data=b"hello") def post(url, json=None, headers=None, timeout=None): return FakeResponse(payload={"ok": True}) result = heartbeat.heartbeat_once(tmp_path, post=post, get=get) assert seen["list"] == seen["download"] == {"session_id": config.SESSION_ID} assert len(result["saved"]) == 1 target = result["saved"][0] assert target.parent == tmp_path and target.name == "note.txt" assert target.read_bytes() == b"hello" assert not list(tmp_path.glob("*.part")) def test_heartbeat_once_network_failure_is_not_fatal(tmp_path): def post(url, json=None, headers=None, timeout=None): raise OSError("no network") result = heartbeat.heartbeat_once(tmp_path, post=post, get=_files_get()) assert result == {"ok": False, "announcement": "", "saved": []} def test_heartbeat_once_refused_does_not_touch_the_outbox(tmp_path): got = [] def get(url, params=None, headers=None, timeout=None, stream=None): got.append(url) return FakeResponse(payload={"files": []}) def post(url, json=None, headers=None, timeout=None): return FakeResponse(payload={"error": "unauthorized"}, status_code=401) result = heartbeat.heartbeat_once(tmp_path, post=post, get=get) assert result == {"ok": False, "announcement": "", "saved": []} assert got == [] # ── download rules ─────────────────────────────────────────────────────────── def test_download_never_overwrites(tmp_path): (tmp_path / "note.txt").write_text("old") target = heartbeat._download("abc", "note.txt", tmp_path, get=_files_get(data=b"new")) assert target.name == "note (1).txt" assert (tmp_path / "note.txt").read_text() == "old" assert target.read_bytes() == b"new" def test_download_refused(tmp_path): assert heartbeat._download("abc", "note.txt", tmp_path, get=_files_get(status_code=500)) is None assert list(tmp_path.iterdir()) == [] def test_download_cleans_up_the_part_file_on_failure(tmp_path): class Exploding(FakeResponse): def iter_content(self, chunk_size=65536): yield b"part of" raise OSError("connection reset") assert heartbeat._download("abc", "note.txt", tmp_path, get=lambda **kw: Exploding()) is None assert list(tmp_path.iterdir()) == [] # no .part left behind def test_safe_filename_strips_paths_and_control_characters(): assert heartbeat._safe_filename("a/b\\c:d.txt") == "a_b_c_d.txt" assert heartbeat._safe_filename("..") == "download" assert heartbeat._safe_filename("") == "download" assert heartbeat._safe_filename(" fine name.txt ") == "fine name.txt" # ── health probe and the loop ──────────────────────────────────────────────── def test_fetch_server_health_warns_loudly_on_a_refused_key(monkeypatch): lines = [] monkeypatch.setattr(heartbeat, "log", lines.append) def get(url, headers=None, timeout=None): return FakeResponse(payload={"key_ok": False}) heartbeat.fetch_server_health(get=get) assert any("DESK_API_KEY" in line for line in lines) def test_fetch_server_health_logs_a_healthy_server(monkeypatch): lines = [] monkeypatch.setattr(heartbeat, "log", lines.append) def get(url, headers=None, timeout=None): return FakeResponse(payload={"key_ok": True, "version": "1"}) heartbeat.fetch_server_health(get=get) assert lines and lines[0].startswith("server:") def test_heartbeat_loop_reports_each_beat(monkeypatch): monkeypatch.setattr(config, "HEARTBEAT_SECONDS", 60) stop = threading.Event() results = [] monkeypatch.setattr(heartbeat, "heartbeat_once", lambda: {"ok": True, "announcement": "hi", "saved": []}) def on_result(result): results.append(result) stop.set() heartbeat.heartbeat_loop(stop, on_result) assert results == [{"ok": True, "announcement": "hi", "saved": []}] def test_heartbeat_loop_disabled_when_heartbeat_seconds_is_zero(monkeypatch): monkeypatch.setattr(config, "HEARTBEAT_SECONDS", 0) called = [] monkeypatch.setattr(heartbeat, "heartbeat_once", lambda: called.append(1)) heartbeat.heartbeat_loop(threading.Event(), None) assert called == []