from urllib.parse import parse_qs, urlparse import pytest from fastapi import HTTPException from banban.service.wechat_mp_notification import WechatMpNotificationService, WechatMpRecipient 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_notification_page", "pages/notification/index") monkeypatch.setattr("banban.service.wechat_mp_notification.settings.wechat_mp_notification_token_expire_minutes", 10080) 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 [ WechatMpRecipient(user_id=4, openid="openid-1"), WechatMpRecipient(user_id=5, openid="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_recipients", fake_list_recipients) monkeypatch.setattr(service, "_send_template_message", fake_send_template_message) await service.notify_leave_message( device_id="TalkingQ_XQSN00001007", child_name="小明", targets=[ { "parent_user_id": 4, "conversation_id": 6, "message_id": 209, "child_id": 8, "child_name": "小明", "device_id": "TalkingQ_XQSN00001007", }, { "parent_user_id": 5, "conversation_id": 7, "message_id": 210, "child_id": 8, "child_name": "小明", "device_id": "TalkingQ_XQSN00001007", }, ], ) assert len(sent) == 2 assert sent[0] == { "openid": "openid-1", "template_id": "template-id", "title": "小明", "condition": "收到新的设备留言", "page": sent[0]["page"], } assert sent[1] == { "openid": "openid-2", "template_id": "template-id", "title": "小明", "condition": "收到新的设备留言", "page": sent[1]["page"], } assert sent[0]["page"].startswith("pages/notification/index?token=") assert sent[1]["page"].startswith("pages/notification/index?token=") assert sent[0]["page"] != sent[1]["page"] first_token = sent[0]["page"].split("token=", 1)[1] second_token = sent[1]["page"].split("token=", 1)[1] assert service._decode_notification_token(first_token)["conversation_id"] == 6 assert service._decode_notification_token(first_token)["user_id"] == 4 assert service._decode_notification_token(second_token)["conversation_id"] == 7 assert service._decode_notification_token(second_token)["user_id"] == 5 @pytest.mark.asyncio async def test_notify_leave_message_falls_back_to_chat_page_without_targets(monkeypatch): configure_enabled_wechat_mp(monkeypatch) service = WechatMpNotificationService() sent = [] async def fake_list_recipients(*, device_id): del device_id return [WechatMpRecipient(user_id=4, openid="openid-1")] async def fake_send_template_message(**kwargs): sent.append(kwargs) return {"ok": True, "msgid": "msg-id"} monkeypatch.setattr(service, "_list_device_family_mp_recipients", 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", } ] @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 [WechatMpRecipient(user_id=4, openid="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_recipients", 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_resolve_notification_link_returns_chat_detail_params(monkeypatch): configure_enabled_wechat_mp(monkeypatch) service = WechatMpNotificationService() token = service._create_notification_token( { "type": "leave_message", "user_id": 4, "conversation_id": 6, "message_id": 209, "child_id": 8, "child_name": "小明", "device_id": "TalkingQ_XQSN00001007", } ) class FakeResult: def mappings(self): return self def first(self): return { "conversation_id": 6, "message_id": 209, "child_id": 8, "child_name": "小明", "device_id": "TalkingQ_XQSN00001007", } class FakeSession: async def execute(self, sql, params): assert params["message_id"] == 209 assert params["conversation_id"] == 6 assert params["child_id"] == 8 assert params["user_id"] == 4 assert params["device_id"] == "TalkingQ_XQSN00001007" return FakeResult() async def close(self): pass async def fake_get_session(): return FakeSession() monkeypatch.setattr(service, "get_session", fake_get_session) result = await service.resolve_notification_link(token=token, user_id=4) assert result["route"] == "pages/chat/detail/index" assert result["params"] == { "id": 6, "source": "im", "name": "家长沟通", "peerKind": "parent", "roleKey": "", "conversationTypeName": "parent_child", "peerId": "4", "childId": 8, "childName": "小明", "parentUserId": 4, "deviceId": "TalkingQ_XQSN00001007", "channelLabel": "微信小程序", "canSend": "1", "messageId": 209, } @pytest.mark.asyncio async def test_resolve_notification_link_rejects_other_user(monkeypatch): configure_enabled_wechat_mp(monkeypatch) service = WechatMpNotificationService() token = service._create_notification_token( { "type": "leave_message", "user_id": 4, "conversation_id": 6, "message_id": 209, "child_id": 8, } ) class FakeResult: def mappings(self): return self def first(self): return None class FakeSession: async def execute(self, sql, params): assert params["message_id"] == 209 assert params["conversation_id"] == 6 assert params["child_id"] == 8 assert params["user_id"] == 5 return FakeResult() async def close(self): pass async def fake_get_session(): return FakeSession() monkeypatch.setattr(service, "get_session", fake_get_session) with pytest.raises(HTTPException) as exc_info: await service.resolve_notification_link(token=token, user_id=5) assert exc_info.value.status_code == 404 @pytest.mark.asyncio async def test_resolve_notification_link_redirects_to_current_family_member_message(monkeypatch): configure_enabled_wechat_mp(monkeypatch) service = WechatMpNotificationService() token = service._create_notification_token( { "type": "leave_message", "user_id": 4, "conversation_id": 6, "message_id": 209, "child_id": 8, "child_name": "小明", "device_id": "TalkingQ_XQSN00001007", } ) class FakeResult: def mappings(self): return self def first(self): return { "conversation_id": 22, "message_id": 216, "child_id": 8, "child_name": "小明", "device_id": "TalkingQ_XQSN00001007", } class FakeSession: async def execute(self, sql, params): assert params["message_id"] == 209 assert params["conversation_id"] == 6 assert params["child_id"] == 8 assert params["user_id"] == 3 assert params["user_id_str"] == "3" assert params["device_id"] == "TalkingQ_XQSN00001007" return FakeResult() async def close(self): pass async def fake_get_session(): return FakeSession() monkeypatch.setattr(service, "get_session", fake_get_session) result = await service.resolve_notification_link(token=token, user_id=3) assert result["route"] == "pages/chat/detail/index" assert result["params"]["id"] == 22 assert result["params"]["parentUserId"] == 3 assert result["params"]["peerId"] == "3" assert result["params"]["messageId"] == 216 @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(*, force_refresh=False): assert force_refresh is False 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(*, force_refresh=False): assert force_refresh is False 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%"}, }, } @pytest.mark.asyncio async def test_template_send_refreshes_stable_token_once_when_cached_token_is_invalid(monkeypatch): configure_enabled_wechat_mp(monkeypatch) service = WechatMpNotificationService() service._access_token = "cached-token" service._access_token_expire_at = 9999999999 calls = [] class FakeResponse: def __init__(self, payload): self._payload = payload def raise_for_status(self): pass def json(self): return self._payload 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): calls.append({"url": url, "params": params, "json": json}) if url.endswith("/cgi-bin/message/template/send") and params == {"access_token": "cached-token"}: return FakeResponse( { "errcode": 40001, "errmsg": "invalid credential, access_token is invalid or not latest", } ) if url.endswith("/cgi-bin/stable_token"): assert json == { "appid": "wx-service", "secret": "secret", "grant_type": "client_credential", "force_refresh": True, } return FakeResponse({"access_token": "fresh-token", "expires_in": 7200}) if url.endswith("/cgi-bin/message/template/send") and params == {"access_token": "fresh-token"}: return FakeResponse({"errcode": 0, "msgid": "msg-id"}) return FakeResponse({"errcode": 99999, "errmsg": "unexpected request"}) 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 == {"errcode": 0, "msgid": "msg-id", "ok": True} assert [call["params"] for call in calls if call["url"].endswith("/cgi-bin/message/template/send")] == [ {"access_token": "cached-token"}, {"access_token": "fresh-token"}, ] assert service._access_token == "fresh-token"