stefanfeng
43bb57eacf
Add BoxIMClient with get_credentials and send_p2p_message methods, using Huihui platform MD5 signing mechanism (sorted keys, 24h timestamp, 12-char alphanumeric nonce). Includes full test coverage.
113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
"""Tests for the Box IM client (Netease Yunxin gateway wrapper)."""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_config():
|
|
return {
|
|
"HUIHUI_IM_BASE_URL": "http://192.168.1.200:60040",
|
|
"HUIHUI_APP_ID": "test_app",
|
|
"HUIHUI_ACCESS_ID": "test_access",
|
|
"HUIHUI_ACCESS_SECRET": "test_secret",
|
|
}
|
|
|
|
|
|
def _make_mock_response(json_data: dict):
|
|
"""Create a properly configured mock for httpx.Response."""
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = json_data
|
|
return mock_response
|
|
|
|
|
|
def _patch_httpx_client(json_data: dict):
|
|
"""Patch httpx.AsyncClient so that `async with httpx.AsyncClient() as c: await c.post(...)` returns json_data."""
|
|
mock_client = AsyncMock()
|
|
mock_client.post.return_value = _make_mock_response(json_data)
|
|
|
|
mock_cm = AsyncMock()
|
|
mock_cm.__aenter__.return_value = mock_client
|
|
mock_cm.__aexit__.return_value = None
|
|
|
|
return patch("httpx.AsyncClient", return_value=mock_cm)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_credentials(mock_config):
|
|
"""get_credentials should return accid and token from the gateway response."""
|
|
with _patch_httpx_client({"code": 200, "data": {"accid": "user123", "token": "tok_xyz"}}):
|
|
from services.boxim_client import BoxIMClient
|
|
|
|
client = BoxIMClient(mock_config)
|
|
result = await client.get_credentials("user123")
|
|
|
|
assert result["accid"] == "user123"
|
|
assert result["token"] == "tok_xyz"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_p2p_message_success(mock_config):
|
|
"""send_p2p_message should return True when the gateway responds with code 200."""
|
|
with _patch_httpx_client({"code": 200}):
|
|
from services.boxim_client import BoxIMClient
|
|
|
|
client = BoxIMClient(mock_config)
|
|
result = await client.send_p2p_message("owner_acc", "target_acc", "Hello")
|
|
|
|
assert result is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_p2p_message_failure(mock_config):
|
|
"""send_p2p_message should return False when the gateway responds with a non-200 code."""
|
|
with _patch_httpx_client({"code": 500, "message": "error"}):
|
|
from services.boxim_client import BoxIMClient
|
|
|
|
client = BoxIMClient(mock_config)
|
|
result = await client.send_p2p_message("owner_acc", "target_acc", "Hello")
|
|
|
|
assert result is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_credentials_returns_none_on_error(mock_config):
|
|
"""get_credentials should return None when the gateway responds with an error code."""
|
|
with _patch_httpx_client({"code": 500, "message": "user not found"}):
|
|
from services.boxim_client import BoxIMClient
|
|
|
|
client = BoxIMClient(mock_config)
|
|
result = await client.get_credentials("nonexistent")
|
|
|
|
assert result is None
|
|
|
|
|
|
def test_build_sign_params_contains_required_fields(mock_config):
|
|
"""_build_sign_params should produce appId, accessId, nonce, timestamp, signature, signType, signVersion."""
|
|
from services.boxim_client import BoxIMClient
|
|
|
|
client = BoxIMClient(mock_config)
|
|
params = client._build_sign_params({"userId": "u1"})
|
|
|
|
assert "appId" in params
|
|
assert "accessId" in params
|
|
assert "nonce" in params
|
|
assert "timestamp" in params
|
|
assert "signature" in params
|
|
assert params["signType"] == "MD5"
|
|
assert params["signVersion"] == "1.0"
|
|
assert len(params["nonce"]) == 12
|
|
|
|
|
|
def test_build_sign_params_signature_is_deterministic(mock_config):
|
|
"""Same inputs should produce valid MD5 signatures."""
|
|
from services.boxim_client import BoxIMClient
|
|
|
|
client = BoxIMClient(mock_config)
|
|
|
|
params1 = client._build_sign_params({"userId": "u1"})
|
|
params2 = client._build_sign_params({"userId": "u1"})
|
|
|
|
assert params1["signature"] is not None
|
|
assert params2["signature"] is not None
|
|
assert len(params1["signature"]) == 32 # MD5 hex length
|