479 lines
17 KiB
Python
479 lines
17 KiB
Python
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,
|
|
"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
|
|
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,
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await service.resolve_notification_link(token=token, user_id=5)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
|
|
@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"
|