From 31804a78ffa86e34abebb585f8102012f23da3ab Mon Sep 17 00:00:00 2001 From: stu2not Date: Tue, 23 Jun 2026 15:21:23 +0800 Subject: [PATCH] Allow disabling WeChat MP mini program links --- .../banban/service/wechat_mp_notification.py | 2 +- talkingq-url/config.py | 1 + .../tests/test_wechat_mp_notification.py | 258 ++++++++++++++++++ 3 files changed, 260 insertions(+), 1 deletion(-) create mode 100644 talkingq-url/tests/test_wechat_mp_notification.py diff --git a/talkingq-url/banban/service/wechat_mp_notification.py b/talkingq-url/banban/service/wechat_mp_notification.py index a3f8b28..7c05af1 100644 --- a/talkingq-url/banban/service/wechat_mp_notification.py +++ b/talkingq-url/banban/service/wechat_mp_notification.py @@ -289,7 +289,7 @@ class WechatMpNotificationService(DatabaseServiceBase): }, } miniprogram_appid = (settings.wechat_mp_miniprogram_appid or "").strip() - if miniprogram_appid and page: + if settings.wechat_mp_miniprogram_link_enabled and miniprogram_appid and page: payload["miniprogram"] = { "appid": miniprogram_appid, "pagepath": page, diff --git a/talkingq-url/config.py b/talkingq-url/config.py index b4c1b48..bb8de98 100644 --- a/talkingq-url/config.py +++ b/talkingq-url/config.py @@ -117,6 +117,7 @@ class Settings(BaseSettings): default="", validation_alias="WECHAT_MP_SLEEP_MODE_TEMPLATE_ID", ) + wechat_mp_miniprogram_link_enabled: bool = Field(default=True, validation_alias="WECHAT_MP_MINIPROGRAM_LINK_ENABLED") wechat_mp_miniprogram_appid: str = Field(default="", validation_alias="WECHAT_MP_MINIPROGRAM_APPID") wechat_mp_chat_page: str = Field(default="pages/chat/detail/index", validation_alias="WECHAT_MP_CHAT_PAGE") wechat_mp_device_page: str = Field(default="pages/device/index", validation_alias="WECHAT_MP_DEVICE_PAGE") diff --git a/talkingq-url/tests/test_wechat_mp_notification.py b/talkingq-url/tests/test_wechat_mp_notification.py new file mode 100644 index 0000000..576f134 --- /dev/null +++ b/talkingq-url/tests/test_wechat_mp_notification.py @@ -0,0 +1,258 @@ +from urllib.parse import parse_qs, urlparse + +import pytest + +from banban.service.wechat_mp_notification import WechatMpNotificationService + + +def configure_enabled_wechat_mp(monkeypatch): + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_push_enabled", True) + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_app_id", "wx-service") + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_app_secret", "secret") + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_api_base_url", "https://api.weixin.qq.com") + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_template_id", "template-id") + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_leave_message_template_id", "") + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_low_battery_template_id", "") + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_sleep_mode_template_id", "") + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_miniprogram_link_enabled", True) + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_miniprogram_appid", "wx-mini") + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_chat_page", "pages/chat/detail/index") + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_device_page", "pages/device/index") + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_oauth_redirect_uri", "https://example.com/banban/wechat-mp/oauth/callback") + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_leave_message_dedup_seconds", 60) + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_low_battery_threshold", 20) + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_low_battery_recovery_threshold", 25) + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_low_battery_dedup_seconds", 21600) + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_sleep_mode_dedup_seconds", 300) + + +@pytest.mark.asyncio +async def test_notify_leave_message_sends_template_to_unique_subscribed_recipients(monkeypatch): + configure_enabled_wechat_mp(monkeypatch) + service = WechatMpNotificationService() + sent = [] + + async def fake_list_recipients(*, device_id): + assert device_id == "TalkingQ_XQSN00001007" + return ["openid-1", "openid-2"] + + async def fake_send_template_message(**kwargs): + sent.append(kwargs) + return {"ok": True, "msgid": "msg-id"} + + monkeypatch.setattr(service, "_list_device_family_mp_openids", fake_list_recipients) + monkeypatch.setattr(service, "_send_template_message", fake_send_template_message) + + await service.notify_leave_message(device_id="TalkingQ_XQSN00001007", child_name="小明") + + assert sent == [ + { + "openid": "openid-1", + "template_id": "template-id", + "title": "小明", + "condition": "收到新的设备留言", + "page": "pages/chat/detail/index", + }, + { + "openid": "openid-2", + "template_id": "template-id", + "title": "小明", + "condition": "收到新的设备留言", + "page": "pages/chat/detail/index", + }, + ] + + +@pytest.mark.asyncio +async def test_notify_leave_message_dedupes_same_device_and_openid(monkeypatch): + configure_enabled_wechat_mp(monkeypatch) + service = WechatMpNotificationService() + sent = [] + + async def fake_list_recipients(*, device_id): + del device_id + return ["openid-1"] + + async def fake_send_template_message(**kwargs): + sent.append(kwargs["openid"]) + return {"ok": True, "msgid": "msg-id"} + + monkeypatch.setattr(service, "_list_device_family_mp_openids", fake_list_recipients) + monkeypatch.setattr(service, "_send_template_message", fake_send_template_message) + + await service.notify_leave_message(device_id="TalkingQ_XQSN00001007", child_name="小明") + await service.notify_leave_message(device_id="TalkingQ_XQSN00001007", child_name="小明") + + assert sent == ["openid-1"] + + +@pytest.mark.asyncio +async def test_bind_url_persists_state_and_builds_oauth_url(monkeypatch): + configure_enabled_wechat_mp(monkeypatch) + service = WechatMpNotificationService() + states = [] + + async def fake_create_bind_state(*, state, user_id): + states.append((state, user_id)) + + monkeypatch.setattr(service, "_create_bind_state", fake_create_bind_state) + + url = await service.create_bind_url(user_id=42) + parsed = urlparse(url) + query = parse_qs(parsed.query) + + assert parsed.netloc == "open.weixin.qq.com" + assert query["appid"] == ["wx-service"] + assert query["redirect_uri"] == ["https://example.com/banban/wechat-mp/oauth/callback"] + assert query["scope"] == ["snsapi_base"] + assert query["response_type"] == ["code"] + assert len(query["state"][0]) > 20 + assert states == [(query["state"][0], 42)] + + +@pytest.mark.asyncio +async def test_bind_oauth_code_marks_unsubscribed_account(monkeypatch): + configure_enabled_wechat_mp(monkeypatch) + service = WechatMpNotificationService() + upserts = [] + + async def fake_consume_bind_state(*, state): + assert state == "state" + return 42 + + async def fake_oauth_access_token(*, code): + assert code == "code" + return {"openid": "openid-1", "unionid": "union-1"} + + async def fake_get_mp_user_info(*, openid): + assert openid == "openid-1" + return {"subscribe": 0} + + async def fake_upsert_parent_wechat_account(**kwargs): + upserts.append(kwargs) + + monkeypatch.setattr(service, "_consume_bind_state", fake_consume_bind_state) + monkeypatch.setattr(service, "_oauth_access_token", fake_oauth_access_token) + monkeypatch.setattr(service, "_get_mp_user_info", fake_get_mp_user_info) + monkeypatch.setattr(service, "_upsert_parent_wechat_account", fake_upsert_parent_wechat_account) + + user_id = await service.bind_oauth_code(code="code", state="state") + + assert user_id == 42 + assert upserts == [ + { + "user_id": 42, + "openid": "openid-1", + "unionid": "union-1", + "subscribed": 0, + } + ] + + +@pytest.mark.asyncio +async def test_template_payload_uses_service_template_fields(monkeypatch): + configure_enabled_wechat_mp(monkeypatch) + service = WechatMpNotificationService() + captured = {} + + class FakeResponse: + def raise_for_status(self): + pass + + def json(self): + return {"errcode": 0, "msgid": "msg-id"} + + class FakeAsyncClient: + def __init__(self, timeout): + self.timeout = timeout + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def post(self, url, params=None, json=None): + captured.update({"url": url, "params": params, "json": json, "timeout": self.timeout}) + return FakeResponse() + + async def fake_get_access_token(): + return "access-token" + + monkeypatch.setattr(service, "_get_access_token", fake_get_access_token) + monkeypatch.setattr("banban.service.wechat_mp_notification.httpx.AsyncClient", FakeAsyncClient) + + result = await service._send_template_message( + openid="openid-1", + template_id="template-id", + title="小明", + condition="收到新的设备留言", + page="pages/chat/detail/index", + ) + + assert result["ok"] is True + assert captured["json"] == { + "touser": "openid-1", + "template_id": "template-id", + "data": { + "thing13": {"value": "小明"}, + "thing5": {"value": "收到新的设备留言"}, + }, + "miniprogram": { + "appid": "wx-mini", + "pagepath": "pages/chat/detail/index", + }, + } + + +@pytest.mark.asyncio +async def test_template_payload_can_disable_miniprogram_link(monkeypatch): + configure_enabled_wechat_mp(monkeypatch) + monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_miniprogram_link_enabled", False) + service = WechatMpNotificationService() + captured = {} + + class FakeResponse: + def raise_for_status(self): + pass + + def json(self): + return {"errcode": 0, "msgid": "msg-id"} + + class FakeAsyncClient: + def __init__(self, timeout): + self.timeout = timeout + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def post(self, url, params=None, json=None): + captured.update({"url": url, "params": params, "json": json, "timeout": self.timeout}) + return FakeResponse() + + async def fake_get_access_token(): + return "access-token" + + monkeypatch.setattr(service, "_get_access_token", fake_get_access_token) + monkeypatch.setattr("banban.service.wechat_mp_notification.httpx.AsyncClient", FakeAsyncClient) + + result = await service._send_template_message( + openid="openid-1", + template_id="template-id", + title="小明", + condition="设备电量低于20%", + page="pages/device/index", + ) + + assert result["ok"] is True + assert captured["json"] == { + "touser": "openid-1", + "template_id": "template-id", + "data": { + "thing13": {"value": "小明"}, + "thing5": {"value": "设备电量低于20%"}, + }, + }