mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-10 19:17:02 +01:00
Krzysztof kuhy Rudnicki
72482629d9
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
84 lines
3.0 KiB
Python
84 lines
3.0 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
|
"""/healthz must go red when the bot cannot hear, not just when it is down."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
from types import SimpleNamespace
|
|
from typing import TYPE_CHECKING
|
|
|
|
import httpx
|
|
from fastapi import Response
|
|
|
|
from signal_bot import app as app_module
|
|
from signal_bot.app import healthz
|
|
from signal_bot.liveness import Liveness
|
|
from signal_bot.versioning import POPE, __version__
|
|
|
|
if TYPE_CHECKING:
|
|
from signal_bot.config import Config
|
|
|
|
DEAD_BODY = {"error": "Failed to get linked devices: Closed unexpectedly"}
|
|
|
|
|
|
def _gate(config: Config, response: httpx.Response) -> Liveness:
|
|
"""A liveness gate whose probe always gets this answer."""
|
|
client = httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response))
|
|
return Liveness(config, client)
|
|
|
|
|
|
def _install(liveness: Liveness) -> None:
|
|
"""Point the app at a connected starter carrying this gate."""
|
|
runtime = SimpleNamespace(bot=SimpleNamespace(liveness=liveness))
|
|
app_module.app.state.starter = SimpleNamespace(runtime=runtime, reason="")
|
|
|
|
|
|
async def test_a_bot_that_has_not_connected_yet_says_connecting() -> None:
|
|
"""Booting is not deafness, and must not be reported as it."""
|
|
app_module.app.state.starter = SimpleNamespace(
|
|
runtime=None, reason="connecting to http://localhost:9922, attempt 3: ConnectError"
|
|
)
|
|
response = Response()
|
|
body = await healthz(response)
|
|
assert response.status_code == httpx.codes.SERVICE_UNAVAILABLE
|
|
assert body["status"] == "connecting"
|
|
assert "attempt 3" in body["reason"]
|
|
|
|
|
|
async def test_a_healthy_bot_answers_ok(config: Config) -> None:
|
|
_install(_gate(config, httpx.Response(200, json=[])))
|
|
response = Response()
|
|
body = await healthz(response)
|
|
assert body["status"] == "ok"
|
|
assert response.status_code == httpx.codes.OK
|
|
|
|
|
|
async def test_a_deaf_bot_answers_503_and_says_why(config: Config) -> None:
|
|
"""The failure this exists for: up, listening, and hearing nothing."""
|
|
live = _gate(replace(config, silence_seconds=0), httpx.Response(400, json=DEAD_BODY))
|
|
for _ in range(3):
|
|
await live.check()
|
|
_install(live)
|
|
|
|
response = Response()
|
|
body = await healthz(response)
|
|
assert response.status_code == httpx.codes.SERVICE_UNAVAILABLE
|
|
assert body["status"] == "deaf"
|
|
assert "Closed unexpectedly" in body["reason"]
|
|
|
|
|
|
async def test_healthz_names_the_build_either_way(config: Config) -> None:
|
|
"""Without this you cannot ask a running bot which version it is."""
|
|
_install(_gate(config, httpx.Response(200, json=[])))
|
|
healthy = await healthz(Response())
|
|
assert healthy["version"] == __version__
|
|
assert healthy["pope"] == POPE.name
|
|
|
|
deaf_gate = _gate(replace(config, silence_seconds=0), httpx.Response(400, json=DEAD_BODY))
|
|
for _ in range(3):
|
|
await deaf_gate.check()
|
|
_install(deaf_gate)
|
|
deaf = await healthz(Response())
|
|
assert deaf["version"] == __version__
|
|
assert deaf["pope"] == POPE.name
|