test_feishu_approval_buttons.py
1 """Tests for Feishu interactive card approval buttons.""" 2 3 import importlib.util 4 import json 5 import sys 6 from pathlib import Path 7 from types import SimpleNamespace 8 from unittest.mock import AsyncMock, MagicMock, patch 9 10 import pytest 11 12 # --------------------------------------------------------------------------- 13 # Ensure the repo root is importable 14 # --------------------------------------------------------------------------- 15 _repo = str(Path(__file__).resolve().parents[2]) 16 if _repo not in sys.path: 17 sys.path.insert(0, _repo) 18 19 20 # --------------------------------------------------------------------------- 21 # Minimal Feishu mock so FeishuAdapter can be imported without lark-oapi 22 # --------------------------------------------------------------------------- 23 def _ensure_feishu_mocks(): 24 """Provide stubs for lark-oapi / aiohttp.web so the import succeeds.""" 25 if importlib.util.find_spec("lark_oapi") is None and "lark_oapi" not in sys.modules: 26 mod = MagicMock() 27 for name in ( 28 "lark_oapi", "lark_oapi.api.im.v1", 29 "lark_oapi.event", "lark_oapi.event.callback_type", 30 ): 31 sys.modules.setdefault(name, mod) 32 if importlib.util.find_spec("aiohttp") is None and "aiohttp" not in sys.modules: 33 aio = MagicMock() 34 sys.modules.setdefault("aiohttp", aio) 35 sys.modules.setdefault("aiohttp.web", aio.web) 36 37 38 _ensure_feishu_mocks() 39 40 from gateway.config import PlatformConfig 41 import gateway.platforms.feishu as feishu_module 42 from gateway.platforms.feishu import FeishuAdapter 43 44 45 # --------------------------------------------------------------------------- 46 # Helpers 47 # --------------------------------------------------------------------------- 48 49 def _make_adapter() -> FeishuAdapter: 50 """Create a FeishuAdapter with mocked internals.""" 51 config = PlatformConfig(enabled=True) 52 adapter = FeishuAdapter(config) 53 adapter._client = MagicMock() 54 return adapter 55 56 57 def _make_card_action_data( 58 action_value: dict, 59 chat_id: str = "oc_12345", 60 open_id: str = "ou_user1", 61 token: str = "tok_abc", 62 ) -> SimpleNamespace: 63 """Create a mock Feishu card action callback data object.""" 64 return SimpleNamespace( 65 event=SimpleNamespace( 66 token=token, 67 context=SimpleNamespace(open_chat_id=chat_id), 68 operator=SimpleNamespace(open_id=open_id), 69 action=SimpleNamespace( 70 tag="button", 71 value=action_value, 72 ), 73 ), 74 ) 75 76 77 def _close_submitted_coro(coro, _loop): 78 """Close scheduled coroutines in sync-handler tests to avoid unawaited warnings.""" 79 coro.close() 80 return SimpleNamespace(add_done_callback=lambda *_args, **_kwargs: None) 81 82 83 # =========================================================================== 84 # send_exec_approval — interactive card with buttons 85 # =========================================================================== 86 87 class TestFeishuExecApproval: 88 """Test send_exec_approval sends an interactive card.""" 89 90 @pytest.mark.asyncio 91 async def test_sends_interactive_card(self): 92 adapter = _make_adapter() 93 94 mock_response = SimpleNamespace( 95 success=lambda: True, 96 data=SimpleNamespace(message_id="msg_001"), 97 ) 98 with patch.object( 99 adapter, "_feishu_send_with_retry", new_callable=AsyncMock, 100 return_value=mock_response, 101 ) as mock_send: 102 result = await adapter.send_exec_approval( 103 chat_id="oc_12345", 104 command="rm -rf /important", 105 session_key="agent:main:feishu:group:oc_12345", 106 description="dangerous deletion", 107 ) 108 109 assert result.success is True 110 assert result.message_id == "msg_001" 111 112 mock_send.assert_called_once() 113 kwargs = mock_send.call_args[1] 114 assert kwargs["chat_id"] == "oc_12345" 115 assert kwargs["msg_type"] == "interactive" 116 117 # Verify card payload contains the command and buttons 118 card = json.loads(kwargs["payload"]) 119 assert card["header"]["template"] == "orange" 120 assert "rm -rf /important" in card["elements"][0]["content"] 121 assert "dangerous deletion" in card["elements"][0]["content"] 122 123 # Check buttons 124 actions = card["elements"][1]["actions"] 125 assert len(actions) == 4 126 action_names = [a["value"]["hermes_action"] for a in actions] 127 assert action_names == [ 128 "approve_once", "approve_session", "approve_always", "deny" 129 ] 130 131 @pytest.mark.asyncio 132 async def test_stores_approval_state(self): 133 adapter = _make_adapter() 134 135 mock_response = SimpleNamespace( 136 success=lambda: True, 137 data=SimpleNamespace(message_id="msg_002"), 138 ) 139 with patch.object( 140 adapter, "_feishu_send_with_retry", new_callable=AsyncMock, 141 return_value=mock_response, 142 ): 143 await adapter.send_exec_approval( 144 chat_id="oc_12345", 145 command="echo test", 146 session_key="my-session-key", 147 ) 148 149 assert len(adapter._approval_state) == 1 150 approval_id = list(adapter._approval_state.keys())[0] 151 state = adapter._approval_state[approval_id] 152 assert state["session_key"] == "my-session-key" 153 assert state["message_id"] == "msg_002" 154 assert state["chat_id"] == "oc_12345" 155 156 @pytest.mark.asyncio 157 async def test_not_connected(self): 158 adapter = _make_adapter() 159 adapter._client = None 160 result = await adapter.send_exec_approval( 161 chat_id="oc_12345", command="ls", session_key="s" 162 ) 163 assert result.success is False 164 165 @pytest.mark.asyncio 166 async def test_truncates_long_command(self): 167 adapter = _make_adapter() 168 169 mock_response = SimpleNamespace( 170 success=lambda: True, 171 data=SimpleNamespace(message_id="msg_003"), 172 ) 173 with patch.object( 174 adapter, "_feishu_send_with_retry", new_callable=AsyncMock, 175 return_value=mock_response, 176 ) as mock_send: 177 long_cmd = "x" * 5000 178 await adapter.send_exec_approval( 179 chat_id="oc_12345", command=long_cmd, session_key="s" 180 ) 181 182 card = json.loads(mock_send.call_args[1]["payload"]) 183 content = card["elements"][0]["content"] 184 assert "..." in content 185 assert len(content) < 5000 186 187 @pytest.mark.asyncio 188 async def test_multiple_approvals_get_unique_ids(self): 189 adapter = _make_adapter() 190 191 mock_response = SimpleNamespace( 192 success=lambda: True, 193 data=SimpleNamespace(message_id="msg_x"), 194 ) 195 with patch.object( 196 adapter, "_feishu_send_with_retry", new_callable=AsyncMock, 197 return_value=mock_response, 198 ): 199 await adapter.send_exec_approval( 200 chat_id="oc_1", command="cmd1", session_key="s1" 201 ) 202 await adapter.send_exec_approval( 203 chat_id="oc_2", command="cmd2", session_key="s2" 204 ) 205 206 assert len(adapter._approval_state) == 2 207 ids = list(adapter._approval_state.keys()) 208 assert ids[0] != ids[1] 209 210 211 # =========================================================================== 212 # _resolve_approval — approval state pop + gateway resolution 213 # =========================================================================== 214 215 class TestResolveApproval: 216 """Test _resolve_approval pops state and calls resolve_gateway_approval.""" 217 218 @pytest.mark.asyncio 219 async def test_resolves_once(self): 220 adapter = _make_adapter() 221 adapter._approval_state[1] = { 222 "session_key": "agent:main:feishu:group:oc_12345", 223 "message_id": "msg_001", 224 "chat_id": "oc_12345", 225 } 226 227 with patch("tools.approval.resolve_gateway_approval", return_value=1) as mock_resolve: 228 await adapter._resolve_approval(1, "once", "Norbert") 229 230 mock_resolve.assert_called_once_with("agent:main:feishu:group:oc_12345", "once") 231 assert 1 not in adapter._approval_state 232 233 @pytest.mark.asyncio 234 async def test_resolves_deny(self): 235 adapter = _make_adapter() 236 adapter._approval_state[2] = { 237 "session_key": "some-session", 238 "message_id": "msg_002", 239 "chat_id": "oc_12345", 240 } 241 242 with patch("tools.approval.resolve_gateway_approval", return_value=1) as mock_resolve: 243 await adapter._resolve_approval(2, "deny", "Alice") 244 245 mock_resolve.assert_called_once_with("some-session", "deny") 246 247 @pytest.mark.asyncio 248 async def test_resolves_session(self): 249 adapter = _make_adapter() 250 adapter._approval_state[3] = { 251 "session_key": "sess-3", 252 "message_id": "msg_003", 253 "chat_id": "oc_99", 254 } 255 256 with patch("tools.approval.resolve_gateway_approval", return_value=1) as mock_resolve: 257 await adapter._resolve_approval(3, "session", "Bob") 258 259 mock_resolve.assert_called_once_with("sess-3", "session") 260 261 @pytest.mark.asyncio 262 async def test_resolves_always(self): 263 adapter = _make_adapter() 264 adapter._approval_state[4] = { 265 "session_key": "sess-4", 266 "message_id": "msg_004", 267 "chat_id": "oc_55", 268 } 269 270 with patch("tools.approval.resolve_gateway_approval", return_value=1) as mock_resolve: 271 await adapter._resolve_approval(4, "always", "Carol") 272 273 mock_resolve.assert_called_once_with("sess-4", "always") 274 275 @pytest.mark.asyncio 276 async def test_already_resolved_drops_silently(self): 277 adapter = _make_adapter() 278 279 with patch("tools.approval.resolve_gateway_approval") as mock_resolve: 280 await adapter._resolve_approval(99, "once", "Nobody") 281 282 mock_resolve.assert_not_called() 283 284 # =========================================================================== 285 # _handle_card_action_event — non-approval card actions 286 # =========================================================================== 287 288 class TestNonApprovalCardAction: 289 """Non-approval card actions should still route as synthetic commands.""" 290 291 @pytest.mark.asyncio 292 async def test_routes_as_synthetic_command(self): 293 adapter = _make_adapter() 294 295 data = _make_card_action_data( 296 action_value={"custom_action": "something_else"}, 297 token="tok_normal", 298 ) 299 300 with ( 301 patch.object( 302 adapter, "_resolve_sender_profile", new_callable=AsyncMock, 303 return_value={"user_id": "ou_u", "user_name": "Dave", "user_id_alt": None}, 304 ), 305 patch.object(adapter, "get_chat_info", new_callable=AsyncMock, return_value={"name": "Test Chat"}), 306 patch.object(adapter, "_handle_message_with_guards", new_callable=AsyncMock) as mock_handle, 307 ): 308 await adapter._handle_card_action_event(data) 309 310 mock_handle.assert_called_once() 311 event = mock_handle.call_args[0][0] 312 assert "/card button" in event.text 313 314 315 # =========================================================================== 316 # _on_card_action_trigger — inline card response for approval actions 317 # =========================================================================== 318 319 class _FakeCallBackCard: 320 def __init__(self): 321 self.type = None 322 self.data = None 323 324 325 class _FakeP2Response: 326 def __init__(self): 327 self.card = None 328 329 330 @pytest.fixture(autouse=False) 331 def _patch_callback_card_types(monkeypatch): 332 """Provide real-ish P2CardActionTriggerResponse / CallBackCard for tests.""" 333 monkeypatch.setattr(feishu_module, "P2CardActionTriggerResponse", _FakeP2Response) 334 monkeypatch.setattr(feishu_module, "CallBackCard", _FakeCallBackCard) 335 336 337 class TestCardActionCallbackResponse: 338 """Test that _on_card_action_trigger returns updated card inline.""" 339 340 def test_drops_action_when_loop_not_ready(self, _patch_callback_card_types): 341 adapter = _make_adapter() 342 adapter._loop = None 343 data = _make_card_action_data({"hermes_action": "approve_once", "approval_id": 1}) 344 345 with patch("asyncio.run_coroutine_threadsafe") as mock_submit: 346 response = adapter._on_card_action_trigger(data) 347 348 assert response is not None 349 assert response.card is None 350 mock_submit.assert_not_called() 351 352 def test_returns_card_for_approve_action(self, _patch_callback_card_types): 353 adapter = _make_adapter() 354 adapter._loop = MagicMock() 355 adapter._loop.is_closed = MagicMock(return_value=False) 356 data = _make_card_action_data( 357 {"hermes_action": "approve_once", "approval_id": 1}, 358 open_id="ou_bob", 359 ) 360 adapter._sender_name_cache["ou_bob"] = ("Bob", 9999999999) 361 362 with patch("asyncio.run_coroutine_threadsafe", side_effect=_close_submitted_coro): 363 response = adapter._on_card_action_trigger(data) 364 365 assert response is not None 366 assert response.card is not None 367 assert response.card.type == "raw" 368 card = response.card.data 369 assert card["header"]["template"] == "green" 370 assert "Approved once" in card["header"]["title"]["content"] 371 assert "Bob" in card["elements"][0]["content"] 372 373 def test_returns_card_for_deny_action(self, _patch_callback_card_types): 374 adapter = _make_adapter() 375 adapter._loop = MagicMock() 376 adapter._loop.is_closed = MagicMock(return_value=False) 377 data = _make_card_action_data( 378 {"hermes_action": "deny", "approval_id": 2}, 379 ) 380 381 with patch("asyncio.run_coroutine_threadsafe", side_effect=_close_submitted_coro): 382 response = adapter._on_card_action_trigger(data) 383 384 assert response.card is not None 385 card = response.card.data 386 assert card["header"]["template"] == "red" 387 assert "Denied" in card["header"]["title"]["content"] 388 389 def test_ignores_missing_approval_id(self, _patch_callback_card_types): 390 adapter = _make_adapter() 391 adapter._loop = MagicMock() 392 adapter._loop.is_closed = MagicMock(return_value=False) 393 data = _make_card_action_data({"hermes_action": "approve_once"}) 394 395 with patch("asyncio.run_coroutine_threadsafe") as mock_submit: 396 response = adapter._on_card_action_trigger(data) 397 398 assert response is not None 399 assert response.card is None 400 mock_submit.assert_not_called() 401 402 def test_no_card_for_non_approval_action(self, _patch_callback_card_types): 403 adapter = _make_adapter() 404 adapter._loop = MagicMock() 405 adapter._loop.is_closed = MagicMock(return_value=False) 406 data = _make_card_action_data({"some_other": "value"}) 407 408 with patch("asyncio.run_coroutine_threadsafe", side_effect=_close_submitted_coro): 409 response = adapter._on_card_action_trigger(data) 410 411 assert response is not None 412 assert response.card is None 413 414 def test_falls_back_to_open_id_when_name_not_cached(self, _patch_callback_card_types): 415 adapter = _make_adapter() 416 adapter._loop = MagicMock() 417 adapter._loop.is_closed = MagicMock(return_value=False) 418 data = _make_card_action_data( 419 {"hermes_action": "approve_session", "approval_id": 3}, 420 open_id="ou_unknown", 421 ) 422 423 with patch("asyncio.run_coroutine_threadsafe", side_effect=_close_submitted_coro): 424 response = adapter._on_card_action_trigger(data) 425 426 card = response.card.data 427 assert "ou_unknown" in card["elements"][0]["content"] 428 429 def test_ignores_expired_cached_name(self, _patch_callback_card_types): 430 adapter = _make_adapter() 431 adapter._loop = MagicMock() 432 adapter._loop.is_closed = MagicMock(return_value=False) 433 data = _make_card_action_data( 434 {"hermes_action": "approve_once", "approval_id": 4}, 435 open_id="ou_expired", 436 ) 437 adapter._sender_name_cache["ou_expired"] = ("Old Name", 1) 438 439 with patch("asyncio.run_coroutine_threadsafe", side_effect=_close_submitted_coro): 440 response = adapter._on_card_action_trigger(data) 441 442 card = response.card.data 443 assert "Old Name" not in card["elements"][0]["content"] 444 assert "ou_expired" in card["elements"][0]["content"]