diff --git a/src/ian/gateways/mcp_server.py b/src/ian/gateways/mcp_server.py index c20ec9f..025d0e2 100644 --- a/src/ian/gateways/mcp_server.py +++ b/src/ian/gateways/mcp_server.py @@ -85,18 +85,19 @@ def check_user_permission( return True, role return False, role -try: - rag.initialize_rag_system() - course_catalog.load_course_data_from_url(COURSE_DATA_URL) -except Exception as e: - eprint(f"初始化錯誤: {e}") +def initialize_dependencies() -> None: + """Initialize external data sources when the MCP server starts.""" + try: + rag.initialize_rag_system() + course_catalog.load_course_data_from_url(COURSE_DATA_URL) + except Exception as e: + eprint(f"初始化錯誤: {e}") -# Initialize member database -try: - init_member_db() - eprint("社員資料庫已初始化") -except Exception as e: - eprint(f"社員資料庫初始化失敗: {e}") + try: + init_member_db() + eprint("社員資料庫已初始化") + except Exception as e: + eprint(f"社員資料庫初始化失敗: {e}") # --------------------------------------------------------------------------- @@ -564,6 +565,8 @@ async def notify_members(role: str, event_date: str = "", note: str = "", custom def entrypoint(http: bool = False, host: str = "0.0.0.0", port: int = 5191): + initialize_dependencies() + if http: # Use FastMCP's built-in streamable-http transport (stateless mode) # This avoids the SSE session leak in mcp/server/sse.py where diff --git a/tests/gateways/test_mcp_server.py b/tests/gateways/test_mcp_server.py new file mode 100644 index 0000000..911e742 --- /dev/null +++ b/tests/gateways/test_mcp_server.py @@ -0,0 +1,260 @@ +# +# SPDX-License-Identifier: GPL-3.0-or-later +# +# Copyright (c) 2026 NTU AI Club +# +# This file is part of Ian, an open-source AI agent framework developed +# and maintained by NTU AI Club. +# +# Ian is licensed under the GNU General Public License, either version 3 +# of the License, or (at your option) any later version. +# +# Ian is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. +# See the GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Ian. If not, see . +# + +import asyncio + +import pytest + +from ian.gateways import mcp_server + + +def _run(coro): + return asyncio.run(coro) + + +@pytest.mark.parametrize( + ("channel_id", "resolved_role", "expected"), + [ + pytest.param("allowed", "非社員", (True, "非社員"), id="allowed-channel"), + pytest.param("", "VIP 社員", (True, "VIP 社員"), id="valid-member"), + pytest.param("", "非社員", (False, "非社員"), id="non-member"), + pytest.param( + "", + "非社員(會籍已過期)", + (False, "非社員(會籍已過期)"), + id="expired-member", + ), + ], +) +def test_check_user_permission_uses_allowed_channels_and_member_role( + monkeypatch, channel_id, resolved_role, expected +): + monkeypatch.setattr(mcp_server, "ALLOWED_CHANNELS", {"allowed"}) + monkeypatch.setattr(mcp_server, "get_member_role", lambda *_: resolved_role) + + assert ( + mcp_server.check_user_permission("Discord", "account-1", channel_id) == expected + ) + + +@pytest.mark.parametrize( + ("member", "name", "email", "expected_parts"), + [ + pytest.param( + {"name": "王 小明", "email": "member+test@example.test"}, + "ignored", + "ignored@example.test", + ( + "已為社員「王 小明」", + "name=%E7%8E%8B%20%E5%B0%8F%E6%98%8E", + "id=member%2Btest%40example.test", + ), + id="bound-member", + ), + pytest.param( + None, + "", + "", + ("請提供您的「姓名」和「Email」",), + id="missing-non-member-info", + ), + pytest.param( + None, "Visitor", "invalid", ("請提供有效的 Email",), id="invalid-email" + ), + pytest.param( + None, + "Guest User", + "guest+event@example.test", + ( + "已為「Guest User」", + "name=Guest%20User", + "id=guest%2Bevent%40example.test", + "不代表已成功報名", + ), + id="non-member-link", + ), + ], +) +def test_generate_checkin_code_handles_member_and_guest_flows( + monkeypatch, member, name, email, expected_parts +): + monkeypatch.setattr(mcp_server, "lookup_member_by_platform", lambda *_: member) + + result = _run(mcp_server.generate_checkin_code("Discord", "account-1", name, email)) + + assert all(part in result for part in expected_parts) + + +@pytest.mark.parametrize( + ("tool_name", "dependency_name", "args", "error_prefix"), + [ + pytest.param( + "bind_email", + "_bind_email_to_platform", + ("member@example.test", "Discord", "account-1"), + "⚠️ 綁定時發生錯誤", + id="bind-email", + ), + pytest.param( + "update_subscribe", + "_update_subscribe", + ("Discord", "account-1", "discord"), + "⚠️ 更新訂閱設定時發生錯誤", + id="update-subscribe", + ), + pytest.param( + "update_personal_prompt", + "_update_personal_prompt", + ("Discord", "account-1", "concise"), + "⚠️ 更新個性備註時發生錯誤", + id="update-personal-prompt", + ), + ], +) +def test_member_tool_wrappers_return_messages_and_handle_exceptions( + monkeypatch, tool_name, dependency_name, args, error_prefix +): + tool = getattr(mcp_server, tool_name) + monkeypatch.setattr( + mcp_server, dependency_name, lambda *_: {"message": "service message"} + ) + assert _run(tool(*args)) == "service message" + + def fail(*_args): + raise RuntimeError("service unavailable") + + monkeypatch.setattr(mcp_server, dependency_name, fail) + assert _run(tool(*args)) == f"{error_prefix}:service unavailable" + + +def test_notify_members_rejects_non_staff_before_loading_data(monkeypatch): + monkeypatch.setattr(mcp_server.notifications, "is_staff_role", lambda _role: False) + monkeypatch.setattr( + mcp_server, + "_get_upcoming_events", + lambda *_: (_ for _ in ()).throw(AssertionError("course data should not load")), + ) + + result = _run(mcp_server.notify_members("一般社員")) + + assert "此功能僅限幹部使用" in result + + +def _stub_staff(monkeypatch): + monkeypatch.setattr(mcp_server.notifications, "is_staff_role", lambda _role: True) + + +def test_notify_members_sends_custom_notification(monkeypatch): + _stub_staff(monkeypatch) + delivery = {"total_members": 2, "discord_ok": 1, "discord_fail": 1} + sent = [] + logs = [] + monkeypatch.setattr("ian.services.reminder_runner.load_members", lambda: ["member"]) + monkeypatch.setattr( + mcp_server.notifications, + "send_notification_to_members", + lambda message, members: sent.append((message, members)) or delivery, + ) + monkeypatch.setattr( + mcp_server.notifications, + "send_discord_channel_message", + lambda channel, message: logs.append((channel, message)) or True, + ) + + result = _run(mcp_server.notify_members("部員", custom_message=" Custom alert ")) + + assert sent == [("NTUAI 通知\n\nCustom alert", ["member"])] + assert "通知對象: 2" in result + assert "Discord: 1 成功, 1 失敗" in result + assert len(logs) == 1 + + +def test_notify_members_reports_missing_event_without_sending(monkeypatch): + _stub_staff(monkeypatch) + monkeypatch.setattr(mcp_server, "_find_event_by_date", lambda _date: None) + monkeypatch.setattr( + mcp_server.notifications, + "send_notification_to_members", + lambda *_: (_ for _ in ()).throw( + AssertionError("notification should not send") + ), + ) + + result = _run(mcp_server.notify_members("部長", event_date="2026/08/01")) + + assert result == "找不到日期為 2026/08/01 的活動,請確認日期格式為 YYYY/MM/DD。" + + +def test_notify_members_sends_formatted_event_notification(monkeypatch): + _stub_staff(monkeypatch) + event = {"title": "Agent Evaluation", "date": "2026/08/01"} + delivery = {"total_members": 3, "discord_ok": 3, "discord_fail": 0} + monkeypatch.setattr(mcp_server, "_find_event_by_date", lambda _date: event) + monkeypatch.setattr( + "ian.services.reminder_runner.load_members", lambda: ["members"] + ) + monkeypatch.setattr( + mcp_server.notifications, + "format_staff_notification", + lambda value, note: f"formatted:{value['title']}:{note}", + ) + monkeypatch.setattr( + mcp_server.notifications, + "send_notification_to_members", + lambda message, members: delivery, + ) + monkeypatch.setattr( + mcp_server.notifications, "send_discord_channel_message", lambda *_: True + ) + + result = _run( + mcp_server.notify_members("社長", event_date=" 2026/08/01 ", note=" reminder ") + ) + + assert "活動: Agent Evaluation (2026/08/01)" in result + assert "Discord: 3 成功, 0 失敗" in result + + +@pytest.mark.parametrize( + ("upcoming", "expected_parts"), + [ + pytest.param([], ("目前沒有即將舉辦的活動",), id="no-upcoming-events"), + pytest.param( + [ + { + "title": "Agent Evaluation", + "date": "2026/08/01", + "weekday": "六", + "time": "19:00", + "venue": "新生", + } + ], + ("1. Agent Evaluation", "日期: 2026/08/01 六", "時間: 19:00", "地點: 新生"), + id="list-upcoming-events", + ), + ], +) +def test_notify_members_lists_upcoming_events(monkeypatch, upcoming, expected_parts): + _stub_staff(monkeypatch) + monkeypatch.setattr(mcp_server, "_get_upcoming_events", lambda limit: upcoming) + + result = _run(mcp_server.notify_members("部員")) + + assert all(part in result for part in expected_parts)