131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
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)
|