集成家长短信通知

talkingq-url/tests/test_parent_phone_authorization.py

@@ -0,0 +1,130 @@
+from dataclasses import dataclass
+
+import httpx
+import pytest
+
+from banban.service.parent import ParentService
+from banban.service.wechat_login import WechatAuthService
+
+
+class FakeSession:
+    async def close(self):
+        pass
+
+
+@dataclass(frozen=True)
+class FakeWechatPhone:
+    phone_number: str
+    pure_phone_number: str
+    country_code: str | None = None
+
+
+class FakeWechatAuthService:
+    def __init__(self):
+        self.codes = []
+
+    async def exchange_phone_code(self, code):
+        self.codes.append(code)
+        return FakeWechatPhone(
+            phone_number="+8613800138000",
+            pure_phone_number="13800138000",
+            country_code="86",
+        )
+
+
+@pytest.mark.asyncio
+async def test_update_phone_from_wechat_code_persists_current_parent_phone(monkeypatch):
+    service = ParentService()
+    wechat_auth_service = FakeWechatAuthService()
+    updates = []
+
+    async def fake_get_session():
+        return FakeSession()
+
+    async def fake_update(self, user_id, nickname=None, avatar_url=None, phone=None):
+        updates.append(
+            {
+                "user_id": user_id,
+                "nickname": nickname,
+                "avatar_url": avatar_url,
+                "phone": phone,
+            }
+        )
+
+    async def fake_get(user_id):
+        return {
+            "user_id": user_id,
+            "openid": "openid_demo",
+            "unionid": None,
+            "nickname": "家长",
+            "avatar_url": None,
+            "phone": "13800138000",
+            "status": 1,
+        }
+
+    monkeypatch.setattr(service, "get_session", fake_get_session)
+    monkeypatch.setattr("banban.service.parent.ParentDAO.update", fake_update)
+    monkeypatch.setattr(service, "get", fake_get)
+
+    parent = await service.update_phone_from_wechat_code(
+        user_id=9,
+        code="phone-code-demo",
+        wechat_auth_service=wechat_auth_service,
+    )
+
+    assert wechat_auth_service.codes == ["phone-code-demo"]
+    assert updates == [
+        {
+            "user_id": 9,
+            "nickname": None,
+            "avatar_url": None,
+            "phone": "13800138000",
+        }
+    ]
+    assert parent["phone"] == "13800138000"
+
+
+@pytest.mark.asyncio
+async def test_exchange_phone_code_reuses_cached_access_token(monkeypatch):
+    requests = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append((request.method, request.url.path, str(request.url.params)))
+        if request.url.path == "/cgi-bin/token":
+            return httpx.Response(200, json={"access_token": "wechat-access-token", "expires_in": 7200})
+        if request.url.path == "/wxa/business/getuserphonenumber":
+            return httpx.Response(
+                200,
+                json={
+                    "errcode": 0,
+                    "phone_info": {
+                        "phoneNumber": "+8613800138000",
+                        "purePhoneNumber": "13800138000",
+                        "countryCode": "86",
+                    },
+                },
+            )
+        return httpx.Response(404, json={"errcode": 404})
+
+    monkeypatch.setattr(WechatAuthService, "_cached_access_token", None)
+    monkeypatch.setattr(WechatAuthService, "_cached_access_token_expires_at", 0)
+    monkeypatch.setattr("banban.service.wechat_login.settings.wechat_app_id", "wechat-app-id")
+    monkeypatch.setattr("banban.service.wechat_login.settings.wechat_app_secret", "wechat-app-secret")
+    monkeypatch.setattr("banban.service.wechat_login.settings.wechat_api_base_url", "https://api.weixin.qq.com")
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(transport=transport, base_url="https://api.weixin.qq.com") as client:
+        uncached_service = WechatAuthService()
+        uncached_service._request_access_token = WechatAuthService(client)._request_access_token
+        uncached_service._request_phone_number = WechatAuthService(client)._request_phone_number
+
+        first = await uncached_service.exchange_phone_code("phone-code-1")
+        second = await uncached_service.exchange_phone_code("phone-code-2")
+
+    assert first.pure_phone_number == "13800138000"
+    assert second.pure_phone_number == "13800138000"
+    token_requests = [item for item in requests if item[1] == "/cgi-bin/token"]
+    phone_requests = [item for item in requests if item[1] == "/wxa/business/getuserphonenumber"]
+    assert len(token_requests) == 1
+    assert len(phone_requests) == 2
+    assert all("access_token=wechat-access-token" in item[2] for item in phone_requests)