test_mcp_server.py
1 """MCP サーバーテスト 2 3 MCPサーバーの list_tools / call_tool の動作を検証する。 4 GoogleAdsApiClient / MetaAdsApiClient はモックし、stdio 層は介さずサーバー関数を直接呼ぶ。 5 """ 6 7 from __future__ import annotations 8 9 import json 10 from typing import Any 11 from unittest.mock import AsyncMock, MagicMock, patch 12 13 import pytest 14 15 16 # --------------------------------------------------------------------------- 17 # ヘルパー 18 # --------------------------------------------------------------------------- 19 20 21 def _import_server_module(): 22 """mureo/mcp/server.py をインポートする""" 23 from mureo.mcp import server as mcp_server_module 24 25 return mcp_server_module 26 27 28 def _import_google_ads_handlers(): 29 from mureo.mcp import _handlers_google_ads 30 31 return _handlers_google_ads 32 33 34 # --------------------------------------------------------------------------- 35 # list_tools テスト 36 # --------------------------------------------------------------------------- 37 38 39 @pytest.mark.unit 40 class TestListTools: 41 """list_tools が正しいツール定義を返すことを検証する""" 42 43 async def test_list_tools_returns_all_tools(self) -> None: 44 """list_tools は全ツール（Google Ads 83 + Meta Ads 77 + Search Console 10 45 + Rollback 2 + Analysis 1 = 173）を返す""" 46 mod = _import_server_module() 47 tools = await mod.handle_list_tools() 48 assert len(tools) == 173 49 50 async def test_list_tools_contains_google_and_meta(self) -> None: 51 """Google Ads と Meta Ads のツールが含まれること""" 52 mod = _import_server_module() 53 tools = await mod.handle_list_tools() 54 names = {t.name for t in tools} 55 assert "google_ads.campaigns.list" in names 56 assert "google_ads.campaigns.get" in names 57 assert "meta_ads.campaigns.list" in names 58 assert "meta_ads.campaigns.get" in names 59 60 async def test_list_tools_campaigns_list_schema(self) -> None: 61 """campaigns.list の inputSchema が customer_id を optional で持つこと""" 62 mod = _import_server_module() 63 tools = await mod.handle_list_tools() 64 tool = next(t for t in tools if t.name == "google_ads.campaigns.list") 65 schema = tool.inputSchema 66 assert schema["type"] == "object" 67 assert "customer_id" in schema["properties"] 68 # customer_id is optional (falls back to credentials.json) 69 assert "customer_id" not in schema.get("required", []) 70 71 async def test_list_tools_campaigns_get_schema(self) -> None: 72 """campaigns.get の inputSchema が campaign_id を required で持つこと""" 73 mod = _import_server_module() 74 tools = await mod.handle_list_tools() 75 tool = next(t for t in tools if t.name == "google_ads.campaigns.get") 76 schema = tool.inputSchema 77 assert schema["type"] == "object" 78 assert "customer_id" in schema["properties"] 79 assert "campaign_id" in schema["properties"] 80 # customer_id is optional, campaign_id is required 81 assert "campaign_id" in schema["required"] 82 83 84 # --------------------------------------------------------------------------- 85 # call_tool テスト 86 # --------------------------------------------------------------------------- 87 88 89 @pytest.mark.unit 90 class TestCallToolCampaignsList: 91 """call_tool で google_ads.campaigns.list を呼ぶテスト""" 92 93 async def test_campaigns_list_calls_client(self) -> None: 94 """campaigns.list が GoogleAdsApiClient.list_campaigns を呼ぶこと""" 95 mod = _import_server_module() 96 ga_mod = _import_google_ads_handlers() 97 98 mock_client = AsyncMock() 99 mock_client.list_campaigns.return_value = [ 100 {"id": "123", "name": "Campaign 1", "status": "ENABLED"} 101 ] 102 103 mock_creds = MagicMock() 104 with ( 105 patch.object( 106 ga_mod, "load_google_ads_credentials", return_value=mock_creds 107 ), 108 patch.object(ga_mod, "create_google_ads_client", return_value=mock_client), 109 ): 110 result = await mod.handle_call_tool( 111 "google_ads.campaigns.list", 112 {"customer_id": "1234567890"}, 113 ) 114 115 mock_client.list_campaigns.assert_awaited_once() 116 assert len(result) == 1 117 assert result[0].type == "text" 118 parsed = json.loads(result[0].text) 119 assert len(parsed) == 1 120 assert parsed[0]["id"] == "123" 121 122 123 @pytest.mark.unit 124 class TestCallToolCampaignsGet: 125 """call_tool で google_ads.campaigns.get を呼ぶテスト""" 126 127 async def test_campaigns_get_calls_client(self) -> None: 128 """campaigns.get が GoogleAdsApiClient.get_campaign を呼ぶこと""" 129 mod = _import_server_module() 130 ga_mod = _import_google_ads_handlers() 131 132 mock_client = AsyncMock() 133 mock_client.get_campaign.return_value = { 134 "id": "456", 135 "name": "Campaign Detail", 136 "status": "PAUSED", 137 "budget_daily": 5000.0, 138 } 139 140 mock_creds = MagicMock() 141 with ( 142 patch.object( 143 ga_mod, "load_google_ads_credentials", return_value=mock_creds 144 ), 145 patch.object(ga_mod, "create_google_ads_client", return_value=mock_client), 146 ): 147 result = await mod.handle_call_tool( 148 "google_ads.campaigns.get", 149 {"customer_id": "1234567890", "campaign_id": "456"}, 150 ) 151 152 mock_client.get_campaign.assert_awaited_once_with("456") 153 assert len(result) == 1 154 parsed = json.loads(result[0].text) 155 assert parsed["id"] == "456" 156 assert parsed["budget_daily"] == 5000.0 157 158 159 @pytest.mark.unit 160 class TestCallToolErrors: 161 """call_tool のエラーケースを検証する""" 162 163 async def test_unknown_tool_raises_error(self) -> None: 164 """未知のツール名で ValueError が発生すること""" 165 mod = _import_server_module() 166 167 with pytest.raises(ValueError, match="Unknown tool"): 168 await mod.handle_call_tool("nonexistent.tool", {}) 169 170 async def test_missing_required_param_customer_id(self) -> None: 171 """campaigns.list で customer_id が欠損 + credentials.jsonにもない場合""" 172 mod = _import_server_module() 173 174 with patch( 175 "mureo.mcp._handlers_google_ads.load_google_ads_credentials", 176 return_value=None, 177 ): 178 result = await mod.handle_call_tool("google_ads.campaigns.list", {}) 179 assert any("Credentials not found" in r.text for r in result) 180 181 async def test_missing_required_param_campaign_id(self) -> None: 182 """campaigns.get で campaign_id が欠損した場合にエラーになること""" 183 mod = _import_server_module() 184 ga_mod = _import_google_ads_handlers() 185 186 mock_creds = MagicMock() 187 with patch.object( 188 ga_mod, "load_google_ads_credentials", return_value=mock_creds 189 ): 190 with pytest.raises(ValueError, match="campaign_id"): 191 await mod.handle_call_tool( 192 "google_ads.campaigns.get", 193 {"customer_id": "1234567890"}, 194 ) 195 196 async def test_no_credentials_returns_error_text(self) -> None: 197 """認証情報がない場合、エラーメッセージを TextContent で返すこと""" 198 ga_mod = _import_google_ads_handlers() 199 mod = _import_server_module() 200 201 with patch.object( 202 ga_mod, 203 "load_google_ads_credentials", 204 return_value=None, 205 ): 206 result = await mod.handle_call_tool( 207 "google_ads.campaigns.list", 208 {"customer_id": "1234567890"}, 209 ) 210 211 assert len(result) == 1 212 assert result[0].type == "text" 213 assert "Credentials not found" in result[0].text