test_feishu_onboard.py
1 """Tests for gateway.platforms.feishu — Feishu scan-to-create registration.""" 2 3 import json 4 from unittest.mock import patch, MagicMock 5 import pytest 6 7 8 def _mock_urlopen(response_data, status=200): 9 """Create a mock for urllib.request.urlopen that returns JSON response_data.""" 10 mock_response = MagicMock() 11 mock_response.read.return_value = json.dumps(response_data).encode("utf-8") 12 mock_response.status = status 13 mock_response.__enter__ = lambda s: s 14 mock_response.__exit__ = MagicMock(return_value=False) 15 return mock_response 16 17 18 class TestPostRegistration: 19 """Tests for the low-level HTTP helper.""" 20 21 @patch("gateway.platforms.feishu.urlopen") 22 def test_post_registration_returns_parsed_json(self, mock_urlopen_fn): 23 from gateway.platforms.feishu import _post_registration 24 25 mock_urlopen_fn.return_value = _mock_urlopen({"nonce": "abc", "supported_auth_methods": ["client_secret"]}) 26 result = _post_registration("https://accounts.feishu.cn", {"action": "init"}) 27 assert result["nonce"] == "abc" 28 assert "client_secret" in result["supported_auth_methods"] 29 30 @patch("gateway.platforms.feishu.urlopen") 31 def test_post_registration_sends_form_encoded_body(self, mock_urlopen_fn): 32 from gateway.platforms.feishu import _post_registration 33 34 mock_urlopen_fn.return_value = _mock_urlopen({}) 35 _post_registration("https://accounts.feishu.cn", {"action": "init", "key": "val"}) 36 call_args = mock_urlopen_fn.call_args 37 request = call_args[0][0] 38 body = request.data.decode("utf-8") 39 assert "action=init" in body 40 assert "key=val" in body 41 assert request.get_header("Content-type") == "application/x-www-form-urlencoded" 42 43 44 class TestInitRegistration: 45 """Tests for the init step.""" 46 47 @patch("gateway.platforms.feishu.urlopen") 48 def test_init_succeeds_when_client_secret_supported(self, mock_urlopen_fn): 49 from gateway.platforms.feishu import _init_registration 50 51 mock_urlopen_fn.return_value = _mock_urlopen({ 52 "nonce": "abc", 53 "supported_auth_methods": ["client_secret"], 54 }) 55 _init_registration("feishu") 56 57 @patch("gateway.platforms.feishu.urlopen") 58 def test_init_raises_when_client_secret_not_supported(self, mock_urlopen_fn): 59 from gateway.platforms.feishu import _init_registration 60 61 mock_urlopen_fn.return_value = _mock_urlopen({ 62 "nonce": "abc", 63 "supported_auth_methods": ["other_method"], 64 }) 65 with pytest.raises(RuntimeError, match="client_secret"): 66 _init_registration("feishu") 67 68 @patch("gateway.platforms.feishu.urlopen") 69 def test_init_uses_lark_url_for_lark_domain(self, mock_urlopen_fn): 70 from gateway.platforms.feishu import _init_registration 71 72 mock_urlopen_fn.return_value = _mock_urlopen({ 73 "nonce": "abc", 74 "supported_auth_methods": ["client_secret"], 75 }) 76 _init_registration("lark") 77 call_args = mock_urlopen_fn.call_args 78 request = call_args[0][0] 79 assert "larksuite.com" in request.full_url 80 81 82 class TestBeginRegistration: 83 """Tests for the begin step.""" 84 85 @patch("gateway.platforms.feishu.urlopen") 86 def test_begin_returns_device_code_and_qr_url(self, mock_urlopen_fn): 87 from gateway.platforms.feishu import _begin_registration 88 89 mock_urlopen_fn.return_value = _mock_urlopen({ 90 "device_code": "dc_123", 91 "verification_uri_complete": "https://accounts.feishu.cn/qr/abc", 92 "user_code": "ABCD-1234", 93 "interval": 5, 94 "expire_in": 600, 95 }) 96 result = _begin_registration("feishu") 97 assert result["device_code"] == "dc_123" 98 assert "qr_url" in result 99 assert "accounts.feishu.cn" in result["qr_url"] 100 assert result["user_code"] == "ABCD-1234" 101 assert result["interval"] == 5 102 assert result["expire_in"] == 600 103 104 @patch("gateway.platforms.feishu.urlopen") 105 def test_begin_sends_correct_archetype(self, mock_urlopen_fn): 106 from gateway.platforms.feishu import _begin_registration 107 108 mock_urlopen_fn.return_value = _mock_urlopen({ 109 "device_code": "dc_123", 110 "verification_uri_complete": "https://example.com/qr", 111 "user_code": "X", 112 "interval": 5, 113 "expire_in": 600, 114 }) 115 _begin_registration("feishu") 116 request = mock_urlopen_fn.call_args[0][0] 117 body = request.data.decode("utf-8") 118 assert "archetype=PersonalAgent" in body 119 assert "auth_method=client_secret" in body 120 121 122 class TestPollRegistration: 123 """Tests for the poll step.""" 124 125 @patch("gateway.platforms.feishu.time") 126 @patch("gateway.platforms.feishu.urlopen") 127 def test_poll_returns_credentials_on_success(self, mock_urlopen_fn, mock_time): 128 from gateway.platforms.feishu import _poll_registration 129 130 mock_time.time.side_effect = [0, 1] 131 mock_time.sleep = MagicMock() 132 133 mock_urlopen_fn.return_value = _mock_urlopen({ 134 "client_id": "cli_app123", 135 "client_secret": "secret456", 136 "user_info": {"open_id": "ou_owner", "tenant_brand": "feishu"}, 137 }) 138 result = _poll_registration( 139 device_code="dc_123", interval=1, expire_in=60, domain="feishu" 140 ) 141 assert result is not None 142 assert result["app_id"] == "cli_app123" 143 assert result["app_secret"] == "secret456" 144 assert result["domain"] == "feishu" 145 assert result["open_id"] == "ou_owner" 146 147 @patch("gateway.platforms.feishu.time") 148 @patch("gateway.platforms.feishu.urlopen") 149 def test_poll_switches_domain_on_lark_tenant_brand(self, mock_urlopen_fn, mock_time): 150 from gateway.platforms.feishu import _poll_registration 151 152 mock_time.time.side_effect = [0, 1, 2] 153 mock_time.sleep = MagicMock() 154 155 pending_resp = _mock_urlopen({ 156 "error": "authorization_pending", 157 "user_info": {"tenant_brand": "lark"}, 158 }) 159 success_resp = _mock_urlopen({ 160 "client_id": "cli_lark", 161 "client_secret": "secret_lark", 162 "user_info": {"open_id": "ou_lark", "tenant_brand": "lark"}, 163 }) 164 mock_urlopen_fn.side_effect = [pending_resp, success_resp] 165 166 result = _poll_registration( 167 device_code="dc_123", interval=0, expire_in=60, domain="feishu" 168 ) 169 assert result is not None 170 assert result["domain"] == "lark" 171 172 @patch("gateway.platforms.feishu.time") 173 @patch("gateway.platforms.feishu.urlopen") 174 def test_poll_success_with_lark_brand_in_same_response(self, mock_urlopen_fn, mock_time): 175 """Credentials and lark tenant_brand in one response must not be discarded.""" 176 from gateway.platforms.feishu import _poll_registration 177 178 mock_time.time.side_effect = [0, 1] 179 mock_time.sleep = MagicMock() 180 181 mock_urlopen_fn.return_value = _mock_urlopen({ 182 "client_id": "cli_lark_direct", 183 "client_secret": "secret_lark_direct", 184 "user_info": {"open_id": "ou_lark_direct", "tenant_brand": "lark"}, 185 }) 186 result = _poll_registration( 187 device_code="dc_123", interval=1, expire_in=60, domain="feishu" 188 ) 189 assert result is not None 190 assert result["app_id"] == "cli_lark_direct" 191 assert result["domain"] == "lark" 192 assert result["open_id"] == "ou_lark_direct" 193 194 @patch("gateway.platforms.feishu.time") 195 @patch("gateway.platforms.feishu.urlopen") 196 def test_poll_returns_none_on_access_denied(self, mock_urlopen_fn, mock_time): 197 from gateway.platforms.feishu import _poll_registration 198 199 mock_time.time.side_effect = [0, 1] 200 mock_time.sleep = MagicMock() 201 202 mock_urlopen_fn.return_value = _mock_urlopen({ 203 "error": "access_denied", 204 }) 205 result = _poll_registration( 206 device_code="dc_123", interval=1, expire_in=60, domain="feishu" 207 ) 208 assert result is None 209 210 @patch("gateway.platforms.feishu.time") 211 @patch("gateway.platforms.feishu.urlopen") 212 def test_poll_returns_none_on_timeout(self, mock_urlopen_fn, mock_time): 213 from gateway.platforms.feishu import _poll_registration 214 215 mock_time.time.side_effect = [0, 999] 216 mock_time.sleep = MagicMock() 217 218 mock_urlopen_fn.return_value = _mock_urlopen({ 219 "error": "authorization_pending", 220 }) 221 result = _poll_registration( 222 device_code="dc_123", interval=1, expire_in=1, domain="feishu" 223 ) 224 assert result is None 225 226 227 class TestRenderQr: 228 """Tests for QR code terminal rendering.""" 229 230 @patch("gateway.platforms.feishu._qrcode_mod", create=True) 231 def test_render_qr_returns_true_on_success(self, mock_qrcode_mod): 232 from gateway.platforms.feishu import _render_qr 233 234 mock_qr = MagicMock() 235 mock_qrcode_mod.QRCode.return_value = mock_qr 236 assert _render_qr("https://example.com/qr") is True 237 mock_qr.add_data.assert_called_once_with("https://example.com/qr") 238 mock_qr.make.assert_called_once_with(fit=True) 239 mock_qr.print_ascii.assert_called_once() 240 241 def test_render_qr_returns_false_when_qrcode_missing(self): 242 from gateway.platforms.feishu import _render_qr 243 244 with patch("gateway.platforms.feishu._qrcode_mod", None): 245 assert _render_qr("https://example.com/qr") is False 246 247 248 class TestProbeBot: 249 """Tests for bot connectivity verification.""" 250 251 @patch("gateway.platforms.feishu.FEISHU_AVAILABLE", True) 252 def test_probe_returns_bot_info_on_success(self): 253 from gateway.platforms.feishu import probe_bot 254 255 with patch("gateway.platforms.feishu._probe_bot_sdk") as mock_sdk: 256 mock_sdk.return_value = {"bot_name": "TestBot", "bot_open_id": "ou_bot123"} 257 result = probe_bot("cli_app", "secret", "feishu") 258 259 assert result is not None 260 assert result["bot_name"] == "TestBot" 261 assert result["bot_open_id"] == "ou_bot123" 262 263 @patch("gateway.platforms.feishu.FEISHU_AVAILABLE", True) 264 def test_probe_returns_none_on_failure(self): 265 from gateway.platforms.feishu import probe_bot 266 267 with patch("gateway.platforms.feishu._probe_bot_sdk") as mock_sdk: 268 mock_sdk.return_value = None 269 result = probe_bot("bad_id", "bad_secret", "feishu") 270 271 assert result is None 272 273 @patch("gateway.platforms.feishu.FEISHU_AVAILABLE", False) 274 @patch("gateway.platforms.feishu.urlopen") 275 def test_http_fallback_when_sdk_unavailable(self, mock_urlopen_fn): 276 """Without lark_oapi, probe falls back to raw HTTP.""" 277 from gateway.platforms.feishu import probe_bot 278 279 token_resp = _mock_urlopen({"code": 0, "tenant_access_token": "t-123"}) 280 bot_resp = _mock_urlopen({"code": 0, "bot": {"bot_name": "HttpBot", "open_id": "ou_http"}}) 281 mock_urlopen_fn.side_effect = [token_resp, bot_resp] 282 283 result = probe_bot("cli_app", "secret", "feishu") 284 assert result is not None 285 assert result["bot_name"] == "HttpBot" 286 287 @patch("gateway.platforms.feishu.FEISHU_AVAILABLE", False) 288 @patch("gateway.platforms.feishu.urlopen") 289 def test_http_fallback_returns_none_on_network_error(self, mock_urlopen_fn): 290 from gateway.platforms.feishu import probe_bot 291 from urllib.error import URLError 292 293 mock_urlopen_fn.side_effect = URLError("connection refused") 294 result = probe_bot("cli_app", "secret", "feishu") 295 assert result is None 296 297 298 class TestQrRegister: 299 """Tests for the public qr_register entry point.""" 300 301 @patch("gateway.platforms.feishu.probe_bot") 302 @patch("gateway.platforms.feishu._render_qr") 303 @patch("gateway.platforms.feishu._poll_registration") 304 @patch("gateway.platforms.feishu._begin_registration") 305 @patch("gateway.platforms.feishu._init_registration") 306 def test_qr_register_success_flow( 307 self, mock_init, mock_begin, mock_poll, mock_render, mock_probe 308 ): 309 from gateway.platforms.feishu import qr_register 310 311 mock_begin.return_value = { 312 "device_code": "dc_123", 313 "qr_url": "https://example.com/qr", 314 "user_code": "ABCD", 315 "interval": 1, 316 "expire_in": 60, 317 } 318 mock_poll.return_value = { 319 "app_id": "cli_app", 320 "app_secret": "secret", 321 "domain": "feishu", 322 "open_id": "ou_owner", 323 } 324 mock_probe.return_value = {"bot_name": "MyBot", "bot_open_id": "ou_bot"} 325 326 result = qr_register() 327 assert result is not None 328 assert result["app_id"] == "cli_app" 329 assert result["app_secret"] == "secret" 330 assert result["bot_name"] == "MyBot" 331 mock_init.assert_called_once() 332 mock_render.assert_called_once() 333 334 @patch("gateway.platforms.feishu._init_registration") 335 def test_qr_register_returns_none_on_init_failure(self, mock_init): 336 from gateway.platforms.feishu import qr_register 337 338 mock_init.side_effect = RuntimeError("not supported") 339 result = qr_register() 340 assert result is None 341 342 @patch("gateway.platforms.feishu._render_qr") 343 @patch("gateway.platforms.feishu._poll_registration") 344 @patch("gateway.platforms.feishu._begin_registration") 345 @patch("gateway.platforms.feishu._init_registration") 346 def test_qr_register_returns_none_on_poll_failure( 347 self, mock_init, mock_begin, mock_poll, mock_render 348 ): 349 from gateway.platforms.feishu import qr_register 350 351 mock_begin.return_value = { 352 "device_code": "dc_123", 353 "qr_url": "https://example.com/qr", 354 "user_code": "ABCD", 355 "interval": 1, 356 "expire_in": 60, 357 } 358 mock_poll.return_value = None 359 360 result = qr_register() 361 assert result is None 362 363 # -- Contract: expected errors → None, unexpected errors → propagate -- 364 365 @patch("gateway.platforms.feishu._init_registration") 366 def test_qr_register_returns_none_on_network_error(self, mock_init): 367 """URLError (network down) is an expected failure → None.""" 368 from gateway.platforms.feishu import qr_register 369 from urllib.error import URLError 370 371 mock_init.side_effect = URLError("DNS resolution failed") 372 result = qr_register() 373 assert result is None 374 375 @patch("gateway.platforms.feishu._init_registration") 376 def test_qr_register_returns_none_on_json_error(self, mock_init): 377 """Malformed server response is an expected failure → None.""" 378 from gateway.platforms.feishu import qr_register 379 380 mock_init.side_effect = json.JSONDecodeError("bad json", "", 0) 381 result = qr_register() 382 assert result is None 383 384 @patch("gateway.platforms.feishu._init_registration") 385 def test_qr_register_propagates_unexpected_errors(self, mock_init): 386 """Bugs (e.g. AttributeError) must not be swallowed — they propagate.""" 387 from gateway.platforms.feishu import qr_register 388 389 mock_init.side_effect = AttributeError("some internal bug") 390 with pytest.raises(AttributeError, match="some internal bug"): 391 qr_register() 392 393 # -- Negative paths: partial/malformed server responses -- 394 395 @patch("gateway.platforms.feishu._render_qr") 396 @patch("gateway.platforms.feishu._begin_registration") 397 @patch("gateway.platforms.feishu._init_registration") 398 def test_qr_register_returns_none_when_begin_missing_device_code( 399 self, mock_init, mock_begin, mock_render 400 ): 401 """Server returns begin response without device_code → RuntimeError → None.""" 402 from gateway.platforms.feishu import qr_register 403 404 mock_begin.side_effect = RuntimeError("Feishu registration did not return a device_code") 405 result = qr_register() 406 assert result is None 407 408 @patch("gateway.platforms.feishu.probe_bot") 409 @patch("gateway.platforms.feishu._render_qr") 410 @patch("gateway.platforms.feishu._poll_registration") 411 @patch("gateway.platforms.feishu._begin_registration") 412 @patch("gateway.platforms.feishu._init_registration") 413 def test_qr_register_succeeds_even_when_probe_fails( 414 self, mock_init, mock_begin, mock_poll, mock_render, mock_probe 415 ): 416 """Registration succeeds but probe fails → result with bot_name=None.""" 417 from gateway.platforms.feishu import qr_register 418 419 mock_begin.return_value = { 420 "device_code": "dc_123", 421 "qr_url": "https://example.com/qr", 422 "user_code": "ABCD", 423 "interval": 1, 424 "expire_in": 60, 425 } 426 mock_poll.return_value = { 427 "app_id": "cli_app", 428 "app_secret": "secret", 429 "domain": "feishu", 430 "open_id": "ou_owner", 431 } 432 mock_probe.return_value = None # probe failed 433 434 result = qr_register() 435 assert result is not None 436 assert result["app_id"] == "cli_app" 437 assert result["bot_name"] is None 438 assert result["bot_open_id"] is None