test_send_message_missing_platforms.py
1 """Tests for _send_mattermost, _send_matrix, _send_homeassistant, _send_dingtalk.""" 2 3 import asyncio 4 import os 5 from types import SimpleNamespace 6 from unittest.mock import AsyncMock, MagicMock, patch 7 8 from tools.send_message_tool import ( 9 _send_dingtalk, 10 _send_homeassistant, 11 _send_mattermost, 12 _send_matrix, 13 ) 14 15 16 # --------------------------------------------------------------------------- 17 # Helpers 18 # --------------------------------------------------------------------------- 19 20 21 def _make_aiohttp_resp(status, json_data=None, text_data=None): 22 """Build a minimal async-context-manager mock for an aiohttp response.""" 23 resp = AsyncMock() 24 resp.status = status 25 resp.json = AsyncMock(return_value=json_data or {}) 26 resp.text = AsyncMock(return_value=text_data or "") 27 return resp 28 29 30 def _make_aiohttp_session(resp): 31 """Wrap a response mock in a session mock that supports async-with for post/put.""" 32 request_ctx = MagicMock() 33 request_ctx.__aenter__ = AsyncMock(return_value=resp) 34 request_ctx.__aexit__ = AsyncMock(return_value=False) 35 36 session = MagicMock() 37 session.post = MagicMock(return_value=request_ctx) 38 session.put = MagicMock(return_value=request_ctx) 39 40 session_ctx = MagicMock() 41 session_ctx.__aenter__ = AsyncMock(return_value=session) 42 session_ctx.__aexit__ = AsyncMock(return_value=False) 43 return session_ctx, session 44 45 46 # --------------------------------------------------------------------------- 47 # _send_mattermost 48 # --------------------------------------------------------------------------- 49 50 51 class TestSendMattermost: 52 def test_success(self): 53 resp = _make_aiohttp_resp(201, json_data={"id": "post123"}) 54 session_ctx, session = _make_aiohttp_session(resp) 55 56 with patch("aiohttp.ClientSession", return_value=session_ctx), \ 57 patch.dict(os.environ, {"MATTERMOST_URL": "", "MATTERMOST_TOKEN": ""}, clear=False): 58 extra = {"url": "https://mm.example.com"} 59 result = asyncio.run(_send_mattermost("tok-abc", extra, "channel1", "hello")) 60 61 assert result == {"success": True, "platform": "mattermost", "chat_id": "channel1", "message_id": "post123"} 62 session.post.assert_called_once() 63 call_kwargs = session.post.call_args 64 assert call_kwargs[0][0] == "https://mm.example.com/api/v4/posts" 65 assert call_kwargs[1]["headers"]["Authorization"] == "Bearer tok-abc" 66 assert call_kwargs[1]["json"] == {"channel_id": "channel1", "message": "hello"} 67 68 def test_http_error(self): 69 resp = _make_aiohttp_resp(400, text_data="Bad Request") 70 session_ctx, _ = _make_aiohttp_session(resp) 71 72 with patch("aiohttp.ClientSession", return_value=session_ctx): 73 result = asyncio.run(_send_mattermost( 74 "tok", {"url": "https://mm.example.com"}, "ch", "hi" 75 )) 76 77 assert "error" in result 78 assert "400" in result["error"] 79 assert "Bad Request" in result["error"] 80 81 def test_missing_config(self): 82 with patch.dict(os.environ, {"MATTERMOST_URL": "", "MATTERMOST_TOKEN": ""}, clear=False): 83 result = asyncio.run(_send_mattermost("", {}, "ch", "hi")) 84 85 assert "error" in result 86 assert "MATTERMOST_URL" in result["error"] or "not configured" in result["error"] 87 88 def test_env_var_fallback(self): 89 resp = _make_aiohttp_resp(200, json_data={"id": "p99"}) 90 session_ctx, session = _make_aiohttp_session(resp) 91 92 with patch("aiohttp.ClientSession", return_value=session_ctx), \ 93 patch.dict(os.environ, {"MATTERMOST_URL": "https://mm.env.com", "MATTERMOST_TOKEN": "env-tok"}, clear=False): 94 result = asyncio.run(_send_mattermost("", {}, "ch", "hi")) 95 96 assert result["success"] is True 97 call_kwargs = session.post.call_args 98 assert "https://mm.env.com" in call_kwargs[0][0] 99 assert call_kwargs[1]["headers"]["Authorization"] == "Bearer env-tok" 100 101 102 # --------------------------------------------------------------------------- 103 # _send_matrix 104 # --------------------------------------------------------------------------- 105 106 107 class TestSendMatrix: 108 def test_success(self): 109 resp = _make_aiohttp_resp(200, json_data={"event_id": "$abc123"}) 110 session_ctx, session = _make_aiohttp_session(resp) 111 112 with patch("aiohttp.ClientSession", return_value=session_ctx), \ 113 patch.dict(os.environ, {"MATRIX_HOMESERVER": "", "MATRIX_ACCESS_TOKEN": ""}, clear=False): 114 extra = {"homeserver": "https://matrix.example.com"} 115 result = asyncio.run(_send_matrix("syt_tok", extra, "!room:example.com", "hello matrix")) 116 117 assert result == { 118 "success": True, 119 "platform": "matrix", 120 "chat_id": "!room:example.com", 121 "message_id": "$abc123", 122 } 123 session.put.assert_called_once() 124 call_kwargs = session.put.call_args 125 url = call_kwargs[0][0] 126 assert url.startswith("https://matrix.example.com/_matrix/client/v3/rooms/%21room%3Aexample.com/send/m.room.message/") 127 assert call_kwargs[1]["headers"]["Authorization"] == "Bearer syt_tok" 128 payload = call_kwargs[1]["json"] 129 assert payload["msgtype"] == "m.text" 130 assert payload["body"] == "hello matrix" 131 132 def test_http_error(self): 133 resp = _make_aiohttp_resp(403, text_data="Forbidden") 134 session_ctx, _ = _make_aiohttp_session(resp) 135 136 with patch("aiohttp.ClientSession", return_value=session_ctx): 137 result = asyncio.run(_send_matrix( 138 "tok", {"homeserver": "https://matrix.example.com"}, 139 "!room:example.com", "hi" 140 )) 141 142 assert "error" in result 143 assert "403" in result["error"] 144 assert "Forbidden" in result["error"] 145 146 def test_missing_config(self): 147 with patch.dict(os.environ, {"MATRIX_HOMESERVER": "", "MATRIX_ACCESS_TOKEN": ""}, clear=False): 148 result = asyncio.run(_send_matrix("", {}, "!room:example.com", "hi")) 149 150 assert "error" in result 151 assert "MATRIX_HOMESERVER" in result["error"] or "not configured" in result["error"] 152 153 def test_env_var_fallback(self): 154 resp = _make_aiohttp_resp(200, json_data={"event_id": "$ev1"}) 155 session_ctx, session = _make_aiohttp_session(resp) 156 157 with patch("aiohttp.ClientSession", return_value=session_ctx), \ 158 patch.dict(os.environ, { 159 "MATRIX_HOMESERVER": "https://matrix.env.com", 160 "MATRIX_ACCESS_TOKEN": "env-tok", 161 }, clear=False): 162 result = asyncio.run(_send_matrix("", {}, "!r:env.com", "hi")) 163 164 assert result["success"] is True 165 url = session.put.call_args[0][0] 166 assert "matrix.env.com" in url 167 168 def test_txn_id_is_unique_across_calls(self): 169 """Each call should generate a distinct transaction ID in the URL.""" 170 txn_ids = [] 171 172 def capture(*args, **kwargs): 173 url = args[0] 174 txn_ids.append(url.rsplit("/", 1)[-1]) 175 ctx = MagicMock() 176 ctx.__aenter__ = AsyncMock(return_value=_make_aiohttp_resp(200, json_data={"event_id": "$x"})) 177 ctx.__aexit__ = AsyncMock(return_value=False) 178 return ctx 179 180 session = MagicMock() 181 session.put = capture 182 session_ctx = MagicMock() 183 session_ctx.__aenter__ = AsyncMock(return_value=session) 184 session_ctx.__aexit__ = AsyncMock(return_value=False) 185 186 extra = {"homeserver": "https://matrix.example.com"} 187 188 import time 189 with patch("aiohttp.ClientSession", return_value=session_ctx): 190 asyncio.run(_send_matrix("tok", extra, "!r:example.com", "first")) 191 time.sleep(0.002) 192 with patch("aiohttp.ClientSession", return_value=session_ctx): 193 asyncio.run(_send_matrix("tok", extra, "!r:example.com", "second")) 194 195 assert len(txn_ids) == 2 196 assert txn_ids[0] != txn_ids[1] 197 198 199 # --------------------------------------------------------------------------- 200 # _send_homeassistant 201 # --------------------------------------------------------------------------- 202 203 204 class TestSendHomeAssistant: 205 def test_success(self): 206 resp = _make_aiohttp_resp(200) 207 session_ctx, session = _make_aiohttp_session(resp) 208 209 with patch("aiohttp.ClientSession", return_value=session_ctx), \ 210 patch.dict(os.environ, {"HASS_URL": "", "HASS_TOKEN": ""}, clear=False): 211 extra = {"url": "https://hass.example.com"} 212 result = asyncio.run(_send_homeassistant("hass-tok", extra, "mobile_app_phone", "alert!")) 213 214 assert result == {"success": True, "platform": "homeassistant", "chat_id": "mobile_app_phone"} 215 session.post.assert_called_once() 216 call_kwargs = session.post.call_args 217 assert call_kwargs[0][0] == "https://hass.example.com/api/services/notify/notify" 218 assert call_kwargs[1]["headers"]["Authorization"] == "Bearer hass-tok" 219 assert call_kwargs[1]["json"] == {"message": "alert!", "target": "mobile_app_phone"} 220 221 def test_http_error(self): 222 resp = _make_aiohttp_resp(401, text_data="Unauthorized") 223 session_ctx, _ = _make_aiohttp_session(resp) 224 225 with patch("aiohttp.ClientSession", return_value=session_ctx): 226 result = asyncio.run(_send_homeassistant( 227 "bad-tok", {"url": "https://hass.example.com"}, 228 "target", "msg" 229 )) 230 231 assert "error" in result 232 assert "401" in result["error"] 233 assert "Unauthorized" in result["error"] 234 235 def test_missing_config(self): 236 with patch.dict(os.environ, {"HASS_URL": "", "HASS_TOKEN": ""}, clear=False): 237 result = asyncio.run(_send_homeassistant("", {}, "target", "msg")) 238 239 assert "error" in result 240 assert "HASS_URL" in result["error"] or "not configured" in result["error"] 241 242 def test_env_var_fallback(self): 243 resp = _make_aiohttp_resp(200) 244 session_ctx, session = _make_aiohttp_session(resp) 245 246 with patch("aiohttp.ClientSession", return_value=session_ctx), \ 247 patch.dict(os.environ, {"HASS_URL": "https://hass.env.com", "HASS_TOKEN": "env-tok"}, clear=False): 248 result = asyncio.run(_send_homeassistant("", {}, "notify_target", "hi")) 249 250 assert result["success"] is True 251 url = session.post.call_args[0][0] 252 assert "hass.env.com" in url 253 254 255 # --------------------------------------------------------------------------- 256 # _send_dingtalk 257 # --------------------------------------------------------------------------- 258 259 260 class TestSendDingtalk: 261 def _make_httpx_resp(self, status_code=200, json_data=None): 262 resp = MagicMock() 263 resp.status_code = status_code 264 resp.json = MagicMock(return_value=json_data or {"errcode": 0, "errmsg": "ok"}) 265 resp.raise_for_status = MagicMock() 266 return resp 267 268 def _make_httpx_client(self, resp): 269 client = AsyncMock() 270 client.post = AsyncMock(return_value=resp) 271 client_ctx = MagicMock() 272 client_ctx.__aenter__ = AsyncMock(return_value=client) 273 client_ctx.__aexit__ = AsyncMock(return_value=False) 274 return client_ctx, client 275 276 def test_success(self): 277 resp = self._make_httpx_resp(json_data={"errcode": 0, "errmsg": "ok"}) 278 client_ctx, client = self._make_httpx_client(resp) 279 280 with patch("httpx.AsyncClient", return_value=client_ctx): 281 extra = {"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=abc"} 282 result = asyncio.run(_send_dingtalk(extra, "ignored", "hello dingtalk")) 283 284 assert result == {"success": True, "platform": "dingtalk", "chat_id": "ignored"} 285 client.post.assert_awaited_once() 286 call_kwargs = client.post.await_args 287 assert call_kwargs[0][0] == "https://oapi.dingtalk.com/robot/send?access_token=abc" 288 assert call_kwargs[1]["json"] == {"msgtype": "text", "text": {"content": "hello dingtalk"}} 289 290 def test_api_error_in_response_body(self): 291 """DingTalk always returns HTTP 200 but signals errors via errcode.""" 292 resp = self._make_httpx_resp(json_data={"errcode": 310000, "errmsg": "sign not match"}) 293 client_ctx, _ = self._make_httpx_client(resp) 294 295 with patch("httpx.AsyncClient", return_value=client_ctx): 296 result = asyncio.run(_send_dingtalk( 297 {"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=bad"}, 298 "ch", "hi" 299 )) 300 301 assert "error" in result 302 assert "sign not match" in result["error"] 303 304 def test_http_error(self): 305 """If raise_for_status throws, the error is caught and returned.""" 306 resp = self._make_httpx_resp(status_code=429) 307 resp.raise_for_status = MagicMock(side_effect=Exception("429 Too Many Requests")) 308 client_ctx, _ = self._make_httpx_client(resp) 309 310 with patch("httpx.AsyncClient", return_value=client_ctx): 311 result = asyncio.run(_send_dingtalk( 312 {"webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=tok"}, 313 "ch", "hi" 314 )) 315 316 assert "error" in result 317 assert "DingTalk send failed" in result["error"] 318 319 def test_http_error_redacts_access_token_in_exception_text(self): 320 token = "supersecret-access-token-123456789" 321 resp = self._make_httpx_resp(status_code=401) 322 resp.raise_for_status = MagicMock( 323 side_effect=Exception( 324 f"POST https://oapi.dingtalk.com/robot/send?access_token={token} returned 401" 325 ) 326 ) 327 client_ctx, _ = self._make_httpx_client(resp) 328 329 with patch("httpx.AsyncClient", return_value=client_ctx): 330 result = asyncio.run( 331 _send_dingtalk( 332 {"webhook_url": f"https://oapi.dingtalk.com/robot/send?access_token={token}"}, 333 "ch", 334 "hi", 335 ) 336 ) 337 338 assert "error" in result 339 assert token not in result["error"] 340 assert "access_token=***" in result["error"] 341 342 def test_missing_config(self): 343 with patch.dict(os.environ, {"DINGTALK_WEBHOOK_URL": ""}, clear=False): 344 result = asyncio.run(_send_dingtalk({}, "ch", "hi")) 345 346 assert "error" in result 347 assert "DINGTALK_WEBHOOK_URL" in result["error"] or "not configured" in result["error"] 348 349 def test_env_var_fallback(self): 350 resp = self._make_httpx_resp(json_data={"errcode": 0, "errmsg": "ok"}) 351 client_ctx, client = self._make_httpx_client(resp) 352 353 with patch("httpx.AsyncClient", return_value=client_ctx), \ 354 patch.dict(os.environ, {"DINGTALK_WEBHOOK_URL": "https://oapi.dingtalk.com/robot/send?access_token=env"}, clear=False): 355 result = asyncio.run(_send_dingtalk({}, "ch", "hi")) 356 357 assert result["success"] is True 358 call_kwargs = client.post.await_args 359 assert "access_token=env" in call_kwargs[0][0]