a4ecb51401
Some checks failed
secrets-guard / encrypted (push) Successful in 24s
shell-lint / shellcheck (push) Successful in 26s
Deploy frontend to LAN dev server / build (push) Successful in 2m11s
Build + push frontend image (Forgejo registry) / build-push (push) Successful in 2m20s
Build + push backend image (Forgejo registry) / build-push (push) Successful in 2m28s
Deploy backend to LAN dev server / build (push) Successful in 2m45s
PR build (required check) / changes (pull_request) Successful in 16s
secrets-guard / encrypted (pull_request) Successful in 16s
shell-lint / shellcheck (pull_request) Successful in 15s
PR build (required check) / build-frontend (pull_request) Has been skipped
PR build (required check) / validate-observability (pull_request) Has been skipped
Deploy frontend to LAN dev server / deploy (push) Successful in 39s
PR build (required check) / build-backend (pull_request) Successful in 1m21s
PR build (required check) / gate (pull_request) Successful in 3s
Deploy backend to LAN dev server / deploy (push) Failing after 2m31s
91 lines
3.4 KiB
Python
91 lines
3.4 KiB
Python
"""Process-level liveness heartbeat for web/worker (app.py _heartbeat_loop /
|
|
_start_heartbeat) — the fix for ProdWebContainerSilent / ProdWorkerContainerSilent
|
|
false-firing on container-stdout silence that isn't a real liveness signal for
|
|
either role."""
|
|
import json
|
|
import time
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from core import audit
|
|
from web import app as appmod
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_heartbeat_thread():
|
|
# _HEARTBEAT_THREAD is a module-level singleton guard (mirrors notify.py's own
|
|
# _STOP/_THREAD pattern) -- reset it around every test so one test starting the
|
|
# thread doesn't leave _start_heartbeat() a no-op for the next.
|
|
appmod._HEARTBEAT_STOP.clear()
|
|
appmod._HEARTBEAT_THREAD = None
|
|
yield
|
|
appmod._HEARTBEAT_STOP.set()
|
|
appmod._HEARTBEAT_THREAD = None
|
|
|
|
|
|
@pytest.mark.parametrize("role", ["web", "worker", "all"])
|
|
def test_heartbeat_loop_tags_the_beat_with_role(monkeypatch, role):
|
|
monkeypatch.setattr(appmod, "ROLE", role)
|
|
calls = []
|
|
monkeypatch.setattr(audit, "log_heartbeat", lambda daemon, interval_s: calls.append((daemon, interval_s)))
|
|
monkeypatch.setattr(appmod.metrics, "record_heartbeat", lambda daemon, interval_s: None)
|
|
|
|
# Setting the stop event first means .wait() returns True immediately, so the
|
|
# loop beats exactly once (the pre-loop beat) and exits without blocking.
|
|
appmod._HEARTBEAT_STOP.set()
|
|
appmod._heartbeat_loop()
|
|
|
|
assert calls == [(role, appmod.HEARTBEAT_INTERVAL)]
|
|
|
|
|
|
def test_start_heartbeat_respects_the_disable_flag(monkeypatch):
|
|
monkeypatch.setenv("THERMOGRAPH_ENABLE_HEARTBEAT", "0")
|
|
started = []
|
|
monkeypatch.setattr(appmod.threading, "Thread", lambda **kw: started.append(kw))
|
|
|
|
appmod._start_heartbeat()
|
|
|
|
assert appmod._HEARTBEAT_THREAD is None
|
|
assert started == []
|
|
|
|
|
|
def test_start_heartbeat_is_idempotent(monkeypatch):
|
|
monkeypatch.delenv("THERMOGRAPH_ENABLE_HEARTBEAT", raising=False)
|
|
starts = []
|
|
|
|
class _FakeThread:
|
|
def __init__(self, **kw):
|
|
starts.append(kw)
|
|
|
|
def start(self):
|
|
pass
|
|
|
|
monkeypatch.setattr(appmod.threading, "Thread", _FakeThread)
|
|
|
|
appmod._start_heartbeat()
|
|
appmod._start_heartbeat()
|
|
|
|
# A second call is a no-op once a thread handle already exists -- mirrors
|
|
# notify.start()'s own idempotency, so a double-entry into the lifespan
|
|
# (or a stray manual call) can't spawn a second beat loop.
|
|
assert len(starts) == 1
|
|
|
|
|
|
def test_app_boot_emits_a_heartbeat_within_one_interval(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(appmod, "ROLE", "web")
|
|
monkeypatch.setattr(appmod, "HEARTBEAT_INTERVAL", 0.01)
|
|
monkeypatch.setattr(audit, "HEARTBEAT_DIR", str(tmp_path))
|
|
monkeypatch.delenv("THERMOGRAPH_ENABLE_HEARTBEAT", raising=False)
|
|
|
|
with TestClient(appmod.app):
|
|
# The pre-loop beat fires synchronously on thread start; a short sleep is
|
|
# just slack for the thread to actually get scheduled. The loop itself
|
|
# keeps running on HEARTBEAT_INTERVAL until lifespan shutdown sets the
|
|
# stop event, so there's nothing to join here.
|
|
time.sleep(0.2)
|
|
|
|
files = list(tmp_path.glob("heartbeat-*.jsonl"))
|
|
assert files, "expected a heartbeat-*.jsonl file to be written during app lifespan"
|
|
lines = [json.loads(line) for line in files[0].read_text().splitlines() if line]
|
|
assert any(rec["daemon"] == "web" and rec["tag"] == "heartbeat" for rec in lines)
|