test_auth_setup.py
1 """Google Ads OAuthフロー + セットアップウィザードのテスト（TDD: RED -> GREEN -> IMPROVE）""" 2 3 from __future__ import annotations 4 5 import http.client 6 import json 7 import threading 8 import time 9 from pathlib import Path 10 from unittest.mock import AsyncMock, MagicMock, patch 11 12 import pytest 13 14 from mureo.auth import GoogleAdsCredentials 15 16 17 # --------------------------------------------------------------------------- 18 # 1. ローカルサーバーがcallbackを受信できること（Meta Ads用に残す） 19 # --------------------------------------------------------------------------- 20 21 22 @pytest.mark.unit 23 def test_oauth_callback_server() -> None: 24 """ローカルHTTPサーバーがOAuthコールバックを受信できること""" 25 from mureo.auth_setup import OAuthCallbackServer 26 27 server = OAuthCallbackServer(port=0) # 空きポート自動選択 28 actual_port = server.server.server_address[1] 29 30 # サーバーをバックグラウンドで起動 31 server_thread = threading.Thread(target=server.wait_for_callback, daemon=True) 32 server_thread.start() 33 34 # コールバックリクエストを送信 35 time.sleep(0.1) # サーバー起動待ち 36 conn = http.client.HTTPConnection("localhost", actual_port) 37 conn.request("GET", "/callback?code=test-auth-code-123") 38 response = conn.getresponse() 39 conn.close() 40 41 assert response.status == 200 42 assert server.authorization_code == "test-auth-code-123" 43 44 45 @pytest.mark.unit 46 def test_oauth_callback_server_error() -> None: 47 """OAuthコールバックでエラーパラメータが返された場合""" 48 from mureo.auth_setup import OAuthCallbackServer 49 50 server = OAuthCallbackServer(port=0) 51 actual_port = server.server.server_address[1] 52 53 server_thread = threading.Thread(target=server.wait_for_callback, daemon=True) 54 server_thread.start() 55 56 time.sleep(0.1) 57 conn = http.client.HTTPConnection("localhost", actual_port) 58 conn.request("GET", "/callback?error=access_denied") 59 response = conn.getresponse() 60 conn.close() 61 62 assert response.status == 200 63 assert server.authorization_code is None 64 assert server.error == "access_denied" 65 66 67 # --------------------------------------------------------------------------- 68 # 4. credentials.json 新規保存 69 # --------------------------------------------------------------------------- 70 71 72 @pytest.mark.unit 73 def test_save_credentials_new(tmp_path: Path) -> None: 74 """新規のcredentials.jsonが正しく保存されること""" 75 from mureo.auth_setup import save_credentials 76 77 cred_path = tmp_path / "credentials.json" 78 79 google_creds = GoogleAdsCredentials( 80 developer_token="dev-tok", 81 client_id="cid", 82 client_secret="csec", 83 refresh_token="rtok", 84 login_customer_id="1234567890", 85 ) 86 87 save_credentials( 88 path=cred_path, 89 google=google_creds, 90 customer_id="1234567890", 91 ) 92 93 assert cred_path.exists() 94 data = json.loads(cred_path.read_text(encoding="utf-8")) 95 assert data["google_ads"]["developer_token"] == "dev-tok" 96 assert data["google_ads"]["client_id"] == "cid" 97 assert data["google_ads"]["client_secret"] == "csec" 98 assert data["google_ads"]["refresh_token"] == "rtok" 99 assert data["google_ads"]["login_customer_id"] == "1234567890" 100 101 102 # --------------------------------------------------------------------------- 103 # 5. credentials.json 既存データとマージ保存 104 # --------------------------------------------------------------------------- 105 106 107 @pytest.mark.unit 108 def test_save_credentials_merge(tmp_path: Path) -> None: 109 """既存のcredentials.jsonにマージして保存されること""" 110 from mureo.auth_setup import save_credentials 111 112 cred_path = tmp_path / "credentials.json" 113 114 # 既存のMeta Ads認証情報を事前に保存 115 existing = { 116 "meta_ads": { 117 "access_token": "existing-meta-token", 118 "app_id": "existing-app-id", 119 } 120 } 121 cred_path.write_text(json.dumps(existing), encoding="utf-8") 122 123 # Google Ads認証情報を追加 124 google_creds = GoogleAdsCredentials( 125 developer_token="dev-tok", 126 client_id="cid", 127 client_secret="csec", 128 refresh_token="rtok", 129 ) 130 131 save_credentials(path=cred_path, google=google_creds) 132 133 data = json.loads(cred_path.read_text(encoding="utf-8")) 134 # Google Adsが追加されている 135 assert data["google_ads"]["developer_token"] == "dev-tok" 136 # Meta Adsが保持されている 137 assert data["meta_ads"]["access_token"] == "existing-meta-token" 138 139 140 # --------------------------------------------------------------------------- 141 # 6. ディレクトリが存在しない場合に作成 142 # --------------------------------------------------------------------------- 143 144 145 @pytest.mark.unit 146 def test_save_credentials_creates_directory(tmp_path: Path) -> None: 147 """~/.mureo/ ディレクトリが存在しない場合に自動作成されること""" 148 from mureo.auth_setup import save_credentials 149 150 nested_path = tmp_path / "nonexistent" / "dir" / "credentials.json" 151 152 google_creds = GoogleAdsCredentials( 153 developer_token="dev-tok", 154 client_id="cid", 155 client_secret="csec", 156 refresh_token="rtok", 157 ) 158 159 save_credentials(path=nested_path, google=google_creds) 160 161 assert nested_path.exists() 162 data = json.loads(nested_path.read_text(encoding="utf-8")) 163 assert data["google_ads"]["developer_token"] == "dev-tok" 164 165 166 # --------------------------------------------------------------------------- 167 # 7. アカウント一覧取得（APIモック） 168 # --------------------------------------------------------------------------- 169 170 171 @pytest.mark.unit 172 @pytest.mark.asyncio 173 async def test_list_accessible_accounts() -> None: 174 """Google Ads APIでアクセス可能なアカウント一覧を取得できること""" 175 from mureo.auth_setup import list_accessible_accounts 176 177 creds = GoogleAdsCredentials( 178 developer_token="dev-tok", 179 client_id="cid", 180 client_secret="csec", 181 refresh_token="rtok", 182 ) 183 184 # Google Ads SDK直接呼び出しのモック 185 mock_ga_client = MagicMock() 186 187 # CustomerService（アカウント一覧取得用） 188 mock_customer_service = MagicMock() 189 mock_response = MagicMock() 190 mock_response.resource_names = [ 191 "customers/1234567890", 192 "customers/9876543210", 193 ] 194 mock_customer_service.list_accessible_customers.return_value = mock_response 195 196 # GoogleAdsService（アカウント名取得用） 197 # どちらも非マネージャーアカウント 198 mock_ga_service = MagicMock() 199 mock_row_1 = MagicMock() 200 mock_row_1.customer.descriptive_name = "テストアカウント1" 201 mock_row_1.customer.manager = False 202 mock_row_2 = MagicMock() 203 mock_row_2.customer.descriptive_name = "テストアカウント2" 204 mock_row_2.customer.manager = False 205 mock_ga_service.search.side_effect = [[mock_row_1], [mock_row_2]] 206 207 def _get_service(name: str) -> MagicMock: 208 if name == "CustomerService": 209 return mock_customer_service 210 return mock_ga_service 211 212 mock_ga_client.get_service.side_effect = _get_service 213 214 with patch( 215 "google.ads.googleads.client.GoogleAdsClient", return_value=mock_ga_client 216 ): 217 accounts = await list_accessible_accounts(creds) 218 219 assert len(accounts) == 2 220 assert accounts[0]["id"] == "1234567890" 221 assert accounts[0]["name"] == "テストアカウント1" 222 assert accounts[0]["is_manager"] is False 223 assert accounts[0]["parent_id"] is None 224 assert accounts[1]["id"] == "9876543210" 225 assert accounts[1]["name"] == "テストアカウント2" 226 assert accounts[1]["is_manager"] is False 227 assert accounts[1]["parent_id"] is None 228 229 230 @pytest.mark.unit 231 @pytest.mark.asyncio 232 async def test_list_accessible_accounts_traverses_mcc_children() -> None: 233 """MCCアカウントの配下の子アカウントも列挙できること""" 234 from mureo.auth_setup import list_accessible_accounts 235 236 creds = GoogleAdsCredentials( 237 developer_token="dev-tok", 238 client_id="cid", 239 client_secret="csec", 240 refresh_token="rtok", 241 ) 242 243 mock_ga_client = MagicMock() 244 245 # listAccessibleCustomersはMCC 1件だけを返す 246 mock_customer_service = MagicMock() 247 mock_response = MagicMock() 248 mock_response.resource_names = ["customers/1111111111"] 249 mock_customer_service.list_accessible_customers.return_value = mock_response 250 251 # 1回目のsearch: MCCの情報 252 mcc_row = MagicMock() 253 mcc_row.customer.descriptive_name = "親MCC" 254 mcc_row.customer.manager = True 255 256 # 2回目のsearch: MCC配下の子アカウント2件 257 child_row_1 = MagicMock() 258 child_row_1.customer_client.id = 2222222222 259 child_row_1.customer_client.descriptive_name = "子アカウントA" 260 child_row_1.customer_client.manager = False 261 child_row_2 = MagicMock() 262 child_row_2.customer_client.id = 3333333333 263 child_row_2.customer_client.descriptive_name = "子アカウントB" 264 child_row_2.customer_client.manager = False 265 266 mock_ga_service = MagicMock() 267 mock_ga_service.search.side_effect = [ 268 [mcc_row], 269 [child_row_1, child_row_2], 270 ] 271 272 def _get_service(name: str) -> MagicMock: 273 if name == "CustomerService": 274 return mock_customer_service 275 return mock_ga_service 276 277 mock_ga_client.get_service.side_effect = _get_service 278 279 with patch( 280 "google.ads.googleads.client.GoogleAdsClient", return_value=mock_ga_client 281 ): 282 accounts = await list_accessible_accounts(creds) 283 284 assert len(accounts) == 3 285 # 親MCC 286 assert accounts[0]["id"] == "1111111111" 287 assert accounts[0]["name"] == "親MCC" 288 assert accounts[0]["is_manager"] is True 289 assert accounts[0]["parent_id"] is None 290 # 子アカウントA（parent_id = MCC） 291 assert accounts[1]["id"] == "2222222222" 292 assert accounts[1]["name"] == "子アカウントA" 293 assert accounts[1]["is_manager"] is False 294 assert accounts[1]["parent_id"] == "1111111111" 295 # 子アカウントB 296 assert accounts[2]["id"] == "3333333333" 297 assert accounts[2]["name"] == "子アカウントB" 298 assert accounts[2]["is_manager"] is False 299 assert accounts[2]["parent_id"] == "1111111111" 300 301 302 @pytest.mark.unit 303 @pytest.mark.asyncio 304 async def test_list_accessible_accounts_empty() -> None: 305 """アクセス可能なアカウントがない場合は空リストを返すこと""" 306 from mureo.auth_setup import list_accessible_accounts 307 308 creds = GoogleAdsCredentials( 309 developer_token="dev-tok", 310 client_id="cid", 311 client_secret="csec", 312 refresh_token="rtok", 313 ) 314 315 mock_ga_client = MagicMock() 316 mock_customer_service = MagicMock() 317 mock_response = MagicMock() 318 mock_response.resource_names = [] 319 mock_customer_service.list_accessible_customers.return_value = mock_response 320 mock_ga_client.get_service.return_value = mock_customer_service 321 322 with patch( 323 "google.ads.googleads.client.GoogleAdsClient", return_value=mock_ga_client 324 ): 325 accounts = await list_accessible_accounts(creds) 326 327 assert accounts == [] 328 329 330 # --------------------------------------------------------------------------- 331 # 8. セットアップ全体フロー（input/OAuth/APIすべてモック） 332 # --------------------------------------------------------------------------- 333 334 335 @pytest.mark.unit 336 @pytest.mark.asyncio 337 async def test_setup_google_ads_flow(tmp_path: Path) -> None: 338 """Google Adsセットアップの全体フローが正しく動作すること""" 339 from mureo.auth_setup import OAuthResult, setup_google_ads 340 341 cred_path = tmp_path / "credentials.json" 342 343 # ユーザー入力のモック 344 user_inputs = iter( 345 [ 346 "test-developer-token", # Developer Token 347 "test-client-id.apps.googleusercontent.com", # Client ID 348 "test-client-secret", # Client Secret 349 ] 350 ) 351 352 mock_accounts = [ 353 { 354 "id": "1234567890", 355 "name": "Account 1234567890", 356 "is_manager": False, 357 "parent_id": None, 358 }, 359 { 360 "id": "9876543210", 361 "name": "Account 9876543210", 362 "is_manager": False, 363 "parent_id": None, 364 }, 365 ] 366 367 mock_oauth_result = OAuthResult( 368 refresh_token="1//test-refresh-token", 369 access_token="ya29.test-access-token", 370 ) 371 372 with ( 373 patch("mureo.auth_setup.input_func", side_effect=user_inputs), 374 patch("mureo.auth_setup._select_account", return_value="1234567890"), 375 patch( 376 "mureo.auth_setup.run_google_oauth", 377 new_callable=AsyncMock, 378 return_value=mock_oauth_result, 379 ), 380 patch( 381 "mureo.auth_setup.list_accessible_accounts", 382 new_callable=AsyncMock, 383 return_value=mock_accounts, 384 ), 385 ): 386 result = await setup_google_ads(credentials_path=cred_path) 387 388 assert result.developer_token == "test-developer-token" 389 assert result.client_id == "test-client-id.apps.googleusercontent.com" 390 assert result.client_secret == "test-client-secret" 391 assert result.refresh_token == "1//test-refresh-token" 392 assert result.login_customer_id == "1234567890" 393 394 # credentials.jsonに保存されていること 395 data = json.loads(cred_path.read_text(encoding="utf-8")) 396 assert data["google_ads"]["developer_token"] == "test-developer-token" 397 assert data["google_ads"]["login_customer_id"] == "1234567890" 398 399 400 @pytest.mark.unit 401 @pytest.mark.asyncio 402 async def test_setup_google_ads_flow_no_accounts(tmp_path: Path) -> None: 403 """アカウントが見つからない場合もcredentials.jsonが保存されること""" 404 from mureo.auth_setup import OAuthResult, setup_google_ads 405 406 cred_path = tmp_path / "credentials.json" 407 408 user_inputs = iter( 409 [ 410 "test-developer-token", 411 "test-client-id", 412 "test-client-secret", 413 ] 414 ) 415 416 mock_oauth_result = OAuthResult( 417 refresh_token="1//test-refresh-token", 418 access_token="ya29.test-access-token", 419 ) 420 421 with ( 422 patch("mureo.auth_setup.input_func", side_effect=user_inputs), 423 patch( 424 "mureo.auth_setup.run_google_oauth", 425 new_callable=AsyncMock, 426 return_value=mock_oauth_result, 427 ), 428 patch( 429 "mureo.auth_setup.list_accessible_accounts", 430 new_callable=AsyncMock, 431 return_value=[], 432 ), 433 ): 434 result = await setup_google_ads(credentials_path=cred_path) 435 436 assert result.developer_token == "test-developer-token" 437 assert result.refresh_token == "1//test-refresh-token" 438 assert result.login_customer_id is None 439 440 data = json.loads(cred_path.read_text(encoding="utf-8")) 441 assert data["google_ads"]["developer_token"] == "test-developer-token" 442 443 444 @pytest.mark.unit 445 @pytest.mark.asyncio 446 async def test_setup_google_ads_flow_selects_mcc_child(tmp_path: Path) -> None: 447 """MCC配下の子アカウントを選択した場合、login_customer_idは親MCCになること""" 448 from mureo.auth_setup import OAuthResult, setup_google_ads 449 450 cred_path = tmp_path / "credentials.json" 451 452 user_inputs = iter( 453 [ 454 "test-developer-token", 455 "test-client-id", 456 "test-client-secret", 457 ] 458 ) 459 460 # MCC + 2つの子アカウント（子はparent_id付き） 461 mock_accounts = [ 462 { 463 "id": "1111111111", 464 "name": "親MCC", 465 "is_manager": True, 466 "parent_id": None, 467 }, 468 { 469 "id": "2222222222", 470 "name": "子アカウントA", 471 "is_manager": False, 472 "parent_id": "1111111111", 473 }, 474 ] 475 476 mock_oauth_result = OAuthResult( 477 refresh_token="1//test-refresh-token", 478 access_token="ya29.test-access-token", 479 ) 480 481 with ( 482 patch("mureo.auth_setup.input_func", side_effect=user_inputs), 483 # 子アカウントAを選択 484 patch("mureo.auth_setup._select_account", return_value="2222222222"), 485 patch( 486 "mureo.auth_setup.run_google_oauth", 487 new_callable=AsyncMock, 488 return_value=mock_oauth_result, 489 ), 490 patch( 491 "mureo.auth_setup.list_accessible_accounts", 492 new_callable=AsyncMock, 493 return_value=mock_accounts, 494 ), 495 ): 496 result = await setup_google_ads(credentials_path=cred_path) 497 498 # login_customer_idは親MCCに設定されること 499 assert result.login_customer_id == "1111111111" 500 # customer_idは選択された子アカウントに設定されること 501 assert result.customer_id == "2222222222" 502 503 # credentials.jsonにも両方保存されていること 504 data = json.loads(cred_path.read_text(encoding="utf-8")) 505 assert data["google_ads"]["login_customer_id"] == "1111111111" 506 assert data["google_ads"]["customer_id"] == "2222222222" 507 508 509 # --------------------------------------------------------------------------- 510 # 9. OAuthフロー全体（InstalledAppFlow） 511 # --------------------------------------------------------------------------- 512 513 514 @pytest.mark.unit 515 @pytest.mark.asyncio 516 async def test_run_google_oauth() -> None: 517 """run_google_oauthがInstalledAppFlowでOAuth認証を実行すること""" 518 from mureo.auth_setup import OAuthResult, run_google_oauth 519 520 mock_credentials = MagicMock() 521 mock_credentials.refresh_token = "1//mock-refresh" 522 mock_credentials.token = "ya29.mock-access" 523 524 mock_flow = MagicMock() 525 mock_flow.run_local_server.return_value = mock_credentials 526 527 with patch( 528 "mureo.auth_setup.InstalledAppFlow.from_client_config", 529 return_value=mock_flow, 530 ) as mock_from_config: 531 result = await run_google_oauth( 532 client_id="test-cid", 533 client_secret="test-csec", 534 ) 535 536 assert isinstance(result, OAuthResult) 537 assert result.refresh_token == "1//mock-refresh" 538 assert result.access_token == "ya29.mock-access" 539 540 # InstalledAppFlowが正しいclient_configで初期化されること 541 mock_from_config.assert_called_once() 542 call_args = mock_from_config.call_args 543 client_config = call_args[0][0] 544 assert client_config["installed"]["client_id"] == "test-cid" 545 assert client_config["installed"]["client_secret"] == "test-csec" 546 assert call_args[1]["scopes"] == [ 547 "https://www.googleapis.com/auth/adwords", 548 "https://www.googleapis.com/auth/webmasters", 549 ] 550 551 # run_local_serverがport=0（自動選択）, prompt="consent"で呼ばれること 552 mock_flow.run_local_server.assert_called_once_with(port=0, prompt="consent") 553 554 555 @pytest.mark.unit 556 @pytest.mark.asyncio 557 async def test_run_google_oauth_no_refresh_token() -> None: 558 """refresh_tokenが取得できない場合にエラーが発生すること""" 559 from mureo.auth_setup import run_google_oauth 560 561 mock_credentials = MagicMock() 562 mock_credentials.refresh_token = None 563 mock_credentials.token = "ya29.mock-access" 564 565 mock_flow = MagicMock() 566 mock_flow.run_local_server.return_value = mock_credentials 567 568 with patch( 569 "mureo.auth_setup.InstalledAppFlow.from_client_config", 570 return_value=mock_flow, 571 ): 572 with pytest.raises(RuntimeError, match="Failed to obtain refresh_token"): 573 await run_google_oauth( 574 client_id="test-cid", 575 client_secret="test-csec", 576 ) 577 578 579 # --------------------------------------------------------------------------- 580 # 10. save_credentials でlogin_customer_idなし 581 # --------------------------------------------------------------------------- 582 583 584 @pytest.mark.unit 585 def test_save_credentials_without_customer_id(tmp_path: Path) -> None: 586 """customer_id未指定でもcredentials.jsonが正しく保存されること""" 587 from mureo.auth_setup import save_credentials 588 589 cred_path = tmp_path / "credentials.json" 590 591 google_creds = GoogleAdsCredentials( 592 developer_token="dev-tok", 593 client_id="cid", 594 client_secret="csec", 595 refresh_token="rtok", 596 ) 597 598 save_credentials(path=cred_path, google=google_creds) 599 600 data = json.loads(cred_path.read_text(encoding="utf-8")) 601 assert data["google_ads"]["developer_token"] == "dev-tok" 602 assert data["google_ads"].get("login_customer_id") is None 603 604 605 # --------------------------------------------------------------------------- 606 # 11. ファイルパーミッション（credentials.jsonは600） 607 # --------------------------------------------------------------------------- 608 609 610 @pytest.mark.unit 611 def test_save_credentials_file_permissions(tmp_path: Path) -> None: 612 """credentials.jsonが0600パーミッションで保存されること""" 613 import stat 614 615 from mureo.auth_setup import save_credentials 616 617 cred_path = tmp_path / "credentials.json" 618 619 google_creds = GoogleAdsCredentials( 620 developer_token="dev-tok", 621 client_id="cid", 622 client_secret="csec", 623 refresh_token="rtok", 624 ) 625 626 save_credentials(path=cred_path, google=google_creds) 627 628 file_mode = cred_path.stat().st_mode 629 # owner read/write のみ 630 assert stat.S_IMODE(file_mode) == 0o600 631 632 633 # --------------------------------------------------------------------------- 634 # 12. OAuth stateパラメータ（CSRF対策）— Meta Ads用コールバックサーバー 635 # --------------------------------------------------------------------------- 636 637 638 @pytest.mark.unit 639 def test_oauth_callback_server_validates_state() -> None: 640 """コールバックサーバーがstateパラメータを正しく検証すること""" 641 from mureo.auth_setup import OAuthCallbackServer 642 643 server = OAuthCallbackServer(port=0, expected_state="correct-state") 644 actual_port = server.server.server_address[1] 645 646 server_thread = threading.Thread(target=server.wait_for_callback, daemon=True) 647 server_thread.start() 648 649 time.sleep(0.1) 650 conn = http.client.HTTPConnection("localhost", actual_port) 651 conn.request("GET", "/callback?code=test-code&state=correct-state") 652 response = conn.getresponse() 653 conn.close() 654 655 assert response.status == 200 656 assert server.authorization_code == "test-code" 657 658 659 @pytest.mark.unit 660 def test_oauth_state_mismatch() -> None: 661 """stateパラメータが不一致の場合エラーになること""" 662 from mureo.auth_setup import OAuthCallbackServer 663 664 server = OAuthCallbackServer(port=0, expected_state="correct-state") 665 actual_port = server.server.server_address[1] 666 667 server_thread = threading.Thread(target=server.wait_for_callback, daemon=True) 668 server_thread.start() 669 670 time.sleep(0.1) 671 conn = http.client.HTTPConnection("localhost", actual_port) 672 conn.request("GET", "/callback?code=test-code&state=wrong-state") 673 response = conn.getresponse() 674 conn.close() 675 676 assert response.status == 403 677 assert server.authorization_code is None 678 assert server.error is not None 679 assert "state" in server.error.lower() 680 681 682 @pytest.mark.unit 683 def test_oauth_state_missing_in_callback() -> None: 684 """コールバックにstateパラメータが無い場合エラーになること""" 685 from mureo.auth_setup import OAuthCallbackServer 686 687 server = OAuthCallbackServer(port=0, expected_state="expected-state") 688 actual_port = server.server.server_address[1] 689 690 server_thread = threading.Thread(target=server.wait_for_callback, daemon=True) 691 server_thread.start() 692 693 time.sleep(0.1) 694 conn = http.client.HTTPConnection("localhost", actual_port) 695 conn.request("GET", "/callback?code=test-code") 696 response = conn.getresponse() 697 conn.close() 698 699 assert response.status == 403 700 assert server.authorization_code is None 701 702 703 # --------------------------------------------------------------------------- 704 # 13. XSS防止（HTMLエスケープ） 705 # --------------------------------------------------------------------------- 706 707 708 @pytest.mark.unit 709 def test_xss_prevention() -> None: 710 """コールバックサーバーのHTML出力がエスケープされていること""" 711 from mureo.auth_setup import OAuthCallbackServer 712 713 server = OAuthCallbackServer(port=0) 714 actual_port = server.server.server_address[1] 715 716 server_thread = threading.Thread(target=server.wait_for_callback, daemon=True) 717 server_thread.start() 718 719 time.sleep(0.1) 720 conn = http.client.HTTPConnection("localhost", actual_port) 721 # XSS攻撃を模したerrorパラメータ 722 conn.request( 723 "GET", 724 "/callback?error=%3Cscript%3Ealert(1)%3C/script%3E", 725 ) 726 response = conn.getresponse() 727 body = response.read().decode("utf-8") 728 conn.close() 729 730 # <script>タグがエスケープされていること 731 assert "<script>" not in body 732 assert "<script>" in body 733 734 735 # --------------------------------------------------------------------------- 736 # 14. InstalledAppFlowがrun_local_serverで例外を投げた場合 737 # --------------------------------------------------------------------------- 738 739 740 @pytest.mark.unit 741 @pytest.mark.asyncio 742 async def test_run_google_oauth_flow_exception() -> None: 743 """InstalledAppFlow.run_local_server()が例外を投げた場合にそのまま伝播すること""" 744 from mureo.auth_setup import run_google_oauth 745 746 mock_flow = MagicMock() 747 mock_flow.run_local_server.side_effect = Exception("Cannot open browser") 748 749 with patch( 750 "mureo.auth_setup.InstalledAppFlow.from_client_config", 751 return_value=mock_flow, 752 ): 753 with pytest.raises(Exception, match="Cannot open browser"): 754 await run_google_oauth( 755 client_id="test-cid", 756 client_secret="test-csec", 757 ) 758 759 760 # --------------------------------------------------------------------------- 761 # MCP設定の配置テスト 762 # --------------------------------------------------------------------------- 763 764 765 @pytest.mark.unit 766 def test_install_mcp_config_global(tmp_path: Path) -> None: 767 """グローバルMCP設定が正しく作成されること""" 768 from mureo.auth_setup import install_mcp_config 769 770 with patch("mureo.auth_setup.Path.home", return_value=tmp_path): 771 result = install_mcp_config(scope="global") 772 773 assert result is not None 774 settings = json.loads(result.read_text(encoding="utf-8")) 775 assert "mureo" in settings["mcpServers"] 776 assert settings["mcpServers"]["mureo"]["command"] == "python" 777 assert settings["mcpServers"]["mureo"]["args"] == ["-m", "mureo.mcp"] 778 779 780 @pytest.mark.unit 781 def test_install_mcp_config_project(tmp_path: Path) -> None: 782 """プロジェクトMCP設定が正しく作成されること""" 783 from mureo.auth_setup import install_mcp_config 784 785 with patch("mureo.auth_setup.Path.cwd", return_value=tmp_path): 786 result = install_mcp_config(scope="project") 787 788 assert result is not None 789 assert result.name == ".mcp.json" 790 settings = json.loads(result.read_text(encoding="utf-8")) 791 assert "mureo" in settings["mcpServers"] 792 793 794 @pytest.mark.unit 795 def test_install_mcp_config_already_exists(tmp_path: Path) -> None: 796 """既にmureoが設定済みならスキップすること""" 797 from mureo.auth_setup import install_mcp_config 798 799 # 既存設定を作成 800 claude_dir = tmp_path / ".claude" 801 claude_dir.mkdir() 802 settings_path = claude_dir / "settings.json" 803 settings_path.write_text( 804 json.dumps( 805 { 806 "mcpServers": { 807 "mureo": {"command": "python", "args": ["-m", "mureo.mcp"]} 808 } 809 } 810 ) 811 ) 812 813 with patch("mureo.auth_setup.Path.home", return_value=tmp_path): 814 result = install_mcp_config(scope="global") 815 816 assert result is None # スキップ 817 818 819 @pytest.mark.unit 820 def test_install_mcp_config_merges_existing(tmp_path: Path) -> None: 821 """既存のsettings.jsonにマージされること""" 822 from mureo.auth_setup import install_mcp_config 823 824 # 他のMCPサーバーが既に設定済み 825 claude_dir = tmp_path / ".claude" 826 claude_dir.mkdir() 827 settings_path = claude_dir / "settings.json" 828 settings_path.write_text( 829 json.dumps( 830 {"mcpServers": {"other-tool": {"command": "node", "args": ["server.js"]}}} 831 ) 832 ) 833 834 with patch("mureo.auth_setup.Path.home", return_value=tmp_path): 835 result = install_mcp_config(scope="global") 836 837 assert result is not None 838 settings = json.loads(result.read_text(encoding="utf-8")) 839 assert "other-tool" in settings["mcpServers"] # 既存は維持 840 assert "mureo" in settings["mcpServers"] # mureoが追加 841 842 843 # --------------------------------------------------------------------------- 844 # install_credential_guard テスト 845 # --------------------------------------------------------------------------- 846 847 848 @pytest.mark.unit 849 def test_install_credential_guard_new(tmp_path: Path) -> None: 850 """Credential guard hooks are added to empty settings.""" 851 from mureo.auth_setup import install_credential_guard 852 853 claude_dir = tmp_path / ".claude" 854 claude_dir.mkdir() 855 settings_path = claude_dir / "settings.json" 856 settings_path.write_text("{}") 857 858 with patch("mureo.auth_setup.Path.home", return_value=tmp_path): 859 result = install_credential_guard() 860 861 assert result is not None 862 settings = json.loads(result.read_text(encoding="utf-8")) 863 pre_tool_use = settings["hooks"]["PreToolUse"] 864 assert len(pre_tool_use) == 2 865 assert pre_tool_use[0]["matcher"] == "Read" 866 assert pre_tool_use[1]["matcher"] == "Bash" 867 assert "[mureo-credential-guard]" in pre_tool_use[0]["hooks"][0]["command"] 868 869 870 @pytest.mark.unit 871 def test_install_credential_guard_preserves_existing_hooks(tmp_path: Path) -> None: 872 """Existing hooks are preserved when adding credential guard.""" 873 from mureo.auth_setup import install_credential_guard 874 875 claude_dir = tmp_path / ".claude" 876 claude_dir.mkdir() 877 settings_path = claude_dir / "settings.json" 878 existing = { 879 "hooks": { 880 "Stop": [ 881 {"matcher": "", "hooks": [{"type": "command", "command": "echo done"}]} 882 ], 883 "PreToolUse": [ 884 { 885 "matcher": "Write", 886 "hooks": [{"type": "command", "command": "echo check"}], 887 } 888 ], 889 }, 890 "someOtherSetting": True, 891 } 892 settings_path.write_text(json.dumps(existing)) 893 894 with patch("mureo.auth_setup.Path.home", return_value=tmp_path): 895 result = install_credential_guard() 896 897 assert result is not None 898 settings = json.loads(result.read_text(encoding="utf-8")) 899 # Existing hooks preserved 900 assert "Stop" in settings["hooks"] 901 assert settings["someOtherSetting"] is True 902 # Existing PreToolUse hook preserved + 2 mureo hooks appended 903 pre_tool_use = settings["hooks"]["PreToolUse"] 904 assert len(pre_tool_use) == 3 905 assert pre_tool_use[0]["matcher"] == "Write" # original 906 assert pre_tool_use[1]["matcher"] == "Read" # mureo 907 assert pre_tool_use[2]["matcher"] == "Bash" # mureo 908 909 910 @pytest.mark.unit 911 def test_install_credential_guard_skip_if_already_installed(tmp_path: Path) -> None: 912 """Skip if mureo credential guard is already installed.""" 913 from mureo.auth_setup import install_credential_guard 914 915 claude_dir = tmp_path / ".claude" 916 claude_dir.mkdir() 917 settings_path = claude_dir / "settings.json" 918 919 # Install once 920 settings_path.write_text("{}") 921 with patch("mureo.auth_setup.Path.home", return_value=tmp_path): 922 install_credential_guard() 923 924 # Install again — should skip 925 with patch("mureo.auth_setup.Path.home", return_value=tmp_path): 926 result = install_credential_guard() 927 928 assert result is None 929 settings = json.loads(settings_path.read_text(encoding="utf-8")) 930 # Still only 2 mureo hooks (no duplicates) 931 assert len(settings["hooks"]["PreToolUse"]) == 2 932 933 934 @pytest.mark.unit 935 def test_install_credential_guard_skip_on_corrupt_json(tmp_path: Path) -> None: 936 """Skip gracefully if settings.json is corrupt.""" 937 from mureo.auth_setup import install_credential_guard 938 939 claude_dir = tmp_path / ".claude" 940 claude_dir.mkdir() 941 settings_path = claude_dir / "settings.json" 942 settings_path.write_text("{invalid json") 943 944 with patch("mureo.auth_setup.Path.home", return_value=tmp_path): 945 result = install_credential_guard() 946 947 assert result is None # skipped, not crashed 948 # File is untouched 949 assert settings_path.read_text() == "{invalid json" 950 951 952 # --------------------------------------------------------------------------- 953 # _select_account テスト 954 # --------------------------------------------------------------------------- 955 956 957 @pytest.mark.unit 958 def test_select_account_fallback_valid_choice() -> None: 959 """simple-term-menuなしで番号入力による選択（行62-76）""" 960 from mureo.auth_setup import _select_account 961 962 accounts = [ 963 {"id": "111", "name": "Account A"}, 964 {"id": "222", "name": "Account B"}, 965 ] 966 967 with patch("simple_term_menu.TerminalMenu", side_effect=ImportError): 968 with patch("builtins.input", return_value="2"): 969 result = _select_account(accounts) 970 971 assert result == "222" 972 973 974 @pytest.mark.unit 975 def test_select_account_fallback_invalid_choice() -> None: 976 """simple-term-menuなしで無効な番号入力（行75-76）""" 977 from mureo.auth_setup import _select_account 978 979 accounts = [ 980 {"id": "111", "name": "Account A"}, 981 ] 982 983 with patch("simple_term_menu.TerminalMenu", side_effect=ImportError): 984 with patch("builtins.input", return_value="999"): 985 result = _select_account(accounts) 986 987 assert result is None 988 989 990 @pytest.mark.unit 991 def test_select_account_fallback_non_numeric() -> None: 992 """simple-term-menuなしで数値以外の入力（行73）""" 993 from mureo.auth_setup import _select_account 994 995 accounts = [ 996 {"id": "111", "name": "Account A"}, 997 ] 998 999 with patch("simple_term_menu.TerminalMenu", side_effect=ImportError): 1000 with patch("builtins.input", return_value="abc"): 1001 result = _select_account(accounts) 1002 1003 assert result is None 1004 1005 1006 @pytest.mark.unit 1007 def test_select_account_terminal_menu_cancel() -> None: 1008 """TerminalMenuでキャンセル（Noneが返る）（行56-58）""" 1009 from mureo.auth_setup import _select_account 1010 1011 accounts = [ 1012 {"id": "111", "name": "Account A"}, 1013 ] 1014 1015 mock_menu = MagicMock() 1016 mock_menu.show.return_value = None # キャンセル 1017 1018 with patch("simple_term_menu.TerminalMenu", return_value=mock_menu): 1019 result = _select_account(accounts) 1020 1021 assert result is None 1022 1023 1024 @pytest.mark.unit 1025 def test_select_account_with_custom_label_fn() -> None: 1026 """label_fn指定時にカスタムラベルが使われる（行46-47）""" 1027 from mureo.auth_setup import _select_account 1028 1029 accounts = [ 1030 {"id": "111", "name": "Account A"}, 1031 ] 1032 1033 mock_menu = MagicMock() 1034 mock_menu.show.return_value = 0 1035 1036 with patch("simple_term_menu.TerminalMenu", return_value=mock_menu) as MockTM: 1037 result = _select_account(accounts, label_fn=lambda a: f"Custom: {a['name']}") 1038 1039 assert result == "111" 1040 # TerminalMenuに渡されたラベルを確認 1041 call_args = MockTM.call_args 1042 assert call_args[0][0] == ["Custom: Account A"] 1043 1044 1045 # --------------------------------------------------------------------------- 1046 # setup_mcp_config テスト 1047 # --------------------------------------------------------------------------- 1048 1049 1050 @pytest.mark.unit 1051 def test_setup_mcp_config_fallback_global() -> None: 1052 """simple-term-menuなしで番号入力でグローバル選択（行388-398）""" 1053 from mureo.auth_setup import setup_mcp_config 1054 1055 with ( 1056 patch("simple_term_menu.TerminalMenu", side_effect=ImportError), 1057 patch("mureo.auth_setup.input_func", return_value="1"), 1058 patch( 1059 "mureo.auth_setup.install_mcp_config", return_value=Path("/tmp/test") 1060 ) as mock_install, 1061 ): 1062 setup_mcp_config() 1063 1064 mock_install.assert_called_once_with(scope="global") 1065 1066 1067 @pytest.mark.unit 1068 def test_setup_mcp_config_fallback_project() -> None: 1069 """simple-term-menuなしで番号入力でプロジェクト選択""" 1070 from mureo.auth_setup import setup_mcp_config 1071 1072 with ( 1073 patch("simple_term_menu.TerminalMenu", side_effect=ImportError), 1074 patch("mureo.auth_setup.input_func", return_value="2"), 1075 patch( 1076 "mureo.auth_setup.install_mcp_config", return_value=Path("/tmp/test") 1077 ) as mock_install, 1078 ): 1079 setup_mcp_config() 1080 1081 mock_install.assert_called_once_with(scope="project") 1082 1083 1084 @pytest.mark.unit 1085 def test_setup_mcp_config_fallback_skip() -> None: 1086 """simple-term-menuなしで番号入力でスキップ（行395-396）""" 1087 from mureo.auth_setup import setup_mcp_config 1088 1089 with ( 1090 patch("simple_term_menu.TerminalMenu", side_effect=ImportError), 1091 patch("mureo.auth_setup.input_func", return_value="3"), 1092 patch("mureo.auth_setup.install_mcp_config") as mock_install, 1093 ): 1094 setup_mcp_config() 1095 1096 mock_install.assert_not_called() 1097 1098 1099 @pytest.mark.unit 1100 def test_setup_mcp_config_fallback_default() -> None: 1101 """simple-term-menuなしで空入力時はデフォルト（グローバル）（行394）""" 1102 from mureo.auth_setup import setup_mcp_config 1103 1104 with ( 1105 patch("simple_term_menu.TerminalMenu", side_effect=ImportError), 1106 patch("mureo.auth_setup.input_func", return_value=""), 1107 patch( 1108 "mureo.auth_setup.install_mcp_config", return_value=Path("/tmp/test") 1109 ) as mock_install, 1110 ): 1111 setup_mcp_config() 1112 1113 mock_install.assert_called_once_with(scope="global") 1114 1115 1116 @pytest.mark.unit 1117 def test_setup_mcp_config_terminal_menu_skip() -> None: 1118 """TerminalMenuでスキップ選択（index=2）（行383）""" 1119 from mureo.auth_setup import setup_mcp_config 1120 1121 mock_menu = MagicMock() 1122 mock_menu.show.return_value = 2 # スキップ 1123 1124 with ( 1125 patch("simple_term_menu.TerminalMenu", return_value=mock_menu), 1126 patch("mureo.auth_setup.install_mcp_config") as mock_install, 1127 ): 1128 setup_mcp_config() 1129 1130 mock_install.assert_not_called() 1131 1132 1133 @pytest.mark.unit 1134 def test_setup_mcp_config_terminal_menu_cancel() -> None: 1135 """TerminalMenuでキャンセル（None）""" 1136 from mureo.auth_setup import setup_mcp_config 1137 1138 mock_menu = MagicMock() 1139 mock_menu.show.return_value = None 1140 1141 with ( 1142 patch("simple_term_menu.TerminalMenu", return_value=mock_menu), 1143 patch("mureo.auth_setup.install_mcp_config") as mock_install, 1144 ): 1145 setup_mcp_config() 1146 1147 mock_install.assert_not_called() 1148 1149 1150 @pytest.mark.unit 1151 def test_setup_mcp_config_already_exists() -> None: 1152 """MCP設定が既に存在する場合（行403-404）""" 1153 from mureo.auth_setup import setup_mcp_config 1154 1155 mock_menu = MagicMock() 1156 mock_menu.show.return_value = 0 # グローバル 1157 1158 with ( 1159 patch("simple_term_menu.TerminalMenu", return_value=mock_menu), 1160 patch("mureo.auth_setup.install_mcp_config", return_value=None) as mock_install, 1161 ): 1162 setup_mcp_config() 1163 1164 mock_install.assert_called_once() 1165 1166 1167 # --------------------------------------------------------------------------- 1168 # OAuthCallbackServer 追加テスト 1169 # --------------------------------------------------------------------------- 1170 1171 1172 @pytest.mark.unit 1173 def test_oauth_callback_server_invalid_request() -> None: 1174 """不正なリクエスト（codeもerrorもない）の場合""" 1175 from mureo.auth_setup import OAuthCallbackServer 1176 1177 server = OAuthCallbackServer(port=0) 1178 actual_port = server.server.server_address[1] 1179 1180 server_thread = threading.Thread(target=server.wait_for_callback, daemon=True) 1181 server_thread.start() 1182 1183 time.sleep(0.1) 1184 conn = http.client.HTTPConnection("localhost", actual_port) 1185 conn.request("GET", "/callback?unexpected=value") 1186 response = conn.getresponse() 1187 conn.close() 1188 1189 assert response.status == 400 1190 assert server.authorization_code is None 1191 assert server.error is None 1192 1193 1194 @pytest.mark.unit 1195 def test_oauth_callback_server_shutdown() -> None: 1196 """shutdownメソッドが正常に動作する""" 1197 from mureo.auth_setup import OAuthCallbackServer 1198 1199 server = OAuthCallbackServer(port=0) 1200 server.shutdown() # 例外なく完了すること 1201 1202 1203 # --------------------------------------------------------------------------- 1204 # save_credentials Meta Ads テスト 1205 # --------------------------------------------------------------------------- 1206 1207 1208 @pytest.mark.unit 1209 def test_save_credentials_meta_ads(tmp_path: Path) -> None: 1210 """Meta Ads認証情報が正しく保存されること""" 1211 from mureo.auth import MetaAdsCredentials 1212 from mureo.auth_setup import save_credentials 1213 1214 cred_path = tmp_path / "credentials.json" 1215 1216 meta_creds = MetaAdsCredentials( 1217 access_token="meta-access-tok", 1218 app_id="app-123", 1219 app_secret="app-secret-456", 1220 ) 1221 1222 save_credentials( 1223 path=cred_path, 1224 meta=meta_creds, 1225 account_id="act_789", 1226 ) 1227 1228 data = json.loads(cred_path.read_text(encoding="utf-8")) 1229 assert data["meta_ads"]["access_token"] == "meta-access-tok" 1230 assert data["meta_ads"]["app_id"] == "app-123" 1231 assert data["meta_ads"]["app_secret"] == "app-secret-456" 1232 assert data["meta_ads"]["account_id"] == "act_789" 1233 1234 1235 @pytest.mark.unit 1236 def test_save_credentials_corrupted_existing(tmp_path: Path) -> None: 1237 """既存のcredentials.jsonが壊れている場合でも新規作成される""" 1238 from mureo.auth_setup import save_credentials 1239 1240 cred_path = tmp_path / "credentials.json" 1241 cred_path.write_text("not valid json", encoding="utf-8") 1242 1243 google_creds = GoogleAdsCredentials( 1244 developer_token="dev-tok", 1245 client_id="cid", 1246 client_secret="csec", 1247 refresh_token="rtok", 1248 ) 1249 1250 save_credentials(path=cred_path, google=google_creds) 1251 1252 data = json.loads(cred_path.read_text(encoding="utf-8")) 1253 assert data["google_ads"]["developer_token"] == "dev-tok" 1254 1255 1256 # --------------------------------------------------------------------------- 1257 # Meta Ads OAuth テスト 1258 # --------------------------------------------------------------------------- 1259 1260 1261 @pytest.mark.unit 1262 def test_generate_meta_auth_url() -> None: 1263 """Meta OAuth認証URLが正しく生成されること""" 1264 from mureo.auth_setup import _generate_meta_auth_url 1265 1266 url = _generate_meta_auth_url(app_id="test-app-id", port=8888, state="abc123") 1267 1268 assert "test-app-id" in url 1269 assert "localhost%3A8888" in url or "localhost:8888" in url 1270 assert "state=abc123" in url 1271 assert "ads_management" in url 1272 1273 1274 @pytest.mark.unit 1275 def test_generate_meta_auth_url_no_state() -> None: 1276 """state=Noneの場合はstateパラメータが含まれない""" 1277 from mureo.auth_setup import _generate_meta_auth_url 1278 1279 url = _generate_meta_auth_url(app_id="test-app-id", port=8888, state=None) 1280 1281 assert "state=" not in url 1282 1283 1284 @pytest.mark.unit 1285 @pytest.mark.asyncio 1286 async def test_exchange_code_for_short_token() -> None: 1287 """Short-Lived Tokenの取得""" 1288 from mureo.auth_setup import _exchange_code_for_short_token 1289 1290 mock_response = MagicMock() 1291 mock_response.json.return_value = {"access_token": "short-lived-tok"} 1292 mock_response.raise_for_status = MagicMock() 1293 1294 with patch("httpx.AsyncClient") as MockClient: 1295 instance = AsyncMock() 1296 instance.get = AsyncMock(return_value=mock_response) 1297 instance.__aenter__ = AsyncMock(return_value=instance) 1298 instance.__aexit__ = AsyncMock(return_value=False) 1299 MockClient.return_value = instance 1300 1301 result = await _exchange_code_for_short_token( 1302 code="test-code", 1303 app_id="test-app", 1304 app_secret="test-secret", 1305 redirect_uri="http://localhost:8888/callback", 1306 ) 1307 1308 assert result == "short-lived-tok" 1309 1310 1311 @pytest.mark.unit 1312 @pytest.mark.asyncio 1313 async def test_exchange_code_for_short_token_error() -> None: 1314 """Short-Lived Token取得失敗時にRuntimeError""" 1315 from mureo.auth_setup import _exchange_code_for_short_token 1316 1317 with patch("httpx.AsyncClient") as MockClient: 1318 instance = AsyncMock() 1319 instance.get = AsyncMock(side_effect=RuntimeError("connection error")) 1320 instance.__aenter__ = AsyncMock(return_value=instance) 1321 instance.__aexit__ = AsyncMock(return_value=False) 1322 MockClient.return_value = instance 1323 1324 with pytest.raises(RuntimeError, match="Short-Lived Token"): 1325 await _exchange_code_for_short_token( 1326 code="test-code", 1327 app_id="test-app", 1328 app_secret="test-secret", 1329 redirect_uri="http://localhost:8888/callback", 1330 ) 1331 1332 1333 @pytest.mark.unit 1334 @pytest.mark.asyncio 1335 async def test_exchange_short_for_long_token() -> None: 1336 """Long-Lived Tokenへの変換""" 1337 from mureo.auth_setup import MetaOAuthResult, _exchange_short_for_long_token 1338 1339 mock_response = MagicMock() 1340 mock_response.json.return_value = { 1341 "access_token": "long-lived-tok", 1342 "expires_in": 5184000, 1343 } 1344 mock_response.raise_for_status = MagicMock() 1345 1346 with patch("httpx.AsyncClient") as MockClient: 1347 instance = AsyncMock() 1348 instance.get = AsyncMock(return_value=mock_response) 1349 instance.__aenter__ = AsyncMock(return_value=instance) 1350 instance.__aexit__ = AsyncMock(return_value=False) 1351 MockClient.return_value = instance 1352 1353 result = await _exchange_short_for_long_token( 1354 short_token="short-tok", 1355 app_id="test-app", 1356 app_secret="test-secret", 1357 ) 1358 1359 assert isinstance(result, MetaOAuthResult) 1360 assert result.access_token == "long-lived-tok" 1361 assert result.expires_in == 5184000 1362 1363 1364 @pytest.mark.unit 1365 @pytest.mark.asyncio 1366 async def test_exchange_short_for_long_token_error() -> None: 1367 """Long-Lived Token変換失敗時にRuntimeError""" 1368 from mureo.auth_setup import _exchange_short_for_long_token 1369 1370 with patch("httpx.AsyncClient") as MockClient: 1371 instance = AsyncMock() 1372 instance.get = AsyncMock(side_effect=RuntimeError("api error")) 1373 instance.__aenter__ = AsyncMock(return_value=instance) 1374 instance.__aexit__ = AsyncMock(return_value=False) 1375 MockClient.return_value = instance 1376 1377 with pytest.raises(RuntimeError, match="Long-Lived Token"): 1378 await _exchange_short_for_long_token( 1379 short_token="short-tok", 1380 app_id="test-app", 1381 app_secret="test-secret", 1382 ) 1383 1384 1385 @pytest.mark.unit 1386 @pytest.mark.asyncio 1387 async def test_list_meta_ad_accounts() -> None: 1388 """Meta広告アカウント一覧取得""" 1389 from mureo.auth_setup import list_meta_ad_accounts 1390 1391 mock_response = MagicMock() 1392 mock_response.json.return_value = { 1393 "data": [ 1394 {"id": "act_111", "name": "Test Account", "account_status": 1}, 1395 ] 1396 } 1397 mock_response.raise_for_status = MagicMock() 1398 1399 with patch("httpx.AsyncClient") as MockClient: 1400 instance = AsyncMock() 1401 instance.get = AsyncMock(return_value=mock_response) 1402 instance.__aenter__ = AsyncMock(return_value=instance) 1403 instance.__aexit__ = AsyncMock(return_value=False) 1404 MockClient.return_value = instance 1405 1406 accounts = await list_meta_ad_accounts("test-token") 1407 1408 assert len(accounts) == 1 1409 assert accounts[0]["id"] == "act_111" 1410 1411 1412 @pytest.mark.unit 1413 @pytest.mark.asyncio 1414 async def test_list_meta_ad_accounts_error() -> None: 1415 """Meta広告アカウント一覧取得失敗時にRuntimeError""" 1416 from mureo.auth_setup import list_meta_ad_accounts 1417 1418 with patch("httpx.AsyncClient") as MockClient: 1419 instance = AsyncMock() 1420 instance.get = AsyncMock(side_effect=RuntimeError("api error")) 1421 instance.__aenter__ = AsyncMock(return_value=instance) 1422 instance.__aexit__ = AsyncMock(return_value=False) 1423 MockClient.return_value = instance 1424 1425 with pytest.raises(RuntimeError, match="Failed to retrieve ad account list"): 1426 await list_meta_ad_accounts("test-token") 1427 1428 1429 @pytest.mark.unit 1430 @pytest.mark.asyncio 1431 async def test_run_meta_oauth() -> None: 1432 """Meta OAuthフロー全体""" 1433 from mureo.auth_setup import MetaOAuthResult, run_meta_oauth 1434 1435 with ( 1436 patch("mureo.auth_setup.OAuthCallbackServer") as MockServer, 1437 patch("mureo.auth_setup.webbrowser.open"), 1438 patch( 1439 "mureo.auth_setup._exchange_code_for_short_token", 1440 new_callable=AsyncMock, 1441 return_value="short-token", 1442 ), 1443 patch( 1444 "mureo.auth_setup._exchange_short_for_long_token", 1445 new_callable=AsyncMock, 1446 return_value=MetaOAuthResult( 1447 access_token="long-lived-tok", expires_in=5184000 1448 ), 1449 ), 1450 ): 1451 server_instance = MockServer.return_value 1452 server_instance.server.server_address = ("localhost", 9999) 1453 server_instance.error = None 1454 server_instance.authorization_code = "test-auth-code" 1455 # wait_for_callbackは即座に返る 1456 server_instance.wait_for_callback = MagicMock() 1457 1458 result = await run_meta_oauth( 1459 app_id="test-app-id", 1460 app_secret="test-app-secret", 1461 ) 1462 1463 assert isinstance(result, MetaOAuthResult) 1464 assert result.access_token == "long-lived-tok" 1465 1466 1467 @pytest.mark.unit 1468 @pytest.mark.asyncio 1469 async def test_run_meta_oauth_error() -> None: 1470 """Meta OAuthフローでエラーが返された場合""" 1471 from mureo.auth_setup import run_meta_oauth 1472 1473 with ( 1474 patch("mureo.auth_setup.OAuthCallbackServer") as MockServer, 1475 patch("mureo.auth_setup.webbrowser.open"), 1476 ): 1477 server_instance = MockServer.return_value 1478 server_instance.server.server_address = ("localhost", 9999) 1479 server_instance.error = "access_denied" 1480 server_instance.authorization_code = None 1481 server_instance.wait_for_callback = MagicMock() 1482 1483 with pytest.raises(RuntimeError, match="access_denied"): 1484 await run_meta_oauth( 1485 app_id="test-app-id", 1486 app_secret="test-app-secret", 1487 ) 1488 1489 1490 @pytest.mark.unit 1491 @pytest.mark.asyncio 1492 async def test_run_meta_oauth_timeout() -> None: 1493 """Meta OAuthフローでタイムアウトした場合""" 1494 from mureo.auth_setup import run_meta_oauth 1495 1496 with ( 1497 patch("mureo.auth_setup.OAuthCallbackServer") as MockServer, 1498 patch("mureo.auth_setup.webbrowser.open"), 1499 ): 1500 server_instance = MockServer.return_value 1501 server_instance.server.server_address = ("localhost", 9999) 1502 server_instance.error = None 1503 server_instance.authorization_code = None # タイムアウト 1504 server_instance.wait_for_callback = MagicMock() 1505 1506 with pytest.raises(RuntimeError, match="Authentication timed out"): 1507 await run_meta_oauth( 1508 app_id="test-app-id", 1509 app_secret="test-app-secret", 1510 ) 1511 1512 1513 @pytest.mark.unit 1514 @pytest.mark.asyncio 1515 async def test_setup_meta_ads_flow(tmp_path: Path) -> None: 1516 """Meta Adsセットアップの全体フロー""" 1517 from mureo.auth_setup import MetaOAuthResult, setup_meta_ads 1518 1519 cred_path = tmp_path / "credentials.json" 1520 1521 user_inputs = iter(["test-app-id", "test-app-secret"]) 1522 1523 mock_oauth_result = MetaOAuthResult( 1524 access_token="long-lived-token", 1525 expires_in=5184000, 1526 ) 1527 1528 mock_accounts = [ 1529 {"id": "act_111", "name": "Account 1", "account_status": 1}, 1530 {"id": "act_222", "name": "Account 2", "account_status": 1}, 1531 ] 1532 1533 with ( 1534 patch("mureo.auth_setup.input_func", side_effect=user_inputs), 1535 patch( 1536 "mureo.auth_setup.run_meta_oauth", 1537 new_callable=AsyncMock, 1538 return_value=mock_oauth_result, 1539 ), 1540 patch( 1541 "mureo.auth_setup.list_meta_ad_accounts", 1542 new_callable=AsyncMock, 1543 return_value=mock_accounts, 1544 ), 1545 patch("mureo.auth_setup._select_account", return_value="act_222"), 1546 ): 1547 result = await setup_meta_ads(credentials_path=cred_path) 1548 1549 assert result.access_token == "long-lived-token" 1550 assert result.app_id == "test-app-id" 1551 1552 data = json.loads(cred_path.read_text(encoding="utf-8")) 1553 assert data["meta_ads"]["access_token"] == "long-lived-token" 1554 assert data["meta_ads"]["account_id"] == "act_222" 1555 1556 1557 @pytest.mark.unit 1558 @pytest.mark.asyncio 1559 async def test_setup_meta_ads_single_account(tmp_path: Path) -> None: 1560 """1アカウントのみの場合は自動選択""" 1561 from mureo.auth_setup import MetaOAuthResult, setup_meta_ads 1562 1563 cred_path = tmp_path / "credentials.json" 1564 1565 user_inputs = iter(["test-app-id", "test-app-secret"]) 1566 1567 mock_oauth_result = MetaOAuthResult( 1568 access_token="long-lived-token", 1569 expires_in=5184000, 1570 ) 1571 1572 mock_accounts = [ 1573 {"id": "act_111", "name": "Single Account", "account_status": 1}, 1574 ] 1575 1576 with ( 1577 patch("mureo.auth_setup.input_func", side_effect=user_inputs), 1578 patch( 1579 "mureo.auth_setup.run_meta_oauth", 1580 new_callable=AsyncMock, 1581 return_value=mock_oauth_result, 1582 ), 1583 patch( 1584 "mureo.auth_setup.list_meta_ad_accounts", 1585 new_callable=AsyncMock, 1586 return_value=mock_accounts, 1587 ), 1588 ): 1589 result = await setup_meta_ads(credentials_path=cred_path) 1590 1591 data = json.loads(cred_path.read_text(encoding="utf-8")) 1592 assert data["meta_ads"]["account_id"] == "act_111" 1593 1594 1595 @pytest.mark.unit 1596 @pytest.mark.asyncio 1597 async def test_setup_meta_ads_no_accounts(tmp_path: Path) -> None: 1598 """アカウントが見つからない場合はRuntimeError""" 1599 from mureo.auth_setup import MetaOAuthResult, setup_meta_ads 1600 1601 cred_path = tmp_path / "credentials.json" 1602 1603 user_inputs = iter(["test-app-id", "test-app-secret"]) 1604 1605 mock_oauth_result = MetaOAuthResult( 1606 access_token="long-lived-token", 1607 expires_in=5184000, 1608 ) 1609 1610 with ( 1611 patch("mureo.auth_setup.input_func", side_effect=user_inputs), 1612 patch( 1613 "mureo.auth_setup.run_meta_oauth", 1614 new_callable=AsyncMock, 1615 return_value=mock_oauth_result, 1616 ), 1617 patch( 1618 "mureo.auth_setup.list_meta_ad_accounts", 1619 new_callable=AsyncMock, 1620 return_value=[], 1621 ), 1622 ): 1623 with pytest.raises(RuntimeError, match="No ad accounts found"): 1624 await setup_meta_ads(credentials_path=cred_path) 1625 1626 1627 @pytest.mark.unit 1628 @pytest.mark.asyncio 1629 async def test_setup_meta_ads_cancel_selection(tmp_path: Path) -> None: 1630 """アカウント選択をキャンセルした場合はデフォルトアカウントを使用""" 1631 from mureo.auth_setup import MetaOAuthResult, setup_meta_ads 1632 1633 cred_path = tmp_path / "credentials.json" 1634 1635 user_inputs = iter(["test-app-id", "test-app-secret"]) 1636 1637 mock_oauth_result = MetaOAuthResult( 1638 access_token="long-lived-token", 1639 expires_in=5184000, 1640 ) 1641 1642 mock_accounts = [ 1643 {"id": "act_111", "name": "Account 1", "account_status": 1}, 1644 {"id": "act_222", "name": "Account 2", "account_status": 1}, 1645 ] 1646 1647 with ( 1648 patch("mureo.auth_setup.input_func", side_effect=user_inputs), 1649 patch( 1650 "mureo.auth_setup.run_meta_oauth", 1651 new_callable=AsyncMock, 1652 return_value=mock_oauth_result, 1653 ), 1654 patch( 1655 "mureo.auth_setup.list_meta_ad_accounts", 1656 new_callable=AsyncMock, 1657 return_value=mock_accounts, 1658 ), 1659 patch("mureo.auth_setup._select_account", return_value=None), 1660 ): 1661 result = await setup_meta_ads(credentials_path=cred_path) 1662 1663 data = json.loads(cred_path.read_text(encoding="utf-8")) 1664 # キャンセル時は最初のアカウントがデフォルト 1665 assert data["meta_ads"]["account_id"] == "act_111" 1666 1667 1668 @pytest.mark.unit 1669 def test_resolve_default_path() -> None: 1670 """デフォルトパスの解決""" 1671 from mureo.auth_setup import _resolve_default_path 1672 1673 path = _resolve_default_path() 1674 assert path.name == "credentials.json" 1675 assert ".mureo" in str(path)