From 363dc96b12c96718b36fda0f9f4467fad7575d06 Mon Sep 17 00:00:00 2001 From: Peng Ding Date: Wed, 8 Apr 2026 20:04:29 +0800 Subject: [PATCH 1/2] refactor: deduplicate shared helpers and unify QR login state machine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extract shared constants and helpers into _helpers.py: - MEDIA_EXT_MAP / MEDIA_MIME_MAP (were duplicated 4x with inconsistencies) - media_filename() — derives filename from Message - parse_direction() / parse_message_type() / parse_time() - process_qr_status() — single QR login credential extraction Update all consumers (cli.py, server/app.py, admin/handlers.py, client.py, examples/media_download.py) to import from _helpers.py. Fixes: .silk → .amr inconsistency in examples/media_download.py Removes: layering violation (cli.py importing _parse_time from server/app.py) Closes #27, closes #23 --- examples/media_download.py | 20 +--- src/weilink/_helpers.py | 140 ++++++++++++++++++++++ src/weilink/admin/handlers.py | 44 ++----- src/weilink/cli.py | 38 ++---- src/weilink/client.py | 36 ++---- src/weilink/server/app.py | 83 ++++--------- tests/test_helpers.py | 218 ++++++++++++++++++++++++++++++++++ 7 files changed, 419 insertions(+), 160 deletions(-) create mode 100644 src/weilink/_helpers.py create mode 100644 tests/test_helpers.py diff --git a/examples/media_download.py b/examples/media_download.py index 5eb5b99..8822a6d 100644 --- a/examples/media_download.py +++ b/examples/media_download.py @@ -21,6 +21,7 @@ from pathlib import Path from weilink import MessageType, WeiLink +from weilink._helpers import media_filename logging.basicConfig( level=getattr(logging, os.environ.get("LOGLEVEL", "INFO").upper(), logging.INFO), @@ -28,26 +29,13 @@ ) logger = logging.getLogger(__name__) -# File extensions by message type -_EXT: dict[MessageType, str] = { - MessageType.IMAGE: ".jpg", - MessageType.VOICE: ".silk", - MessageType.VIDEO: ".mp4", -} - -def _save_path(save_dir: Path, msg_type: MessageType, msg: object) -> Path: +def _save_path(save_dir: Path, msg: object) -> Path: """Build a unique save path for a media message.""" from weilink.models import Message assert isinstance(msg, Message) - mid = msg.message_id or 0 - - # Use original file name for FILE type - if msg_type == MessageType.FILE and msg.file: - name = msg.file.file_name or f"{mid}.bin" - else: - name = f"{mid}{_EXT.get(msg_type, '.bin')}" + name = media_filename(msg) path = save_dir / name # Avoid overwriting — append counter if file exists @@ -100,7 +88,7 @@ def main() -> None: wl.send(user, f"[Download failed: {e}]") continue - path = _save_path(save_dir, msg.msg_type, msg) + path = _save_path(save_dir, msg) path.write_bytes(data) logger.info( "Saved %s (%d bytes) -> %s", msg.msg_type.name, len(data), path diff --git a/src/weilink/_helpers.py b/src/weilink/_helpers.py new file mode 100644 index 0000000..663c8d1 --- /dev/null +++ b/src/weilink/_helpers.py @@ -0,0 +1,140 @@ +"""Shared constants and helpers used across CLI, server, and admin.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from weilink.models import BotInfo, Message, MessageDirection, MessageType + + +# -- Media constants --------------------------------------------------------- + +MEDIA_EXT_MAP: dict[MessageType, str] = { + MessageType.IMAGE: ".jpg", + MessageType.VOICE: ".amr", + MessageType.VIDEO: ".mp4", +} + +MEDIA_MIME_MAP: dict[MessageType, str] = { + MessageType.IMAGE: "image/jpeg", + MessageType.VOICE: "audio/amr", + MessageType.VIDEO: "video/mp4", +} + + +# -- QR login ---------------------------------------------------------------- + + +@dataclass(frozen=True) +class QRResult: + """Result of interpreting a ``poll_qr_status`` response. + + Attributes: + status: Normalized status string — one of + ``"confirmed"``, ``"scanned"``, ``"expired"``, or ``"waiting"``. + bot_info: Populated only when *status* is ``"confirmed"``. + """ + + status: str + bot_info: BotInfo | None = None + + +def process_qr_status(status_resp: dict[str, Any]) -> QRResult: + """Interpret a raw ``poll_qr_status`` response. + + Args: + status_resp: Dict returned by ``_protocol.poll_qr_status()``. + + Returns: + A :class:`QRResult` with normalized status and, on confirmation, + the extracted :class:`BotInfo`. + """ + from weilink._protocol import BASE_URL + + status = status_resp.get("status", "") + + if status == "confirmed": + bot_info = BotInfo( + bot_id=status_resp.get("ilink_bot_id", ""), + base_url=status_resp.get("baseurl", BASE_URL), + token=status_resp.get("bot_token", ""), + user_id=status_resp.get("ilink_user_id", ""), + ) + return QRResult(status="confirmed", bot_info=bot_info) + + if status == "scaned": + return QRResult(status="scanned") + + if status == "expired": + return QRResult(status="expired") + + return QRResult(status="waiting") + + +# -- Media helpers ----------------------------------------------------------- + + +def media_filename(msg: Message) -> str: + """Derive a filename for a media message. + + Uses the original ``file_name`` when available, otherwise falls back + to ``{message_id}{ext}`` using :data:`MEDIA_EXT_MAP`. + """ + if msg.file and msg.file.file_name: + return msg.file.file_name + ext = MEDIA_EXT_MAP.get(msg.msg_type, ".bin") + return f"{msg.message_id}{ext}" + + +# -- Parsing helpers --------------------------------------------------------- + + +def parse_direction(s: str) -> int | None: + """Parse a direction string to its integer value. + + Returns: + ``MessageDirection.USER`` (1) for ``"received"``, + ``MessageDirection.BOT`` (2) for ``"sent"``, + or ``None`` for unrecognized input. + """ + d = s.lower() + if d == "received": + return MessageDirection.USER + if d == "sent": + return MessageDirection.BOT + return None + + +def parse_message_type(s: str) -> int | None: + """Parse a message type name to its integer value. + + Returns: + The ``MessageType`` integer (e.g. 2 for ``"IMAGE"``), + or ``None`` for unrecognized input. + """ + try: + return MessageType[s.upper()].value + except KeyError: + return None + + +def parse_time(s: str) -> int | None: + """Parse an ISO 8601 string or unix milliseconds to *int* milliseconds. + + Returns: + Millisecond timestamp, or ``None`` on failure. + """ + if not s: + return None + try: + return int(s) + except ValueError: + pass + try: + from datetime import datetime + + dt = datetime.fromisoformat(s.replace("Z", "+00:00")) + return int(dt.timestamp() * 1000) + except (ValueError, TypeError): + return None diff --git a/src/weilink/admin/handlers.py b/src/weilink/admin/handlers.py index b88258d..795cff2 100644 --- a/src/weilink/admin/handlers.py +++ b/src/weilink/admin/handlers.py @@ -12,7 +12,7 @@ from typing import TYPE_CHECKING, Any, ClassVar from weilink import __version__ -from weilink.models import BotInfo +from weilink._helpers import MEDIA_MIME_MAP, media_filename, process_qr_status from .static import ADMIN_HTML, load_locale @@ -248,13 +248,10 @@ def _handle_poll_login(self, query: dict[str, list[str]]) -> None: self._send_json({"status": "waiting"}) return - status = status_resp.get("status", "") + qr = process_qr_status(status_resp) - if status == "confirmed": - bot_token = status_resp.get("bot_token", "") - base_url = status_resp.get("baseurl", proto.BASE_URL) - bot_id = status_resp.get("ilink_bot_id", "") - user_id = status_resp.get("ilink_user_id", "") + if qr.status == "confirmed": + assert qr.bot_info is not None name = pending["name"] with self._lock: @@ -265,12 +262,7 @@ def _handle_poll_login(self, query: dict[str, list[str]]) -> None: token_path = wl._base_path / name / "token.json" session = wl._create_session(name, token_path) - session.bot_info = BotInfo( - bot_id=bot_id, - base_url=base_url, - token=bot_token, - user_id=user_id, - ) + session.bot_info = qr.bot_info session.cursor = "" wl._save_session_state(session) @@ -278,15 +270,15 @@ def _handle_poll_login(self, query: dict[str, list[str]]) -> None: self._send_json( { "status": "confirmed", - "bot_id": bot_id, + "bot_id": qr.bot_info.bot_id, "session_name": name, } ) - elif status == "scaned": + elif qr.status == "scanned": self._send_json({"status": "scaned"}) - elif status == "expired": + elif qr.status == "expired": del self._pending_logins[qrcode] self._send_json({"status": "expired"}) @@ -427,17 +419,6 @@ def _handle_get_messages(self, query: dict[str, list[str]]) -> None: m["message_id"] = str(m["message_id"]) self._send_json({"messages": messages, "total": total}) - _MIME_MAP: ClassVar[dict[str, str]] = { - "IMAGE": "image/jpeg", - "VOICE": "audio/amr", - "VIDEO": "video/mp4", - } - _EXT_MAP: ClassVar[dict[str, str]] = { - "IMAGE": ".jpg", - "VOICE": ".amr", - "VIDEO": ".mp4", - } - def _handle_download_media(self, message_id_str: str) -> None: """Download media from a stored message and serve the bytes.""" store = self.weilink._message_store @@ -463,13 +444,8 @@ def _handle_download_media(self, message_id_str: str) -> None: return # Derive filename and MIME type - mt = msg.msg_type.name - if msg.file and msg.file.file_name: - filename = msg.file.file_name - else: - ext = self._EXT_MAP.get(mt, ".bin") - filename = f"{msg.message_id}{ext}" - content_type = self._MIME_MAP.get(mt, "application/octet-stream") + filename = media_filename(msg) + content_type = MEDIA_MIME_MAP.get(msg.msg_type, "application/octet-stream") self.send_response(200) self.send_header("Content-Type", content_type) diff --git a/src/weilink/cli.py b/src/weilink/cli.py index 1ef8c8b..5fd692f 100644 --- a/src/weilink/cli.py +++ b/src/weilink/cli.py @@ -264,17 +264,9 @@ def _run_download(args: argparse.Namespace) -> None: out_dir.mkdir(parents=True, exist_ok=True) # Derive filename - from weilink.models import MessageType - - ext_map = { - MessageType.IMAGE: ".jpg", - MessageType.VOICE: ".amr", - MessageType.VIDEO: ".mp4", - } - if msg.file and msg.file.file_name: - name = msg.file.file_name - else: - name = f"{msg.message_id}{ext_map.get(msg.msg_type, '.bin')}" + from weilink._helpers import media_filename + + name = media_filename(msg) out_path = out_dir / name out_path.write_bytes(data) @@ -299,7 +291,7 @@ def _run_history(args: argparse.Namespace) -> None: wl.close() sys.exit(1) - from weilink.models import MessageType + from weilink._helpers import parse_direction, parse_message_type, parse_time kwargs: dict[str, Any] = {} if args.user: @@ -307,9 +299,8 @@ def _run_history(args: argparse.Namespace) -> None: if args.bot: kwargs["bot_id"] = args.bot if args.type: - try: - kwargs["msg_type"] = MessageType[args.type.upper()].value - except KeyError: + mt = parse_message_type(args.type) + if mt is None: err = f"Unknown message type: {args.type}" if _json_flag(args): print(json.dumps({"error": err})) @@ -317,22 +308,17 @@ def _run_history(args: argparse.Namespace) -> None: print(f"Error: {err}", file=sys.stderr) wl.close() sys.exit(1) + kwargs["msg_type"] = mt if args.direction: - d = args.direction.lower() - if d == "received": - kwargs["direction"] = 1 - elif d == "sent": - kwargs["direction"] = 2 + d = parse_direction(args.direction) + if d is not None: + kwargs["direction"] = d if args.since: - from weilink.server.app import _parse_time - - ts = _parse_time(args.since) + ts = parse_time(args.since) if ts is not None: kwargs["since_ms"] = ts if args.until: - from weilink.server.app import _parse_time - - ts = _parse_time(args.until) + ts = parse_time(args.until) if ts is not None: kwargs["until_ms"] = ts if args.text: diff --git a/src/weilink/client.py b/src/weilink/client.py index e2bd2b6..15274b1 100644 --- a/src/weilink/client.py +++ b/src/weilink/client.py @@ -19,6 +19,7 @@ from weilink._vendor.filelock import FileLock from weilink import _protocol as proto +from weilink._helpers import process_qr_status from weilink.models import ( BotInfo, FileInfo, @@ -837,35 +838,28 @@ def login(self, name: str | None = None, force: bool = False) -> BotInfo: print(".", end="", flush=True) continue - status = status_resp.get("status", "") - logger.debug("QR poll status=%s, keys=%s", status, list(status_resp.keys())) + qr = process_qr_status(status_resp) + logger.debug( + "QR poll status=%s, keys=%s", qr.status, list(status_resp.keys()) + ) - if status == "confirmed": + if qr.status == "confirmed": + assert qr.bot_info is not None logger.debug( "QR confirmed, response keys: %s", list(status_resp.keys()) ) - bot_token = status_resp.get("bot_token", "") - base_url = status_resp.get("baseurl", proto.BASE_URL) - bot_id = status_resp.get("ilink_bot_id", "") - user_id = status_resp.get("ilink_user_id", "") - - session.bot_info = BotInfo( - bot_id=bot_id, - base_url=base_url, - token=bot_token, - user_id=user_id, - ) + session.bot_info = qr.bot_info session.cursor = "" with self._data_lock: self._save_session_state(session) - print(f"\nLogin successful! Bot ID: {bot_id}") + print(f"\nLogin successful! Bot ID: {qr.bot_info.bot_id}") return session.bot_info - if status == "scaned": + if qr.status == "scanned": print("\nScanned, confirm on your phone...", end="", flush=True) continue - if status == "expired": + if qr.status == "expired": print("\nQR code expired, refreshing...") qr_resp = proto.get_qr_code() qrcode = qr_resp["qrcode"] @@ -874,14 +868,6 @@ def login(self, name: str | None = None, force: bool = False) -> BotInfo: print("Waiting for scan...", end="", flush=True) continue - # "wait" = server says no scan yet; any other unknown status - # is treated the same — keep polling. - if status not in ("wait", ""): - logger.info( - "QR poll unknown status=%r, full_resp=%s", - status, - str(status_resp)[:500], - ) print(".", end="", flush=True) raise proto.ILinkError(ret=-1, errmsg="QR code login timed out (5 min)") diff --git a/src/weilink/server/app.py b/src/weilink/server/app.py index 4ad71c1..d7bc547 100644 --- a/src/weilink/server/app.py +++ b/src/weilink/server/app.py @@ -14,7 +14,14 @@ from pathlib import Path from typing import Any, Literal -from weilink import Message, MessageType, WeiLink +from weilink import Message, WeiLink +from weilink._helpers import ( + media_filename, + parse_direction, + parse_message_type, + parse_time, + process_qr_status, +) from weilink._protocol import ILinkError, SessionExpiredError logger = logging.getLogger(__name__) @@ -228,7 +235,7 @@ async def download(message_id: str, save_dir: str = "") -> str: return json.dumps({"error": str(e)}) # Determine file name - name = _media_filename(msg) + name = media_filename(msg) out_dir = Path(save_dir) if save_dir else _DEFAULT_DOWNLOAD_DIR out_dir.mkdir(parents=True, exist_ok=True) out_path = out_dir / name @@ -242,19 +249,6 @@ async def download(message_id: str, save_dir: str = "") -> str: return json.dumps({"path": str(out_path), "size": len(data)}) -def _media_filename(msg: Message) -> str: - """Derive a reasonable file name from a media message.""" - ext_map = { - MessageType.IMAGE: ".jpg", - MessageType.VOICE: ".amr", - MessageType.VIDEO: ".mp4", - } - if msg.file and msg.file.file_name: - return msg.file.file_name - ext = ext_map.get(msg.msg_type, ".bin") - return f"{msg.message_id}{ext}" - - def sessions() -> str: """List all WeiLink sessions and their connection status. @@ -348,17 +342,11 @@ async def login( await asyncio.sleep(2) continue - status = status_resp.get("status", "") + qr = process_qr_status(status_resp) - if status == "confirmed": + if qr.status == "confirmed": + assert qr.bot_info is not None wl = _get_client() - bot_token = status_resp.get("bot_token", "") - base_url = status_resp.get("baseurl", proto.BASE_URL) - bot_id = status_resp.get("ilink_bot_id", "") - user_id = status_resp.get("ilink_user_id", "") - - from weilink.models import BotInfo - name = session_name_stored or "default" if name in wl._sessions: session = wl._sessions[name] @@ -366,9 +354,7 @@ async def login( token_path = wl._base_path / name / "token.json" session = wl._create_session(name, token_path) - session.bot_info = BotInfo( - bot_id=bot_id, base_url=base_url, token=bot_token, user_id=user_id - ) + session.bot_info = qr.bot_info session.cursor = "" wl._save_session_state(session) @@ -376,14 +362,13 @@ async def login( return json.dumps( { "status": "confirmed", - "bot_id": bot_id, + "bot_id": qr.bot_info.bot_id, "session": name, "message": "Login successful!", } ) - if status == "scaned": - # Return immediately on scan — agent knows user is interacting + if qr.status == "scanned": last_status = "scanned" return json.dumps( { @@ -392,7 +377,7 @@ async def login( } ) - if status == "expired": + if qr.status == "expired": _pending_login = None return json.dumps( { @@ -450,24 +435,21 @@ def history( if bot_id: kwargs["bot_id"] = bot_id if msg_type: - try: - kwargs["msg_type"] = MessageType[msg_type.upper()].value - except KeyError: + mt = parse_message_type(msg_type) + if mt is None: return json.dumps({"error": f"Unknown msg_type: {msg_type}"}) + kwargs["msg_type"] = mt if direction: - d = direction.lower() - if d == "received": - kwargs["direction"] = 1 - elif d == "sent": - kwargs["direction"] = 2 - else: + d = parse_direction(direction) + if d is None: return json.dumps({"error": f"Unknown direction: {direction}"}) + kwargs["direction"] = d if since: - ts = _parse_time(since) + ts = parse_time(since) if ts is not None: kwargs["since_ms"] = ts if until: - ts = _parse_time(until) + ts = parse_time(until) if ts is not None: kwargs["until_ms"] = ts if text_contains: @@ -480,23 +462,6 @@ def history( ) -def _parse_time(s: str) -> int | None: - """Parse an ISO 8601 string or unix milliseconds to int ms.""" - if not s: - return None - try: - return int(s) - except ValueError: - pass - try: - from datetime import datetime - - dt = datetime.fromisoformat(s.replace("Z", "+00:00")) - return int(dt.timestamp() * 1000) - except (ValueError, TypeError): - return None - - async def logout(session_name: str = "") -> str: """Log out a WeiLink session. diff --git a/tests/test_helpers.py b/tests/test_helpers.py new file mode 100644 index 0000000..c2e042e --- /dev/null +++ b/tests/test_helpers.py @@ -0,0 +1,218 @@ +"""Tests for weilink._helpers module.""" + +from __future__ import annotations + +import pytest + +from weilink._helpers import ( + MEDIA_EXT_MAP, + MEDIA_MIME_MAP, + QRResult, + media_filename, + parse_direction, + parse_message_type, + parse_time, + process_qr_status, +) +from weilink.models import BotInfo, FileInfo, Message, MessageType + + +# -- Constants --------------------------------------------------------------- + + +class TestMediaMaps: + def test_ext_map_entries(self): + assert MEDIA_EXT_MAP[MessageType.IMAGE] == ".jpg" + assert MEDIA_EXT_MAP[MessageType.VOICE] == ".amr" + assert MEDIA_EXT_MAP[MessageType.VIDEO] == ".mp4" + + def test_ext_map_excludes_text_and_file(self): + assert MessageType.TEXT not in MEDIA_EXT_MAP + assert MessageType.FILE not in MEDIA_EXT_MAP + + def test_mime_map_entries(self): + assert MEDIA_MIME_MAP[MessageType.IMAGE] == "image/jpeg" + assert MEDIA_MIME_MAP[MessageType.VOICE] == "audio/amr" + assert MEDIA_MIME_MAP[MessageType.VIDEO] == "video/mp4" + + def test_mime_map_excludes_text_and_file(self): + assert MessageType.TEXT not in MEDIA_MIME_MAP + assert MessageType.FILE not in MEDIA_MIME_MAP + + +# -- media_filename ---------------------------------------------------------- + + +class TestMediaFilename: + def test_uses_file_name_when_present(self): + msg = Message( + from_user="u@im.wechat", + msg_type=MessageType.FILE, + file=FileInfo(file_name="report.pdf"), + message_id=42, + ) + assert media_filename(msg) == "report.pdf" + + def test_image_fallback(self): + msg = Message( + from_user="u@im.wechat", + msg_type=MessageType.IMAGE, + message_id=100, + ) + assert media_filename(msg) == "100.jpg" + + def test_voice_fallback(self): + msg = Message( + from_user="u@im.wechat", + msg_type=MessageType.VOICE, + message_id=200, + ) + assert media_filename(msg) == "200.amr" + + def test_video_fallback(self): + msg = Message( + from_user="u@im.wechat", + msg_type=MessageType.VIDEO, + message_id=300, + ) + assert media_filename(msg) == "300.mp4" + + def test_text_fallback_to_bin(self): + msg = Message( + from_user="u@im.wechat", + msg_type=MessageType.TEXT, + message_id=400, + ) + assert media_filename(msg) == "400.bin" + + +# -- parse_direction --------------------------------------------------------- + + +class TestParseDirection: + def test_received(self): + assert parse_direction("received") == 1 + + def test_sent(self): + assert parse_direction("sent") == 2 + + def test_case_insensitive(self): + assert parse_direction("RECEIVED") == 1 + assert parse_direction("Sent") == 2 + + def test_invalid(self): + assert parse_direction("invalid") is None + + def test_empty(self): + assert parse_direction("") is None + + +# -- parse_message_type ------------------------------------------------------ + + +class TestParseMessageType: + def test_image(self): + assert parse_message_type("IMAGE") == 2 + + def test_case_insensitive(self): + assert parse_message_type("image") == 2 + assert parse_message_type("Voice") == 3 + + def test_all_types(self): + assert parse_message_type("TEXT") == 1 + assert parse_message_type("FILE") == 4 + assert parse_message_type("VIDEO") == 5 + + def test_invalid(self): + assert parse_message_type("invalid") is None + + def test_empty(self): + assert parse_message_type("") is None + + +# -- parse_time -------------------------------------------------------------- + + +class TestParseTime: + def test_unix_ms(self): + assert parse_time("1234567890") == 1234567890 + + def test_iso_utc(self): + result = parse_time("2024-01-01T00:00:00Z") + assert result == 1704067200000 + + def test_iso_offset(self): + result = parse_time("2024-01-01T08:00:00+08:00") + assert result == 1704067200000 + + def test_invalid(self): + assert parse_time("not-a-time") is None + + def test_empty(self): + assert parse_time("") is None + + +# -- process_qr_status ------------------------------------------------------ + + +class TestProcessQrStatus: + def test_confirmed(self): + resp = { + "status": "confirmed", + "bot_token": "tok123", + "baseurl": "https://example.com", + "ilink_bot_id": "bot@im.bot", + "ilink_user_id": "user@im.wechat", + } + qr = process_qr_status(resp) + assert qr.status == "confirmed" + assert isinstance(qr.bot_info, BotInfo) + assert qr.bot_info.bot_id == "bot@im.bot" + assert qr.bot_info.base_url == "https://example.com" + assert qr.bot_info.token == "tok123" + assert qr.bot_info.user_id == "user@im.wechat" + + def test_confirmed_missing_fields(self): + resp = {"status": "confirmed"} + qr = process_qr_status(resp) + assert qr.status == "confirmed" + assert qr.bot_info.bot_id == "" + assert qr.bot_info.token == "" + + def test_scaned_normalized(self): + resp = {"status": "scaned"} + qr = process_qr_status(resp) + assert qr.status == "scanned" + assert qr.bot_info is None + + def test_expired(self): + resp = {"status": "expired"} + qr = process_qr_status(resp) + assert qr.status == "expired" + assert qr.bot_info is None + + def test_wait(self): + resp = {"status": "wait"} + qr = process_qr_status(resp) + assert qr.status == "waiting" + assert qr.bot_info is None + + def test_empty_status(self): + resp = {"status": ""} + qr = process_qr_status(resp) + assert qr.status == "waiting" + + def test_missing_status(self): + resp = {} + qr = process_qr_status(resp) + assert qr.status == "waiting" + + def test_unknown_status(self): + resp = {"status": "something_new"} + qr = process_qr_status(resp) + assert qr.status == "waiting" + + def test_frozen(self): + qr = QRResult(status="waiting") + with pytest.raises(AttributeError): + qr.status = "confirmed" # type: ignore[misc] From 0230845663f029e569b16da7e2641e5a7a2e5f4f Mon Sep 17 00:00:00 2001 From: Peng Ding Date: Wed, 8 Apr 2026 20:07:08 +0800 Subject: [PATCH 2/2] fix: add type narrowing assertions for ty check --- tests/test_helpers.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_helpers.py b/tests/test_helpers.py index c2e042e..6e934a8 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -176,6 +176,7 @@ def test_confirmed_missing_fields(self): resp = {"status": "confirmed"} qr = process_qr_status(resp) assert qr.status == "confirmed" + assert qr.bot_info is not None assert qr.bot_info.bot_id == "" assert qr.bot_info.token == "" @@ -215,4 +216,4 @@ def test_unknown_status(self): def test_frozen(self): qr = QRResult(status="waiting") with pytest.raises(AttributeError): - qr.status = "confirmed" # type: ignore[misc] + qr.status = "confirmed"