"""Discord DM alerts: send_dm REST mechanics (mocked), the notify dispatch gate, the on/off toggle endpoint, and that linking opts in.""" import types import pytest from fastapi.testclient import TestClient from web import app as appmod from accounts import db from notifications import discord from notifications import discord_link as dl from notifications import notify V2 = "/thermograph/api/v2" PW = "supersecret123" @pytest.fixture(scope="module", autouse=True) def _tables(): db.Base.metadata.create_all(db.sync_engine) # --- send_dm REST mechanics -------------------------------------------------- class _Resp: def __init__(self, code, data=None): self.status_code = code; self._data = data or {} def json(self): return self._data def _mock_posts(monkeypatch, script): """Route discord.httpx.post by URL; `script` maps a URL substring -> _Resp.""" calls = [] def _post(url, json=None, headers=None, timeout=None): calls.append((url, json)) for frag, resp in script.items(): if frag in url: return resp return _Resp(404) monkeypatch.setattr(discord.httpx, "post", _post) return calls def test_send_dm_opens_channel_then_posts(monkeypatch): monkeypatch.setattr(discord, "BOT_TOKEN", "bot-abc") calls = _mock_posts(monkeypatch, { "/users/@me/channels": _Resp(200, {"id": "chan-1"}), "/channels/chan-1/messages": _Resp(204), }) ok = discord.send_dm("discord-9", "Tehran hit a Near Record high", "104F, 99th pct", "/day#x") assert ok is True assert calls[0][0].endswith("/users/@me/channels") and calls[0][1] == {"recipient_id": "discord-9"} assert "/channels/chan-1/messages" in calls[1][0] embed = calls[1][1]["embeds"][0] assert embed["title"].startswith("Tehran") and embed["url"].endswith("/day#x") def test_send_dm_disabled_or_no_recipient(monkeypatch): monkeypatch.setattr(discord, "BOT_TOKEN", "") assert discord.send_dm("d", "t", "b") is False monkeypatch.setattr(discord, "BOT_TOKEN", "bot") assert discord.send_dm("", "t", "b") is False def test_send_dm_gives_up_if_channel_open_fails(monkeypatch): monkeypatch.setattr(discord, "BOT_TOKEN", "bot-abc") calls = _mock_posts(monkeypatch, {"/users/@me/channels": _Resp(403)}) assert discord.send_dm("d", "t", "b") is False assert len(calls) == 1 # never tried to post a message def test_bot_post_retries_once_on_429(monkeypatch): monkeypatch.setattr(discord, "BOT_TOKEN", "bot-abc") monkeypatch.setattr(discord.time, "sleep", lambda s: None) # don't actually wait seq = iter([_Resp(429, {"retry_after": 0.01}), _Resp(200, {"id": "c"})]) monkeypatch.setattr(discord.httpx, "post", lambda *a, **k: next(seq)) resp = discord._bot_post("/users/@me/channels", {"recipient_id": "d"}) assert resp.status_code == 200 # the retry succeeded # --- notify dispatch gate ---------------------------------------------------- def _fake_session(user): return types.SimpleNamespace(get=lambda model, uid: user) def _sub(): return types.SimpleNamespace(user_id="u1", lat=47.6, lon=-122.3) def test_dispatch_sends_for_linked_opted_in_user(monkeypatch): monkeypatch.setattr(discord, "dm_enabled", lambda: True) sent = [] monkeypatch.setattr(discord, "send_dm", lambda *a, **k: sent.append(a) or True) user = types.SimpleNamespace(discord_id="d9", discord_dm=True) notify._dispatch_discord(_fake_session(user), _sub(), "T", "B", "2026-07-19") assert len(sent) == 1 and sent[0][0] == "d9" def test_dispatch_skips_unlinked_or_opted_out(monkeypatch): monkeypatch.setattr(discord, "dm_enabled", lambda: True) sent = [] monkeypatch.setattr(discord, "send_dm", lambda *a, **k: sent.append(a) or True) # linked but opted out notify._dispatch_discord(_fake_session(types.SimpleNamespace(discord_id="d", discord_dm=False)), _sub(), "T", "B", "2026-07-19") # not linked notify._dispatch_discord(_fake_session(types.SimpleNamespace(discord_id=None, discord_dm=True)), _sub(), "T", "B", "2026-07-19") assert sent == [] def test_dispatch_noop_when_bot_unconfigured(monkeypatch): monkeypatch.setattr(discord, "dm_enabled", lambda: False) called = [] monkeypatch.setattr(discord, "send_dm", lambda *a, **k: called.append(1)) notify._dispatch_discord(_fake_session(types.SimpleNamespace(discord_id="d", discord_dm=True)), _sub(), "T", "B", "2026-07-19") assert called == [] # never even looked the user up beyond the gate # --- the on/off toggle + link opt-in ----------------------------------------- def _login(client, email): r = client.post(f"{V2}/auth/register", json={"email": email, "password": PW}) assert r.status_code in (201, 400) assert client.post(f"{V2}/auth/login", data={"username": email, "password": PW}).status_code == 204 class _MockClient: def __init__(self, *a, **k): pass async def __aenter__(self): return self async def __aexit__(self, *a): return False async def post(self, url, **k): return _Resp(200, {"access_token": "tok"}) async def get(self, url, **k): return _Resp(200, {"id": "discord-777"}) def test_linking_opts_in_and_toggle_flips(monkeypatch): monkeypatch.setattr(dl, "CLIENT_ID", "app") monkeypatch.setattr(dl, "CLIENT_SECRET", "sec") monkeypatch.setattr(dl.httpx, "AsyncClient", _MockClient) c = TestClient(appmod.app) _login(c, "dm-toggle@example.com") uid = c.get(f"{V2}/users/me").json()["id"] # Link -> opted in by default. r = c.get(f"{V2}/discord/link/callback?code=x&state={dl._sign_state(uid)}", follow_redirects=False) assert r.status_code == 303 me = c.get(f"{V2}/users/me").json() assert me["discord_id"] == "discord-777" and me["discord_dm"] is True # Mute without unlinking. assert c.post(f"{V2}/discord/dm", json={"enabled": False}).status_code == 204 me = c.get(f"{V2}/users/me").json() assert me["discord_id"] == "discord-777" and me["discord_dm"] is False # Turn back on. assert c.post(f"{V2}/discord/dm", json={"enabled": True}).status_code == 204 assert c.get(f"{V2}/users/me").json()["discord_dm"] is True def test_dm_toggle_requires_auth(): c = TestClient(appmod.app) assert c.post(f"{V2}/discord/dm", json={"enabled": True}).status_code == 401