diff --git a/nanobot/channels/weixin/connect.py b/nanobot/channels/weixin/connect.py index bfb448e70..eb9d2e8a8 100644 --- a/nanobot/channels/weixin/connect.py +++ b/nanobot/channels/weixin/connect.py @@ -22,6 +22,7 @@ class WeixinConnectSession: channel: WeixinChannel current_poll_base_url: str refresh_count: int + force: bool created_wall: float deadline: float last_error: str | None = None @@ -72,7 +73,7 @@ class WeixinConnectStore: channel.connect_open_client() try: - qrcode_id, qr_url = await channel.connect_fetch_qr_code() + qrcode_id, qr_url = await channel.connect_fetch_qr_code(force=force) except Exception as exc: await self._close_channel(channel) raise ChannelConnectError( @@ -89,6 +90,7 @@ class WeixinConnectStore: channel=channel, current_poll_base_url=channel.connect_base_url, refresh_count=0, + force=force, created_wall=now_wall, deadline=time.monotonic() + 600, ) @@ -187,7 +189,7 @@ class WeixinConnectStore: } try: session.qrcode_id, session.qr_url = ( - await session.channel.connect_fetch_qr_code() + await session.channel.connect_fetch_qr_code(force=session.force) ) except Exception as exc: self._sessions.pop(session_id, None) @@ -204,6 +206,17 @@ class WeixinConnectStore: ) if status == "binded_redirect": + if session.force: + self._sessions.pop(session_id, None) + await self._close_channel(session.channel) + return { + "session_id": session_id, + "status": "failed", + "message": ( + "Unable to complete a new WeChat login. " + "Start again and scan with the account you want to connect." + ), + } if not session.channel.connect_load_state(): self._sessions.pop(session_id, None) await self._close_channel(session.channel) @@ -234,7 +247,7 @@ class WeixinConnectStore: } try: session.qrcode_id, session.qr_url = ( - await session.channel.connect_fetch_qr_code() + await session.channel.connect_fetch_qr_code(force=session.force) ) except Exception as exc: self._sessions.pop(session_id, None) diff --git a/nanobot/channels/weixin/runtime.py b/nanobot/channels/weixin/runtime.py index a8bd18e2c..366ecd895 100644 --- a/nanobot/channels/weixin/runtime.py +++ b/nanobot/channels/weixin/runtime.py @@ -726,9 +726,9 @@ class WeixinChannel(BaseChannel): break return tokens - async def _fetch_qr_code(self) -> tuple[str, str]: - """Fetch a fresh QR code. Returns (qrcode_id, scan_url).""" - local_tokens = self._local_token_list() + async def _fetch_qr_code(self, *, force: bool = False) -> tuple[str, str]: + """Fetch a QR code without existing credentials when forced.""" + local_tokens = [] if force else self._local_token_list() data = await self._api_post( "ilink/bot/get_bot_qrcode?bot_type=3", {"local_token_list": local_tokens}, @@ -755,11 +755,11 @@ class WeixinChannel(BaseChannel): raise RuntimeError(f"Failed to get QR code from WeChat API: {data}") return qrcode_id, (qrcode_img_content or qrcode_id) - async def _qr_login(self) -> bool: - """Perform QR code login flow. Returns True on success.""" + async def _qr_login(self, *, force: bool = False) -> bool: + """Perform QR login; forced flows accept only newly confirmed credentials.""" try: refresh_count = 0 - qrcode_id, scan_url = await self._fetch_qr_code() + qrcode_id, scan_url = await self._fetch_qr_code(force=force) self._print_qr_code(scan_url) current_poll_base_url = self.config.base_url verify_code = "" @@ -825,11 +825,16 @@ class WeixinChannel(BaseChannel): if refresh_count > MAX_QR_REFRESH_COUNT: self.logger.warning("WeChat verification failed too many times") return False - qrcode_id, scan_url = await self._fetch_qr_code() + qrcode_id, scan_url = await self._fetch_qr_code(force=force) current_poll_base_url = self.config.base_url self._print_qr_code(scan_url) continue elif status == "binded_redirect": + if force: + self.logger.error( + "Forced WeChat login returned an existing binding without new credentials" + ) + return False if self._token or self._load_state(): self.logger.info("WeChat account is already connected") return True @@ -846,7 +851,7 @@ class WeixinChannel(BaseChannel): MAX_QR_REFRESH_COUNT, ) return False - qrcode_id, scan_url = await self._fetch_qr_code() + qrcode_id, scan_url = await self._fetch_qr_code(force=force) current_poll_base_url = self.config.base_url verify_code = "" self._print_qr_code(scan_url) @@ -893,8 +898,8 @@ class WeixinChannel(BaseChannel): self._client = self._new_http_client(httpx.Timeout(60, connect=30)) self._running = True - async def connect_fetch_qr_code(self) -> tuple[str, str]: - return await self._fetch_qr_code() + async def connect_fetch_qr_code(self, *, force: bool = False) -> tuple[str, str]: + return await self._fetch_qr_code(force=force) async def connect_poll_qr_code( self, @@ -954,7 +959,7 @@ class WeixinChannel(BaseChannel): self._client = self._new_http_client(httpx.Timeout(60, connect=30)) self._running = True # Enable polling loop in _qr_login() try: - return await self._qr_login() + return await self._qr_login(force=force) finally: self._running = False if self._client: diff --git a/nanobot/channels/weixin/tests/test_connect.py b/nanobot/channels/weixin/tests/test_connect.py index c32edd187..511c1881c 100644 --- a/nanobot/channels/weixin/tests/test_connect.py +++ b/nanobot/channels/weixin/tests/test_connect.py @@ -25,7 +25,9 @@ async def test_weixin_connect_store_saves_confirmed_qr_login( ) monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path) - async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]: + async def fake_fetch_qr_code( + self: WeixinChannel, **_kwargs: Any + ) -> tuple[str, str]: return "qr-1", "https://qr.example/1" async def fake_api_get_with_base( @@ -86,14 +88,31 @@ async def test_weixin_reconnect_keeps_existing_account_until_scan_succeeds( ) monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path) - async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]: - return "qr-reconnect", "https://qr.example/reconnect" + observed_force: list[bool] = [] + + async def fake_fetch_qr_code( + self: WeixinChannel, + *, + force: bool = False, + ) -> tuple[str, str]: + observed_force.append(force) + return f"qr-reconnect-{len(observed_force)}", "https://qr.example/reconnect" + + async def fake_api_get_with_base( + self: WeixinChannel, + **_kwargs: Any, + ) -> dict[str, str]: + return {"status": "expired"} monkeypatch.setattr(WeixinChannel, "_fetch_qr_code", fake_fetch_qr_code) + monkeypatch.setattr(WeixinChannel, "_api_get_with_base", fake_api_get_with_base) store = WeixinConnectStore() started = await store.start(force=True) + refreshed = await store.poll(started["session_id"]) + assert refreshed["status"] == "pending" + assert observed_force == [True, True] assert json.loads(state_file.read_text(encoding="utf-8")) == existing cancelled = await store.cancel(started["session_id"]) assert cancelled["status"] == "cancelled" @@ -116,7 +135,9 @@ async def test_weixin_cancel_wins_over_inflight_confirmation( poll_started = asyncio.Event() release_poll = asyncio.Event() - async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]: + async def fake_fetch_qr_code( + self: WeixinChannel, **_kwargs: Any + ) -> tuple[str, str]: return "qr-cancel", "https://qr.example/cancel" async def fake_api_get_with_base( @@ -162,7 +183,9 @@ async def test_weixin_connect_store_handles_verification_code( ) monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path) - async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]: + async def fake_fetch_qr_code( + self: WeixinChannel, **_kwargs: Any + ) -> tuple[str, str]: return "qr-verify", "https://qr.example/verify" responses = [ @@ -204,7 +227,7 @@ async def test_weixin_connect_store_handles_verification_code( @pytest.mark.asyncio -async def test_weixin_connect_store_treats_existing_binding_as_success( +async def test_weixin_connect_store_rejects_existing_binding_during_forced_login( tmp_path, monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -221,7 +244,12 @@ async def test_weixin_connect_store_treats_existing_binding_as_success( ) monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path) - async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]: + async def fake_fetch_qr_code( + self: WeixinChannel, + *, + force: bool = False, + ) -> tuple[str, str]: + assert force is True return "qr-existing", "https://qr.example/existing" async def fake_api_get_with_base( @@ -237,8 +265,8 @@ async def test_weixin_connect_store_treats_existing_binding_as_success( started = await store.start(force=True) completed = await store.poll(started["session_id"]) - assert completed["status"] == "succeeded" - assert "already connected" in completed["message"] + assert completed["status"] == "failed" + assert "new WeChat login" in completed["message"] assert json.loads((state_dir / "account.json").read_text())["token"] == "working-token" @@ -255,7 +283,9 @@ async def test_weixin_connect_store_rejects_existing_binding_without_local_crede ) monkeypatch.setattr("nanobot.config.loader._current_config_path", config_path) - async def fake_fetch_qr_code(self: WeixinChannel) -> tuple[str, str]: + async def fake_fetch_qr_code( + self: WeixinChannel, **_kwargs: Any + ) -> tuple[str, str]: return "qr-missing", "https://qr.example/missing" async def fake_api_get_with_base( @@ -268,7 +298,7 @@ async def test_weixin_connect_store_rejects_existing_binding_without_local_crede monkeypatch.setattr(WeixinChannel, "_api_get_with_base", fake_api_get_with_base) store = WeixinConnectStore() - started = await store.start(force=True) + started = await store.start(force=False) completed = await store.poll(started["session_id"]) assert completed["status"] == "failed" diff --git a/nanobot/channels/weixin/tests/test_weixin_channel.py b/nanobot/channels/weixin/tests/test_weixin_channel.py index 667e2a236..cac4e1d55 100644 --- a/nanobot/channels/weixin/tests/test_weixin_channel.py +++ b/nanobot/channels/weixin/tests/test_weixin_channel.py @@ -197,34 +197,54 @@ def test_save_state_with_empty_runtime_token_preserves_persisted_account(tmp_pat @pytest.mark.asyncio -async def test_login_force_does_not_short_circuit_on_persisted_account(tmp_path) -> None: +async def test_login_force_ignores_persisted_account_through_qr_flow(tmp_path) -> None: + persisted = { + "token": "persisted-token", + "get_updates_buf": "persisted-cursor", + "context_tokens": {"wx-user": "ctx-persisted"}, + "typing_tickets": {"wx-user": {"ticket": "ticket-persisted"}}, + "base_url": "https://persisted.example", + } channel = WeixinChannel( - WeixinConfig(enabled=True, allow_from=["*"], state_dir=str(tmp_path)), + WeixinConfig( + enabled=True, + allow_from=["*"], + token="configured-token", + state_dir=str(tmp_path), + ), MessageBus(), ) (tmp_path / "account.json").write_text( - json.dumps( - { - "token": "persisted-token", - "get_updates_buf": "persisted-cursor", - "context_tokens": {"wx-user": "ctx-persisted"}, - "typing_tickets": {"wx-user": {"ticket": "ticket-persisted"}}, - "base_url": "https://persisted.example", - } - ), + json.dumps(persisted), encoding="utf-8", ) - channel._qr_login = AsyncMock(return_value=False) + channel._print_qr_code = lambda _url: None + channel._api_post = AsyncMock( + side_effect=[ + {"qrcode": "qr-1", "qrcode_img_content": "url-1"}, + {"qrcode": "qr-2", "qrcode_img_content": "url-2"}, + ] + ) + channel._api_get_with_base = AsyncMock( + side_effect=[ + {"status": "expired"}, + {"status": "binded_redirect"}, + ] + ) ok = await channel.login(force=True) assert ok is False - channel._qr_login.assert_awaited_once() + assert [call.args[1]["local_token_list"] for call in channel._api_post.await_args_list] == [ + [], + [], + ] assert channel._token == "" assert channel._get_updates_buf == "" assert channel._context_tokens == {} assert channel._typing_tickets == {} assert channel.config.base_url == "https://ilinkai.weixin.qq.com" + assert json.loads((tmp_path / "account.json").read_text()) == persisted @pytest.mark.asyncio