Bolt-Assistant/tests/test_heartbeat.py

"""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 == []