46 lines
2.3 KiB
Python
46 lines
2.3 KiB
Python
import pytest
|
|
import requests
|
|
|
|
|
|
def test_client_uses_service_account_and_exact_template(monkeypatch):
|
|
from app.services import wechat_oa_client as svc
|
|
monkeypatch.setattr(svc.settings, 'WECHAT_OA_APPID', 'service-app')
|
|
monkeypatch.setattr(svc.settings, 'WECHAT_OA_SECRET', 'test-secret')
|
|
monkeypatch.setattr(svc.settings, 'WECHAT_OA_MINI_APPID', 'mini-app')
|
|
calls = []
|
|
class Response:
|
|
def __init__(self, data): self.data = data
|
|
def raise_for_status(self): pass
|
|
def json(self): return self.data
|
|
def request(method, url, **kwargs):
|
|
calls.append((method, url, kwargs))
|
|
return Response({'access_token': 'test-token', 'expires_in': 7200} if url.endswith('stable_token') else {'errcode': 0, 'msgid': 123})
|
|
monkeypatch.setattr(svc.requests, 'request', request)
|
|
client = svc.WechatOAClient()
|
|
for _ in range(2):
|
|
assert client.send(openid='service-openid', template_id='template', data={'const5': {'value': '通过'}}, event_key='key', page='pages/example') == '123'
|
|
assert len(calls) == 3
|
|
assert calls[0][2]['json']['appid'] == 'service-app'
|
|
assert calls[1][2]['json']['touser'] == 'service-openid'
|
|
assert calls[1][2]['json']['miniprogram']['appid'] == 'mini-app'
|
|
|
|
|
|
def test_transport_timeout_does_not_expose_url_or_secret(monkeypatch):
|
|
from app.services import wechat_oa_client as svc
|
|
client = svc.WechatOAClient()
|
|
monkeypatch.setattr(client, 'access_token', lambda *args: 'sensitive-token')
|
|
def fail(*args, **kwargs):
|
|
raise requests.Timeout('https://api.weixin.qq.com/?secret=TOP_SECRET')
|
|
monkeypatch.setattr(svc.requests, 'request', fail)
|
|
with pytest.raises(svc.WechatDeliveryUnknown) as exc:
|
|
client.send(openid='oid', template_id='template', data={}, event_key='key')
|
|
assert 'TOP_SECRET' not in str(exc.value)
|
|
|
|
|
|
def test_notice_audit_masks_oauth_state_code_and_binding_fields():
|
|
from app.core.api_request_log_middleware import _redact_notice_query, _redact_obj
|
|
query = _redact_notice_query('/api/public/wechat-notices/oauth/callback', 'code=secret-code&state=secret-state')
|
|
assert 'secret-code' not in query and 'secret-state' not in query
|
|
data = _redact_obj({'sessionId': 'private', 'authorizeUrl': 'private', 'openid': 'private'})
|
|
assert set(data.values()) == {'***'}
|