diff --git a/README.md b/README.md index dfc2561..3d8566f 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,15 @@ The best MCP server for Ptt. Proudly built by PyPtt< * `ghcr.io/pyptt/ptt_mcp_server:latest`: 指定要運行的 Docker 映像檔。 * `"env"`: 將 `PTT_ID` 和 `PTT_PW` 直接設定為環境變數。**請務必替換為您自己的 PTT 帳號和密碼。** + **多帳號設定(選用):** + 若您有多組 PTT 帳號,可改用環境變數 `PTT_ACCOUNTS`(格式:`名稱=帳號:密碼`,多組以逗號分隔)一次設定多組,之後即可用 `switch_account("名稱")` 切換帳號、`list_accounts()` 查詢可用名稱: + + ``` + PTT_ACCOUNTS=default=acc1:pw1,alt=acc2:pw2 + ``` + + 名稱不可含 `=`、帳號不可含 `:`、密碼不可含 `,`(PTT 帳密實務上不會用到這些字元)。使用 Docker 時,請在 `args` 中同時加上 `"-e", "PTT_ACCOUNTS"`。原本的單一 `PTT_ID`/`PTT_PW` 設定仍然相容,會自動成為名為 `default` 的帳號。 + 3. **啟動與測試**: 您的 MCP 客戶端現在應該能自動啟動 PTT MCP 伺服器了。您可以嘗試一個簡單的指令來測試連線,例如要求它登入 PTT。 diff --git a/README_ENG.md b/README_ENG.md index 1b66df1..03fd787 100644 --- a/README_ENG.md +++ b/README_ENG.md @@ -58,6 +58,15 @@ Using Docker is the most recommended way to deploy the PTT MCP Server, as it pro * `-e PTT_ID` and `-e PTT_PW`: Tells Docker to pass the `PTT_ID` and `PTT_PW` environment variables to the container. * `ghcr.io/PyPtt/ptt_mcp_server:latest`: Specifies the Docker image to run. * `env`: Sets `PTT_ID` and `PTT_PW` directly as environment variables. **Be sure to replace these with your own PTT account ID and password.** + + **Multiple accounts (optional):** + If you have more than one PTT account, you can configure them all at once with the `PTT_ACCOUNTS` environment variable (format: `name=id:password`, comma-separated), then switch between them with `switch_account("name")` and list the available names with `list_accounts()`: + + ``` + PTT_ACCOUNTS=default=acc1:pw1,alt=acc2:pw2 + ``` + + Names must not contain `=`, IDs must not contain `:`, and passwords must not contain `,` (PTT credentials never use these characters in practice). When using Docker, also add `"-e", "PTT_ACCOUNTS"` to `args`. The single `PTT_ID`/`PTT_PW` setup remains supported and becomes an account named `default`. 3. Launch and Test: Your MCP client should now be able to start the PTT MCP server automatically. You can try a simple command, such as asking it to log into PTT, to test the connection. diff --git a/src/api_post.py b/src/api_post.py index d743353..9e635cd 100644 --- a/src/api_post.py +++ b/src/api_post.py @@ -7,6 +7,22 @@ from utils import _call_ptt_service +def parse_date_str(date_str: str) -> datetime: + # Add a dummy year to make it a full date for comparison. + # Assuming all dates are within the current year for simplicity. + # For cross-year comparisons, more complex logic would be needed. + + if date_str.count('/') == 2: + return datetime.strptime(date_str, "%Y/%m/%d") + + current_year = datetime.now().year + return datetime.strptime(f"{current_year}/{date_str}", "%Y/%m/%d") + + +def _md(d: datetime): + return (d.month, d.day) + + def register_tools(mcp: FastMCP, memory_storage: Dict[str, Any], version: str): @mcp.tool() def get_board_rules() -> Dict[str, Any]: @@ -53,24 +69,14 @@ def get_post_index_range(board: str, target_date_str: str) -> Dict[str, Any]: 失敗時: {'success': False, 'message': str} """ - # Helper to parse and compare dates - def parse_date_str(date_str: str) -> datetime: - # Add a dummy year to make it a full date for comparison. - # Assuming all dates are within the current year for simplicity. - # For cross-year comparisons, more complex logic would be needed. - - if date_str.count('/') == 2: - return datetime.strptime(date_str, "%Y/%m/%d") - - current_year = datetime.now().year - return datetime.strptime(f"{current_year}/{date_str}", "%Y/%m/%d") - try: target_date = parse_date_str(target_date_str) except ValueError: return {"success": False, "message": f"Invalid target_date_str format: {target_date_str}. Expected 'YYYY/MM/DD'."} + # ponytail: 只比對 month/day(PTT list_date 無年份)。若目標日期區間跨年(12月→1月),binary search 的單調性會被打破、結果可能不對;需要跨年支援時得改用文章 header 的完整日期。 + # 1. Get the newest index for the board newest_index_response = _call_ptt_service( memory_storage, @@ -115,9 +121,9 @@ def parse_date_str(date_str: str) -> datetime: low = mid + 1 continue - if post_date < target_date: + if _md(post_date) < _md(target_date): low = mid + 1 - elif post_date == target_date: + elif _md(post_date) == _md(target_date): start_index = mid high = mid - 1 # Try to find an earlier one else: # post_date > target_date @@ -148,9 +154,9 @@ def parse_date_str(date_str: str) -> datetime: low = mid + 1 continue - if post_date > target_date: + if _md(post_date) > _md(target_date): high = mid - 1 - elif post_date == target_date: + elif _md(post_date) == _md(target_date): end_index = mid low = mid + 1 # Try to find a later one else: # post_date < target_date @@ -168,8 +174,8 @@ def parse_date_str(date_str: str) -> datetime: index=start_index, query=True, ) - if not start_post_response.get('success') or not start_post_response.get('data') or parse_date_str( - start_post_response['data'].get('list_date', '')) != target_date: + if not start_post_response.get('success') or not start_post_response.get('data') or _md(parse_date_str( + start_post_response['data'].get('list_date', ''))) != _md(target_date): return {"success": False, "message": f"在 {board} 板找不到日期 {target_date} 的任何文章 (start_index verification failed)."} @@ -181,8 +187,8 @@ def parse_date_str(date_str: str) -> datetime: index=end_index, query=True, ) - if not end_post_response.get('success') or not end_post_response.get('data') or parse_date_str( - end_post_response['data'].get('list_date', '')) != target_date: + if not end_post_response.get('success') or not end_post_response.get('data') or _md(parse_date_str( + end_post_response['data'].get('list_date', ''))) != _md(target_date): return {"success": False, "message": f"在 {board} 板找不到日期 {target_date} 的任何文章 (end_index verification failed)."} diff --git a/src/api_ptt.py b/src/api_ptt.py index 6efac73..95291ac 100644 --- a/src/api_ptt.py +++ b/src/api_ptt.py @@ -1,3 +1,4 @@ +import threading from typing import Dict, Any, Optional, List, Tuple import PyPtt @@ -5,6 +6,75 @@ from utils import _call_ptt_service, _handle_ptt_exception +# ponytail: module-level 鎖序列化池變動,防 FastMCP 併發呼叫造成 double-login/race。 +_pool_lock = threading.Lock() + + +def _login_new_service( + ptt_id: str, ptt_pw: str +) -> Tuple[Optional[Any], Dict[str, Any]]: + """建一個新 Service 並登入。成功回 (svc, 成功dict);失敗先 close() 掉失敗的 svc + 再回 (None, 錯誤dict)。此 helper 不碰 active session 或池。""" + ptt_service = PyPtt.Service({}) + try: + ptt_service.call( + "login", + { + "ptt_id": ptt_id, + "ptt_pw": ptt_pw, + "kick_other_session": True, + }, + ) + return ptt_service, {"success": True, "message": "登入成功"} + except Exception as e: + # 洩漏防護:登入失敗的 Service 內有 daemon thread + 連線,離開前必須 close。 + try: + ptt_service.close() + except Exception: + pass + return None, _handle_ptt_exception(e, {}) + + +def _is_alive(svc: Any) -> bool: + """驗活:對池裡的 Service 做一個便宜的 authenticated 呼叫探測,丟例外即視為 dead。""" + # ponytail: PTT 會斷閒置連線,光看 _is_login flag 偵測不到 server 端斷線,故主動用 get_time 探測。 + try: + svc.call("get_time") + return True + except Exception: + return False + + +def _activate(memory_storage: Dict[str, Any], name: str) -> Dict[str, Any]: + """把帳號 name 設為 active(登入並放進池)。原子性:新登入失敗不動 active session。""" + accounts = memory_storage["accounts"] + with _pool_lock: + pool = memory_storage["pool"] + svc = pool.get(name) + + # 池裡有但已死 → close 汰換(離開池的 Service 都要 close 防洩漏)。 + if svc is not None and not _is_alive(svc): + try: + svc.close() + except Exception: + pass + pool.pop(name, None) + svc = None + + if svc is None: + svc, result = _login_new_service(accounts[name]["id"], accounts[name]["pw"]) + if svc is None: + # 原子性關鍵:登入失敗,不動 ptt_bot / current_account / ptt_id / ptt_pw。 + return result + pool[name] = svc + + memory_storage["ptt_bot"] = svc + memory_storage["current_account"] = name + # 同步憑證,讓 bare login() 等仍與 active 帳號一致。 + memory_storage["ptt_id"] = accounts[name]["id"] + memory_storage["ptt_pw"] = accounts[name]["pw"] + return {"success": True, "message": "登入成功"} + def register_tools(mcp: FastMCP, memory_storage: Dict[str, Any], version: str): @mcp.tool() @@ -34,8 +104,20 @@ def logout() -> Dict[str, Any]: if ptt_service is None: return {"success": False, "message": "尚未登入,無需登出"} - result = _call_ptt_service(memory_storage, "logout", success_message="登出成功") - memory_storage["ptt_bot"] = None + # 只登出 active 帳號:logout 後 close、從池移除、ptt_bot=None;其他池內帳號不動。 + with _pool_lock: + result = _call_ptt_service( + memory_storage, "logout", success_message="登出成功" + ) + try: + ptt_service.close() + except Exception: + pass + pool = memory_storage["pool"] + for pool_name, pool_svc in list(pool.items()): + if pool_svc is ptt_service: + del pool[pool_name] + memory_storage["ptt_bot"] = None return result @mcp.tool() @@ -68,31 +150,83 @@ def login() -> Dict[str, Any]: - 'NEED_MODERATOR_PERMISSION': 需要看板管理員權限。 - 'UNKNOWN_ERROR': 操作時發生未知錯誤。 """ - # 如果已經有一個 bot 實例,先登出舊的 - if memory_storage["ptt_bot"] is not None: - try: - memory_storage["ptt_bot"].call("logout") - memory_storage["ptt_bot"] = None # 清除 session - except Exception: - pass + return _activate(memory_storage, memory_storage["current_account"]) - ptt_service = PyPtt.Service({}) - try: - ptt_service.call( - "login", - { - "ptt_id": memory_storage["ptt_id"], - "ptt_pw": memory_storage["ptt_pw"], - "kick_other_session": True, - }, - ) - # 登入成功後,將 bot 實例存起來 - memory_storage["ptt_bot"] = ptt_service + @mcp.tool() + def list_accounts() -> Dict[str, Any]: + """列出已在環境變數設定的 PTT 帳號名稱(不含密碼),以及目前登入中的帳號名稱。 - return {"success": True, "message": "登入成功"} - except Exception as e: - memory_storage["ptt_bot"] = None - return _handle_ptt_exception(e, {}) + Returns: + Dict[str, Any]: {'success': True, + 'accounts': ['default', 'alt', ...], + 'current': 'default' 或 None} + """ + # ponytail: 用 whoami 那套真實狀態判定(讀 PyPtt 私有 _is_login),沒 active 登入才不謊報。 + api = getattr(memory_storage.get("ptt_bot"), "_api", None) + current = ( + memory_storage.get("current_account") + if getattr(api, "_is_login", False) + else None + ) + return { + "success": True, + "accounts": list(memory_storage.get("accounts", {}).keys()), + "current": current, + } + + @mcp.tool() + def switch_account(name: str) -> Dict[str, Any]: + """切換到指定名稱的 PTT 帳號並登入。 + + 帳號需事先透過環境變數 PTT_ACCOUNTS 設定(見 README)。切換後會記住此帳號, + 後續 login() 也會使用它。可先用 list_accounts() 查詢可用名稱。 + + Args: + name (str): 帳號名稱。 + + Returns: + Dict[str, Any]: 成功 {'success': True, + 'message': "已切換到帳號 'xxx' 並登入成功", + 'current': 'xxx'} + 找不到名稱 {'success': False, + 'message': "找不到帳號 'xxx'", + 'available': [...]} + 登入失敗則回傳 login 相同的錯誤格式。 + """ + accounts = memory_storage.get("accounts", {}) + if name not in accounts: + return { + "success": False, + "message": f"找不到帳號 '{name}'", + "available": list(accounts.keys()), + } + result = _activate(memory_storage, name) + if result.get("success"): + result["message"] = f"已切換到帳號 '{name}' 並登入成功" + result["current"] = name + return result + + @mcp.tool() + def whoami() -> Dict[str, Any]: + """回傳目前登入中的 PTT 帳號資訊,實際登入狀態直接向 PyPtt 查詢。 + + Returns: + Dict[str, Any]: {'success': True, + 'account': 'default', # 帳號名稱(來自 PTT_ACCOUNTS 標籤) + 'ptt_id': 'CodingMan', # PyPtt 目前登入的實際 PTT ID;未登入為 None + 'logged_in': True} # PyPtt 回報是否已登入 + """ + # ponytail: 讀 PyPtt 私有屬性取真實狀態;PyPtt 無公開存取方法,改版壞了再換 + api = getattr(memory_storage.get("ptt_bot"), "_api", None) + logged_in = bool(getattr(api, "_is_login", False)) + # PyPtt 登出不清 ptt_id,未登入時不回報殘留值 + ptt_id = (getattr(api, "ptt_id", "") or None) if logged_in else None + return { + "success": True, + "account": memory_storage.get("current_account"), + "ptt_id": ptt_id, + "logged_in": logged_in, + } @mcp.tool() def get_post( diff --git a/src/auto_version.py b/src/auto_version.py index 27d27b6..1b83391 100644 --- a/src/auto_version.py +++ b/src/auto_version.py @@ -38,6 +38,10 @@ def get_version() -> tuple[Optional[str], Optional[str]]: return None, None # Should not be reached +def _version_key(version_str: str) -> tuple: + return tuple(int(part) for part in version_str.split('.')) + + def main(): remote_version, current_version = get_version() @@ -45,7 +49,7 @@ def main(): print("Failed to retrieve version information.") return - if int(remote_version.replace('.', '')) <= int(current_version.replace('.', '')): + if _version_key(remote_version) <= _version_key(current_version): print(current_version) else: print(remote_version) diff --git a/src/mcp_server.py b/src/mcp_server.py index 97fad68..ba5e437 100644 --- a/src/mcp_server.py +++ b/src/mcp_server.py @@ -1,3 +1,4 @@ +import atexit import os from typing import Dict, Any @@ -7,21 +8,75 @@ import api_ptt from _version import __version__ + +def parse_accounts(raw: str) -> Dict[str, Dict[str, str]]: + """解析 PTT_ACCOUNTS 字串:'名稱=帳號:密碼,名稱2=帳號2:密碼2'。""" + accounts: Dict[str, Dict[str, str]] = {} + # ponytail: 逗號分隔項目、= 分名稱、: 分帳密;名稱不可含 =、帳號不可含 :、密碼不可含 ,。 + # PTT 帳號為英數、密碼 ≤8 碼,實務上不會撞到這些分隔字元;若未來要支援特殊字元再換 JSON/檔案。 + for entry in raw.split(","): + entry = entry.strip() + if not entry: + continue + if "=" not in entry: + raise ValueError( + f"PTT_ACCOUNTS 項目格式錯誤(需 名稱=帳號:密碼): {entry!r}" + ) + name, cred = entry.split("=", 1) + name = name.strip() + if ":" not in cred: + raise ValueError(f"PTT_ACCOUNTS['{name}'] 需為 帳號:密碼 格式") + acc_id, acc_pw = cred.split(":", 1) + acc_id, acc_pw = acc_id.strip(), acc_pw.strip() + if not name or not acc_id or not acc_pw: + raise ValueError(f"PTT_ACCOUNTS 項目名稱/帳號/密碼不可為空: {entry!r}") + accounts[name] = {"id": acc_id, "pw": acc_pw} + return accounts + + PTT_ID = os.getenv("PTT_ID") PTT_PW = os.getenv("PTT_PW") +PTT_ACCOUNTS_RAW = os.getenv("PTT_ACCOUNTS") + +accounts: Dict[str, Dict[str, str]] = ( + parse_accounts(PTT_ACCOUNTS_RAW) if PTT_ACCOUNTS_RAW else {} +) -if not PTT_ID or not PTT_PW: - raise ValueError("PTT_ID and PTT_PW environment variables must be set.") +# 相容舊設定:PTT_ID/PTT_PW 視為名為 default 的帳號 +if PTT_ID and PTT_PW: + accounts.setdefault("default", {"id": PTT_ID, "pw": PTT_PW}) + +if not accounts: + raise ValueError("請設定 PTT_ACCOUNTS,或 PTT_ID 與 PTT_PW 環境變數。") + +initial = "default" if "default" in accounts else next(iter(accounts)) mcp: FastMCP = FastMCP(f"Ptt MCP Server v{__version__}") -MEMORY_STORAGE: Dict[str, Any] = {"ptt_bot": None, "ptt_id": PTT_ID, "ptt_pw": PTT_PW} +MEMORY_STORAGE: Dict[str, Any] = { + "ptt_bot": None, + "pool": {}, + "ptt_id": accounts[initial]["id"], + "ptt_pw": accounts[initial]["pw"], + "accounts": accounts, + "current_account": initial, +} + + +def _close_pool() -> None: + # 關站時關掉池內所有 Service(各有 daemon thread + 連線),避免洩漏。 + for svc in MEMORY_STORAGE["pool"].values(): + try: + svc.close() + except Exception: + pass def main(): api_ptt.register_tools(mcp, MEMORY_STORAGE, __version__) api_post.register_tools(mcp, MEMORY_STORAGE, __version__) + atexit.register(_close_pool) mcp.run() diff --git a/src/test_api_post.py b/src/test_api_post.py new file mode 100644 index 0000000..e0dd9a2 --- /dev/null +++ b/src/test_api_post.py @@ -0,0 +1,38 @@ +import os +import sys +import types + + +def _load_helpers(): + # Allow running from the repo root: `python src/test_api_post.py`. + src_dir = os.path.dirname(os.path.abspath(__file__)) + if src_dir not in sys.path: + sys.path.insert(0, src_dir) + + # api_post (via utils) imports PyPtt and fastmcp at module load time. Those + # are heavy runtime deps that may not be installed in a test environment, so + # stub them out before importing the pure date helpers we want to test. + sys.modules.setdefault("PyPtt", types.ModuleType("PyPtt")) + if "fastmcp" not in sys.modules: + fastmcp_stub = types.ModuleType("fastmcp") + setattr(fastmcp_stub, "FastMCP", object) + sys.modules["fastmcp"] = fastmcp_stub + + from api_post import _md, parse_date_str + + return _md, parse_date_str + + +def test_md_compare(): + _md, parse_date_str = _load_helpers() + + # A target date carrying a year (e.g. "1987/09/06") must match a yearless + # PTT list_date ("9/06") once compared via _md(). The original broken case. + assert _md(parse_date_str("1987/09/06")) == _md(parse_date_str("9/06")) + assert _md(parse_date_str("6/29")) == (6, 29) + assert _md(parse_date_str("6/29")) != _md(parse_date_str("6/30")) + + +if __name__ == "__main__": + test_md_compare() + print("OK") diff --git a/src/test_auto_version.py b/src/test_auto_version.py new file mode 100644 index 0000000..a4db103 --- /dev/null +++ b/src/test_auto_version.py @@ -0,0 +1,13 @@ +from auto_version import _version_key + + +def test_version_key(): + assert _version_key("0.3.0") > _version_key("0.2.15") + assert _version_key("1.0.0") > _version_key("0.20.0") + assert _version_key("0.10.0") > _version_key("0.9.0") + assert _version_key("0.3.0") == _version_key("0.3.0") + + +if __name__ == "__main__": + test_version_key() + print("OK") diff --git a/src/test_parse_accounts.py b/src/test_parse_accounts.py new file mode 100644 index 0000000..2085e76 --- /dev/null +++ b/src/test_parse_accounts.py @@ -0,0 +1,77 @@ +import os +import sys +import types + + +class _FakeFastMCP: + # mcp_server 會在模組載入時執行 FastMCP(name),stub 需可帶參數實例化。 + def __init__(self, *args, **kwargs): + pass + + +def _load_parse_accounts(): + # Allow running from the repo root: `python src/test_parse_accounts.py`. + src_dir = os.path.dirname(os.path.abspath(__file__)) + if src_dir not in sys.path: + sys.path.insert(0, src_dir) + + # mcp_server (via api_ptt/api_post/utils) imports PyPtt and fastmcp at + # module load time. Those are heavy runtime deps that may not be installed + # in a test environment, so stub them out before importing. + sys.modules.setdefault("PyPtt", types.ModuleType("PyPtt")) + fastmcp_mod = sys.modules.get("fastmcp") + if fastmcp_mod is None: + fastmcp_mod = types.ModuleType("fastmcp") + sys.modules["fastmcp"] = fastmcp_mod + # 其他測試檔的 stub 以 object 佔位,無法帶參數實例化,這裡一併升級。 + if getattr(fastmcp_mod, "FastMCP", None) in (None, object): + setattr(fastmcp_mod, "FastMCP", _FakeFastMCP) + + # mcp_server 載入時會讀取環境變數並驗證至少有一組帳號。 + os.environ.pop("PTT_ACCOUNTS", None) + os.environ.setdefault("PTT_ID", "testid") + os.environ.setdefault("PTT_PW", "testpw") + + from mcp_server import parse_accounts + + return parse_accounts + + +def test_parse_accounts_basic(): + parse_accounts = _load_parse_accounts() + + assert parse_accounts("default=a:b,alt=c:d") == { + "default": {"id": "a", "pw": "b"}, + "alt": {"id": "c", "pw": "d"}, + } + + +def test_parse_accounts_strips_whitespace(): + parse_accounts = _load_parse_accounts() + + assert parse_accounts(" default = a:b ") == {"default": {"id": "a", "pw": "b"}} + + +def test_parse_accounts_invalid(): + parse_accounts = _load_parse_accounts() + + for raw in [ + "noequal", # 缺 = + "name=nocolon", # 缺 : + "=a:b", # 空名稱 + "name=:pw", # 空帳號 + "name=a:", # 空密碼 + ]: + try: + parse_accounts(raw) + except ValueError: + pass + else: + raise AssertionError(f"expected ValueError for {raw!r}") + + +if __name__ == "__main__": + test_parse_accounts_basic() + test_parse_accounts_strips_whitespace() + test_parse_accounts_invalid() + print("OK") diff --git a/src/test_switch_account.py b/src/test_switch_account.py new file mode 100644 index 0000000..2da3c6f --- /dev/null +++ b/src/test_switch_account.py @@ -0,0 +1,148 @@ +import os +import sys +import types + + +class _FakeService: + # 記錄 login/close 呼叫、支援 get_time 驗活與 logout 的假 Service。 + def __init__(self, alive=True, login_error=None): + self.login_calls = 0 + self.close_calls = 0 + self.alive = alive + self.login_error = login_error + self._api = types.SimpleNamespace(_is_login=False, ptt_id=None) + + def call(self, api, args=None): + if api == "login": + self.login_calls += 1 + if self.login_error is not None: + raise self.login_error + self._api._is_login = True + self._api.ptt_id = (args or {}).get("ptt_id") + return None + if api == "get_time": + if not self.alive: + raise RuntimeError("dead connection") + return "12:34" + if api == "logout": + self._api._is_login = False + return None + return None + + def close(self): + self.close_calls += 1 + + +class _FakeMCP: + # Minimal stand-in for FastMCP that just records registered tools. + def __init__(self): + self.tools = {} + + def tool(self): + def decorator(func): + self.tools[func.__name__] = func + return func + + return decorator + + +def _load_api_ptt(): + # Allow running from the repo root: `python src/test_switch_account.py`. + src_dir = os.path.dirname(os.path.abspath(__file__)) + if src_dir not in sys.path: + sys.path.insert(0, src_dir) + + # api_ptt (and utils) import PyPtt and fastmcp at module load time. Those + # are heavy runtime deps that may not be installed in a test environment, + # so stub them out before importing. + sys.modules.setdefault("PyPtt", types.ModuleType("PyPtt")) + if "fastmcp" not in sys.modules: + fastmcp_stub = types.ModuleType("fastmcp") + setattr(fastmcp_stub, "FastMCP", object) + sys.modules["fastmcp"] = fastmcp_stub + + import api_ptt + + return api_ptt + + +def _register_tools(api_ptt): + mcp = _FakeMCP() + memory_storage = { + "ptt_bot": None, + "pool": {}, + "ptt_id": "acc1", + "ptt_pw": "pw1", + "accounts": { + "default": {"id": "acc1", "pw": "pw1"}, + "alt": {"id": "acc2", "pw": "pw2"}, + }, + "current_account": "default", + } + api_ptt.register_tools(mcp, memory_storage, "0.0.0") + return mcp.tools, memory_storage + + +def test_switch_account_unknown_name(): + api_ptt = _load_api_ptt() + tools, memory_storage = _register_tools(api_ptt) + + # 找不到帳號時提前返回,不進 _activate,故無需 stub 登入。 + result = tools["switch_account"]("nope") + + assert result["success"] is False + assert sorted(result["available"]) == ["alt", "default"] + # 找不到帳號時不得改動現有帳密與登入狀態 + assert memory_storage["ptt_id"] == "acc1" + assert memory_storage["ptt_pw"] == "pw1" + assert memory_storage["current_account"] == "default" + + +def test_switch_account_success(): + api_ptt = _load_api_ptt() + tools, memory_storage = _register_tools(api_ptt) + + original = api_ptt._login_new_service + setattr( + api_ptt, + "_login_new_service", + lambda ptt_id, ptt_pw: (_FakeService(), {"success": True, "message": "登入成功"}), + ) + try: + result = tools["switch_account"]("alt") + finally: + setattr(api_ptt, "_login_new_service", original) + + assert result["success"] is True + assert result["current"] == "alt" + assert result["message"] == "已切換到帳號 'alt' 並登入成功" + assert memory_storage["ptt_id"] == "acc2" + assert memory_storage["ptt_pw"] == "pw2" + assert memory_storage["current_account"] == "alt" + # active svc 進池、ptt_bot 指到它 + assert memory_storage["pool"]["alt"] is memory_storage["ptt_bot"] + + +def test_list_accounts(): + api_ptt = _load_api_ptt() + tools, memory_storage = _register_tools(api_ptt) + + result = tools["list_accounts"]() + assert result["success"] is True + assert sorted(result["accounts"]) == ["alt", "default"] + # 未登入(ptt_bot=None)不得謊報 current + assert result["current"] is None + + # 有 active 登入時,current 反映真實登入帳號 + memory_storage["current_account"] = "alt" + memory_storage["ptt_bot"] = types.SimpleNamespace( + _api=types.SimpleNamespace(_is_login=True, ptt_id="acc2") + ) + assert tools["list_accounts"]()["current"] == "alt" + + +if __name__ == "__main__": + test_switch_account_unknown_name() + test_switch_account_success() + test_list_accounts() + print("OK") diff --git a/src/test_switch_account_pool.py b/src/test_switch_account_pool.py new file mode 100644 index 0000000..a5e57f7 --- /dev/null +++ b/src/test_switch_account_pool.py @@ -0,0 +1,177 @@ +"""Service 池行為測試:原子切換、死連線汰換、洩漏防護(close)、真實登入狀態。 + +沿用 test_switch_account 的 _load_api_ptt/_register_tools 與假 Service。 +以「工廠依預排順序發假 Service」驅動 _login_new_service 的真實路徑, +藉此連 close-on-failed-login 一併驗到。 +""" + +from test_switch_account import _FakeService, _load_api_ptt, _register_tools + +# _handle_ptt_exception 會參照這些 PyPtt 例外類別,stub 需補上才不會 AttributeError。 +_PTT_EXCEPTIONS = [ + "RequireLogin", "UnregisteredUser", "NoSuchBoard", "NoSuchPost", + "NoPermission", "LoginError", "WrongIDorPassword", "CantResponse", + "NoFastComment", "NoSuchUser", "NoSuchMail", "MailboxFull", "NoMoney", + "SetContactMailFirst", "WrongPassword", "NeedModeratorPermission", +] + + +class _ServiceFactory: + # 取代 PyPtt.Service:依測試預排順序把假 Service 發給 _login_new_service。 + def __init__(self): + self.handed_out = [] + self._queue = [] + + def prepare(self, *svcs): + self._queue.extend(svcs) + + def __call__(self, config=None): + svc = self._queue.pop(0) + self.handed_out.append(svc) + return svc + + @property + def login_total(self): + return sum(s.login_calls for s in self.handed_out) + + +def _setup(api_ptt): + ptt = api_ptt.PyPtt + for name in _PTT_EXCEPTIONS: + if not hasattr(ptt, name): + setattr(ptt, name, type(name, (Exception,), {})) + factory = _ServiceFactory() + setattr(ptt, "Service", factory) + return factory + + +def test_atomic_on_failed_switch(): + # (a)(e) 切到密碼錯的帳號後,active session 與 current 維持不變;失敗的 svc 被 close。 + api_ptt = _load_api_ptt() + factory = _setup(api_ptt) + tools, ms = _register_tools(api_ptt) + ms["accounts"]["bad"] = {"id": "accX", "pw": "wrong"} + + svc_default = _FakeService() + wrong = api_ptt.PyPtt.WrongIDorPassword("bad pw") + svc_bad = _FakeService(login_error=wrong) + factory.prepare(svc_default, svc_bad) + + assert tools["switch_account"]("default")["success"] is True + assert ms["ptt_bot"] is svc_default + assert ms["current_account"] == "default" + + result = tools["switch_account"]("bad") + assert result["success"] is False + assert ms["ptt_bot"] is svc_default # 原子性:active 不變 + assert ms["current_account"] == "default" + assert ms["ptt_id"] == "acc1" + assert svc_bad.close_calls == 1 # 洩漏防護 + assert "bad" not in ms["pool"] + + +def test_switch_to_new_account(): + # (b) 成功切到新帳號:池多一筆、ptt_bot 指到新 svc、current 更新。 + api_ptt = _load_api_ptt() + factory = _setup(api_ptt) + tools, ms = _register_tools(api_ptt) + + svc_default = _FakeService() + svc_alt = _FakeService() + factory.prepare(svc_default, svc_alt) + + tools["switch_account"]("default") + result = tools["switch_account"]("alt") + + assert result["success"] is True + assert set(ms["pool"]) == {"default", "alt"} + assert ms["ptt_bot"] is svc_alt + assert ms["current_account"] == "alt" + + +def test_switch_back_to_alive_reuses(): + # (c) 切回池中活著的帳號:不觸發新 login,ptt_bot 重指到既有 svc。 + api_ptt = _load_api_ptt() + factory = _setup(api_ptt) + tools, ms = _register_tools(api_ptt) + + svc_default = _FakeService(alive=True) + svc_alt = _FakeService(alive=True) + factory.prepare(svc_default, svc_alt) + + tools["switch_account"]("default") + tools["switch_account"]("alt") + logins_before = factory.login_total + services_before = len(factory.handed_out) + + result = tools["switch_account"]("default") + assert result["success"] is True + assert factory.login_total == logins_before # 沒有新 login + assert len(factory.handed_out) == services_before # 沒有新 Service + assert ms["ptt_bot"] is svc_default + assert ms["current_account"] == "default" + + +def test_switch_to_dead_relogs(): + # (d)(e) 切到池中死掉的帳號:舊 svc 被 close,且發生一次重登。 + api_ptt = _load_api_ptt() + factory = _setup(api_ptt) + tools, ms = _register_tools(api_ptt) + + svc_default = _FakeService(alive=True) + svc_alt = _FakeService(alive=True) + svc_default2 = _FakeService(alive=True) + factory.prepare(svc_default, svc_alt, svc_default2) + + tools["switch_account"]("default") + tools["switch_account"]("alt") + svc_default.alive = False # 模擬 server 端閒置斷線 + + result = tools["switch_account"]("default") + assert result["success"] is True + assert svc_default.close_calls == 1 # 死連線被汰換並 close + assert svc_default2.login_calls == 1 # 一次重登 + assert ms["ptt_bot"] is svc_default2 + assert ms["pool"]["default"] is svc_default2 + + +def test_logout_closes_active_only(): + # logout 只登出/close/移除 active 帳號,其他池內帳號不動。 + api_ptt = _load_api_ptt() + factory = _setup(api_ptt) + tools, ms = _register_tools(api_ptt) + + svc_default = _FakeService() + svc_alt = _FakeService() + factory.prepare(svc_default, svc_alt) + + tools["switch_account"]("default") + tools["switch_account"]("alt") # active = alt + + result = tools["logout"]() + assert result["success"] is True + assert svc_alt.close_calls == 1 + assert "alt" not in ms["pool"] + assert ms["ptt_bot"] is None + assert ms["pool"]["default"] is svc_default # 其他池內帳號不動 + assert svc_default.close_calls == 0 + + +def test_list_accounts_current_none_when_not_logged_in(): + # (f) 未登入時 current 為 None。 + api_ptt = _load_api_ptt() + _setup(api_ptt) + tools, ms = _register_tools(api_ptt) + + assert ms["ptt_bot"] is None + assert tools["list_accounts"]()["current"] is None + + +if __name__ == "__main__": + test_atomic_on_failed_switch() + test_switch_to_new_account() + test_switch_back_to_alive_reuses() + test_switch_to_dead_relogs() + test_logout_closes_active_only() + test_list_accounts_current_none_when_not_logged_in() + print("OK") diff --git a/src/test_whoami.py b/src/test_whoami.py new file mode 100644 index 0000000..67042f8 --- /dev/null +++ b/src/test_whoami.py @@ -0,0 +1,45 @@ +import types + +from test_switch_account import _load_api_ptt, _register_tools + + +def test_whoami_not_logged_in(): + tools, memory_storage = _register_tools(_load_api_ptt()) + + result = tools["whoami"]() + assert result["success"] is True + assert result["account"] == "default" + assert result["ptt_id"] is None + assert result["logged_in"] is False + + +def test_whoami_logged_in(): + tools, memory_storage = _register_tools(_load_api_ptt()) + memory_storage["ptt_bot"] = types.SimpleNamespace( + _api=types.SimpleNamespace(_is_login=True, ptt_id="CodingMan") + ) + memory_storage["current_account"] = "alt" + + result = tools["whoami"]() + assert result["account"] == "alt" + assert result["ptt_id"] == "CodingMan" + assert result["logged_in"] is True + + +def test_whoami_stale_ptt_id_after_logout(): + # PyPtt 登出後不會清 ptt_id,whoami 不得回報殘留值 + tools, memory_storage = _register_tools(_load_api_ptt()) + memory_storage["ptt_bot"] = types.SimpleNamespace( + _api=types.SimpleNamespace(_is_login=False, ptt_id="stale") + ) + + result = tools["whoami"]() + assert result["logged_in"] is False + assert result["ptt_id"] is None + + +if __name__ == "__main__": + test_whoami_not_logged_in() + test_whoami_logged_in() + test_whoami_stale_ptt_id_after_logout() + print("OK")