shanxia2.0/backend/tests/test_wechat_oa_client.py

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()) == {'***'}