34 lines
1.3 KiB
Python
34 lines
1.3 KiB
Python
"""THERMOGRAPH_ROLE: which processes are allowed to own the subscription
|
|
notifier (web/worker/all split), plus the /healthz liveness route."""
|
|
from fastapi.testclient import TestClient
|
|
|
|
from core import singleton
|
|
from web import app as appmod
|
|
|
|
|
|
def test_default_role_is_all_and_permits_the_notifier(monkeypatch):
|
|
monkeypatch.setattr(appmod, "ROLE", "all")
|
|
monkeypatch.setattr(singleton, "claim_leader", lambda: True)
|
|
assert appmod._should_run_notifier() is True
|
|
|
|
|
|
def test_web_role_never_runs_the_notifier_even_if_it_would_win_leader(monkeypatch):
|
|
monkeypatch.setattr(appmod, "ROLE", "web")
|
|
monkeypatch.setattr(singleton, "claim_leader", lambda: True)
|
|
assert appmod._should_run_notifier() is False
|
|
|
|
|
|
def test_worker_role_runs_the_notifier_only_if_it_wins_leader(monkeypatch):
|
|
monkeypatch.setattr(appmod, "ROLE", "worker")
|
|
monkeypatch.setattr(singleton, "claim_leader", lambda: False)
|
|
assert appmod._should_run_notifier() is False
|
|
monkeypatch.setattr(singleton, "claim_leader", lambda: True)
|
|
assert appmod._should_run_notifier() is True
|
|
|
|
|
|
def test_healthz_reports_status_and_configured_role(monkeypatch):
|
|
monkeypatch.setattr(appmod, "ROLE", "worker")
|
|
client = TestClient(appmod.app)
|
|
r = client.get("/healthz")
|
|
assert r.status_code == 200
|
|
assert r.json() == {"status": "ok", "role": "worker"}
|