stefanfeng
0cc43a0d8c
Prevents stale signature values from leaking into the MD5 signing calculation, matching the news_service.py pattern. Adds a test that passes a stale signature in extra params and verifies the returned signature is freshly computed.
135 lines
4.7 KiB
Python
135 lines
4.7 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_excludes_signature_and_accessSecret_from_signing_string(mock_config):
|
|
"""signature and accessSecret must be excluded from the signing string to match news_service.py."""
|
|
from services.boxim_client import BoxIMClient
|
|
|
|
client = BoxIMClient(mock_config)
|
|
|
|
# Pass params that already contain a stale "signature" value
|
|
params_with_stale_sig = client._build_sign_params({
|
|
"userId": "u1",
|
|
"signature": "OLD_STALE_SIG",
|
|
})
|
|
|
|
# The returned signature must be freshly computed (32-char MD5 uppercase),
|
|
# NOT the stale value we passed in.
|
|
assert params_with_stale_sig["signature"] != "OLD_STALE_SIG"
|
|
assert len(params_with_stale_sig["signature"]) == 32
|
|
|
|
# Calling with the same extra params but no stale signature should also work.
|
|
params_clean = client._build_sign_params({"userId": "u1"})
|
|
assert len(params_clean["signature"]) == 32
|
|
|
|
|
|
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
|