119 lines
4.1 KiB
Python
119 lines
4.1 KiB
Python
"""Contract tests for the self-hosted BOXIM client."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from services.boxim_client import BoxIMClient, BoxIMError
|
|
|
|
|
|
@pytest.fixture
|
|
def config():
|
|
return {
|
|
"HUIHUI_PLATFORM_BASE_URL": "https://open.example/api",
|
|
"BOXIM_API_BASE_URL": "https://im.example/api",
|
|
"HUIHUI_APP_ID": "test_app",
|
|
"HUIHUI_ACCESS_ID": "test_access",
|
|
"HUIHUI_ACCESS_SECRET": "test_secret",
|
|
}
|
|
|
|
|
|
def _response(payload: dict, status_code: int = 200):
|
|
response = MagicMock()
|
|
response.status_code = status_code
|
|
response.json.return_value = payload
|
|
return response
|
|
|
|
|
|
def _client_patch(*, post_payload=None, request_payload=None, status_code=200):
|
|
client = AsyncMock()
|
|
if post_payload is not None:
|
|
client.post.return_value = _response(post_payload, status_code)
|
|
if request_payload is not None:
|
|
client.request.return_value = _response(request_payload, status_code)
|
|
context = AsyncMock()
|
|
context.__aenter__.return_value = client
|
|
context.__aexit__.return_value = None
|
|
return patch("services.boxim_client.httpx.AsyncClient", return_value=context), client
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exchange_access_token_uses_huihui_bearer_and_signed_form(config):
|
|
mocked, client = _client_patch(
|
|
post_payload={"code": 0, "data": {"accessToken": "box-token", "accessTokenExpiresIn": 3600}}
|
|
)
|
|
with mocked:
|
|
result = await BoxIMClient(config).exchange_access_token("huihui-token")
|
|
|
|
assert result["accessToken"] == "box-token"
|
|
call = client.post.await_args
|
|
assert call.args[0] == "https://open.example/api/im/box/netease"
|
|
assert call.kwargs["headers"]["Authorization"] == "Bearer huihui-token"
|
|
assert call.kwargs["data"]["appId"] == "test_app"
|
|
assert len(call.kwargs["data"]["signature"]) == 32
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_self_and_incremental_private_messages_use_boxim_header(config):
|
|
client_instance = BoxIMClient(config)
|
|
mocked, client = _client_patch(
|
|
request_payload={"code": 200, "data": {"id": 42, "nickName": "Owner"}}
|
|
)
|
|
with mocked:
|
|
profile = await client_instance.get_self("box-token")
|
|
assert profile["id"] == 42
|
|
assert client.request.await_args.kwargs["headers"] == {"accessToken": "box-token"}
|
|
|
|
mocked, client = _client_patch(
|
|
request_payload={"code": 200, "data": [{"id": 101, "sendId": 7, "recvId": 42}]}
|
|
)
|
|
with mocked:
|
|
messages = await client_instance.fetch_private_messages("box-token", "100")
|
|
assert messages[0]["id"] == 101
|
|
assert client.request.await_args.kwargs["params"] == {"minId": "100"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_private_message_matches_boxim_payload(config):
|
|
mocked, client = _client_patch(
|
|
request_payload={"code": 200, "data": {"id": 88, "localId": 12345}}
|
|
)
|
|
with mocked:
|
|
result = await BoxIMClient(config).send_private_message(
|
|
"box-token", "77", "你好", local_id="12345"
|
|
)
|
|
|
|
assert result["id"] == 88
|
|
call = client.request.await_args
|
|
assert call.args[:2] == ("POST", "https://im.example/api/message/private/send")
|
|
assert call.kwargs["json"] == {
|
|
"localId": 12345,
|
|
"recvId": 77,
|
|
"content": "你好",
|
|
"type": 0,
|
|
"receipt": False,
|
|
"atUserIds": [],
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_boxim_auth_error_is_explicit(config):
|
|
mocked, _ = _client_patch(
|
|
request_payload={"code": 400, "message": "未登录"}, status_code=200
|
|
)
|
|
with mocked, pytest.raises(BoxIMError) as exc_info:
|
|
await BoxIMClient(config).get_self("expired")
|
|
assert exc_info.value.auth_error is True
|
|
|
|
|
|
def test_sign_params_include_production_required_fields(config):
|
|
params = BoxIMClient(config)._build_sign_params()
|
|
assert params["appId"] == "test_app"
|
|
assert params["accessId"] == "test_access"
|
|
assert params["signType"] == "MD5"
|
|
assert params["signVersion"] == "1.0"
|
|
assert len(params["nonce"]) == 12
|
|
assert len(params["timestamp"]) == 14
|
|
assert len(params["signature"]) == 32
|
|
assert "accessSecret" not in params
|