93 lines
3.4 KiB
Python
93 lines
3.4 KiB
Python
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from app.services.news_service import NewsPlatformService
|
|
|
|
|
|
class _Response:
|
|
def __init__(self, payload, status_code=200):
|
|
self._payload = payload
|
|
self.status_code = status_code
|
|
self.text = str(payload)
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
|
|
class _Client:
|
|
responses = []
|
|
calls = []
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
async def patch(self, url, **kwargs):
|
|
self.__class__.calls.append(("PATCH", url, kwargs))
|
|
return self.__class__.responses.pop(0)
|
|
|
|
|
|
class HuihuiProfileSyncTests(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
self.service = NewsPlatformService()
|
|
self.service._auth_url = AsyncMock(return_value="https://99hui.com/api/usercenter")
|
|
self.service._client = AsyncMock(return_value={
|
|
"appId": "app", "accessId": "access", "accessSecret": "secret",
|
|
"clientCode": "", "orgId": "",
|
|
})
|
|
self.db = SimpleNamespace(execute=AsyncMock(), commit=AsyncMock())
|
|
self.user = SimpleNamespace(
|
|
id=51, account="13721560046", platform_uid="platform-51",
|
|
nickname="黎佳怡", real_name="黎佳怡", sex=0, avatar_url="https://img/avatar.jpg",
|
|
)
|
|
_Client.calls = []
|
|
|
|
async def test_updates_current_profile_and_public_page(self):
|
|
_Client.responses = [
|
|
_Response({"code": 0, "data": True}),
|
|
_Response({"code": 0, "data": True}),
|
|
]
|
|
session = {"token": "token", "platform_uid": "platform-51"}
|
|
with patch("app.services.news_service.get_session", AsyncMock(return_value=session)), \
|
|
patch("app.services.news_service.httpx.AsyncClient", _Client):
|
|
ok, err = await self.service.update_user_profile(
|
|
self.db, self.user,
|
|
nick_name="黎佳怡", real_name="黎佳怡",
|
|
avatar="https://img/avatar.jpg",
|
|
)
|
|
|
|
self.assertTrue(ok, err)
|
|
self.assertEqual(_Client.calls[0][1], "https://99hui.com/api/usercenter/v2/users/current")
|
|
self.assertEqual(_Client.calls[1][1], "https://99hui.com/api/usercenter/users/page/platform-51")
|
|
public_params = _Client.calls[1][2]["params"]
|
|
self.assertEqual(public_params["nickName"], "黎佳怡")
|
|
self.assertEqual(public_params["icon"], "https://img/avatar.jpg")
|
|
self.db.commit.assert_awaited_once()
|
|
|
|
async def test_public_page_failure_is_not_reported_as_success(self):
|
|
_Client.responses = [
|
|
_Response({"code": 0, "data": True}),
|
|
_Response({"code": 500, "message": "page update failed"}),
|
|
]
|
|
session = {"token": "token", "platform_uid": "platform-51"}
|
|
with patch("app.services.news_service.get_session", AsyncMock(return_value=session)), \
|
|
patch("app.services.news_service.httpx.AsyncClient", _Client):
|
|
ok, err = await self.service.update_user_profile(
|
|
self.db, self.user,
|
|
nick_name="黎佳怡", avatar="https://img/avatar.jpg",
|
|
)
|
|
|
|
self.assertFalse(ok)
|
|
self.assertIn("TA的主页同步失败", err)
|
|
self.db.commit.assert_not_awaited()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|