From dd371ec4cc5411273c3079d26fe5ee8dd2b0c79f Mon Sep 17 00:00:00 2001 From: FlameSky Date: Sun, 18 Oct 2026 06:56:30 +0800 Subject: [PATCH] fix(wechat): refresh stale or rejected access_token (#7053) --- app/modules/wechat/wechat.py | 9 +- tests/test_wechat_access_token.py | 139 ++++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+), 4 deletions(-) create mode 100644 tests/test_wechat_access_token.py diff --git a/app/modules/wechat/wechat.py b/app/modules/wechat/wechat.py index 3a15128ac..bab24f98d 100644 --- a/app/modules/wechat/wechat.py +++ b/app/modules/wechat/wechat.py @@ -114,7 +114,7 @@ class WeChat: if not self._access_token: token_flag = False else: - if (datetime.now() - self._access_token_time).seconds >= self._expires_in: + if (datetime.now() - self._access_token_time).total_seconds() >= self._expires_in: token_flag = False if not token_flag or force: @@ -647,7 +647,8 @@ class WeChat: """ 提交企业微信 JSON 请求并处理令牌过期。 - 错误码 42001 会刷新令牌并触发重试；无响应和非 200 响应抛出异常。 + 错误码 40014、41001、42001 均表示令牌不可用，会强制刷新令牌并触发重试； + 无响应和非 200 响应抛出异常。 :param url: 可包含 access_token 占位符的接口地址 :param req_json: 需要序列化为 JSON 的请求体 @@ -669,9 +670,9 @@ class WeChat: if ret_json.get("errcode") == 0: return True else: - if ret_json.get("errcode") == 42001: + if ret_json.get("errcode") in (40014, 41001, 42001): self.__get_access_token(force=True) - error_msg = (f"access_token 已过期，尝试重新获取 access_token," + error_msg = (f"access_token 已失效，尝试重新获取 access_token," f"errcode: {ret_json.get('errcode')}, errmsg: {ret_json.get('errmsg')}") raise RetryException(error_msg) else: diff --git a/tests/test_wechat_access_token.py b/tests/test_wechat_access_token.py new file mode 100644 index 000000000..626103088 --- /dev/null +++ b/tests/test_wechat_access_token.py @@ -0,0 +1,139 @@ +""" +验证企业微信 access_token 的有效期判断，以及令牌失效后的强制重取边界。 +""" + +import time +from datetime import datetime, timedelta +from types import SimpleNamespace +from unittest.mock import Mock, patch + +import pytest + +from app.modules.wechat.wechat import WeChat +from app.runtime import execution + +_PROXY = "http://proxy.local" +_SEND_URL_PREFIX = f"{_PROXY}/cgi-bin/message/send?access_token=" + +_CONFIG = { + "name": "wechat-test", + "WECHAT_CORPID": "corpid", + "WECHAT_APP_SECRET": "secret", + "WECHAT_APP_ID": "1000002", + "WECHAT_PROXY": _PROXY, +} + + +def _token_response(token: str) -> Mock: + """ + 构造 gettoken 响应，字段与企业微信返回保持一致。 + + :param token: 响应中携带的访问令牌 + :return: 行为等价于 200 响应的替身对象 + """ + response = Mock() + response.status_code = 200 + response.json.return_value = { + "errcode": 0, + "errmsg": "ok", + "access_token": token, + "expires_in": 7200, + } + return response + + +def _send_response(errcode: int) -> Mock: + """ + 构造 message/send 响应，只保留判定需要的错误码。 + + :param errcode: 企业微信接口返回的错误码 + :return: 行为等价于 200 响应的替身对象 + """ + response = Mock() + response.status_code = 200 + response.json.return_value = {"errcode": errcode, "errmsg": f"errcode-{errcode}"} + return response + + +def _create_client(request_utils: Mock, token: str = "token-1") -> WeChat: + """ + 在替身 HTTP 层上创建客户端，避免构造过程发出真实请求。 + + :param request_utils: 被替换的 RequestUtils 类替身 + :param token: 构造阶段 gettoken 返回的令牌 + :return: 已完成初始化的企业微信客户端 + """ + request_utils.return_value.get_res.return_value = _token_response(token) + request_utils.return_value.post.return_value = _send_response(0) + return WeChat(**_CONFIG) + + +def _skip_retry_backoff(monkeypatch: pytest.MonkeyPatch) -> None: + """ + 去掉重试装饰器的退避等待，让用例只验证重试次数与结果。 + + :param monkeypatch: pytest 提供的属性替换夹具 + """ + monkeypatch.setattr( + execution, + "time", + SimpleNamespace(sleep=lambda *_: None, time=time.time, monotonic=time.monotonic), + ) + + +def test_stale_token_is_refreshed_before_send() -> None: + """ + 令牌空闲数天后发送消息时必须先重取，不受 timedelta.seconds 按天回绕的影响。 + """ + with patch("app.modules.wechat.wechat.RequestUtils") as request_utils: + client = _create_client(request_utils) + # 8 天 1 小时 43 分：timedelta.seconds 只剩 6180 秒，小于有效期 7200 秒 + client._access_token = "token-stale" + client._access_token_time = datetime.now() - timedelta(days=8, hours=1, minutes=43) + client._expires_in = 7200 + request_utils.return_value.get_res.return_value = _token_response("token-fresh") + + assert client.send_msg(title="标题", text="内容", userid="user-1") is True + + posted_url = request_utils.return_value.post.call_args.kwargs["url"] + assert posted_url.startswith(_SEND_URL_PREFIX) + assert posted_url == f"{_SEND_URL_PREFIX}token-fresh" + assert request_utils.return_value.get_res.call_count == 2 + + +@pytest.mark.parametrize("errcode", [40014, 41001, 42001]) +def test_invalid_token_is_refreshed_and_sent_again( + monkeypatch: pytest.MonkeyPatch, errcode: int +) -> None: + """ + 接口返回令牌失效类错误码时应强制重取令牌并重试发送。 + + :param monkeypatch: pytest 提供的属性替换夹具 + :param errcode: 企业微信返回的令牌失效类错误码 + """ + _skip_retry_backoff(monkeypatch) + with patch("app.modules.wechat.wechat.RequestUtils") as request_utils: + client = _create_client(request_utils) + post = request_utils.return_value.post + post.side_effect = [_send_response(errcode), _send_response(0)] + request_utils.return_value.get_res.return_value = _token_response("token-fresh") + + assert client.send_msg(title="标题", text="内容", userid="user-1") is True + + assert post.call_count == 2 + assert post.call_args.kwargs["url"] == f"{_SEND_URL_PREFIX}token-fresh" + assert request_utils.return_value.get_res.call_count == 2 + + +def test_other_errors_do_not_refresh_token() -> None: + """ + 非令牌类错误码只记录失败，不触发令牌重取。 + """ + with patch("app.modules.wechat.wechat.RequestUtils") as request_utils: + client = _create_client(request_utils) + request_utils.return_value.post.return_value = _send_response(40003) + + assert client.send_msg(title="标题", text="内容", userid="user-1") is False + + assert request_utils.return_value.post.call_count == 1 + assert request_utils.return_value.get_res.call_count == 1