"""The Test button: it exists to tell a working channel from a broken one.""" from typing import Any import pytest from fastapi.testclient import TestClient from app.core.config import settings from app.flow.alerts import AlertsConfig, Channel PREFIX = f"{settings.API_V1_STR}/alerts" def _one_channel(client: TestClient) -> Any: alerts = client.app.state.flow_controller.alerts # type: ignore[attr-defined] alerts.config = AlertsConfig( channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})] ) return alerts def test_a_channel_that_cannot_deliver_is_not_a_success( client: TestClient, superuser_token_headers: dict[str, str], monkeypatch: pytest.MonkeyPatch, ) -> None: alerts = _one_channel(client) async def refuse(_config: dict[str, Any], _alert: Any) -> None: raise RuntimeError("nobody home") monkeypatch.setattr(alerts, "_send_ntfy", refuse) response = client.post(f"{PREFIX}/test/phone", headers=superuser_token_headers) assert response.status_code == 502 assert "nobody home" in response.json()["detail"] def test_a_channel_that_delivers_says_so( client: TestClient, superuser_token_headers: dict[str, str], monkeypatch: pytest.MonkeyPatch, ) -> None: alerts = _one_channel(client) sent = [] async def accept(_config: dict[str, Any], alert: Any) -> None: sent.append(alert) monkeypatch.setattr(alerts, "_send_ntfy", accept) response = client.post(f"{PREFIX}/test/phone", headers=superuser_token_headers) assert response.status_code == 200 assert len(sent) == 1 def test_an_unknown_channel_is_a_404( client: TestClient, superuser_token_headers: dict[str, str] ) -> None: _one_channel(client) response = client.post(f"{PREFIX}/test/nowhere", headers=superuser_token_headers) assert response.status_code == 404