shanxia2.0/backend/tests/test_wechat_notice_binding.py

import json
from types import SimpleNamespace

import pytest
from fastapi import HTTPException
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


class MemoryRedis:
    def __init__(self):
        self.values = {}

    def set(self, key, value, ex=None, nx=False):
        if nx and key in self.values:
            return False
        self.values[key] = value
        return True

    def get(self, key):
        return self.values.get(key)

    def delete(self, key):
        return self.values.pop(key, None)

    def eval(self, script, count, key, *args):
        if 'ARGV[1]' in script:
            if self.get(key) == args[0]:
                self.delete(key)
                return 1
            return 0
        if 'GET' in script and 'DEL' in script:
            value = self.get(key)
            self.delete(key)
            return value
        raise AssertionError('unsupported script')


@pytest.fixture
def binding(monkeypatch):
    from app.services import wechat_notice_binding as svc
    from app.models.wechat_notice import WechatBinding
    engine = create_engine('sqlite://')
    WechatBinding.__table__.create(engine)
    db = sessionmaker(bind=engine)()
    redis = MemoryRedis()
    monkeypatch.setattr(svc, 'get_sync_redis_client', lambda: redis)
    monkeypatch.setattr(svc.settings, 'WECHAT_OA_ENABLED', True)
    monkeypatch.setattr(svc.settings, 'WECHAT_OA_SECRET', 'test-only-secret')
    monkeypatch.setattr(svc.settings, 'WECHAT_OA_PUBLIC_BASE_URL', 'https://portal.example.test')
    client = SimpleNamespace(oauth_openid=lambda code: 'oa-openid', subscribed=lambda oid: True)
    monkeypatch.setattr(svc, 'client', client)
    user = {'user_id': 'u1', 'tenant_no': 't1', 'region_id': 'd1'}
    yield svc, db, redis, user, WechatBinding
    db.close()


def scan(svc, started):
    from urllib.parse import urlparse, parse_qs
    state = parse_qs(urlparse(started['authorizeUrl']).query)['state'][0]
    return state, svc.oauth_callback(state, 'code')


def test_scan_requires_original_user_confirmation_and_rejects_replay(binding):
    svc, db, redis, user, Model = binding
    started = svc.start_binding(user)
    state, scanned = scan(svc, started)
    assert scanned['status'] == 'scanned'
    assert db.query(Model).count() == 0
    status = svc.binding_session(user, started['sessionId'])
    assert 'openid' not in status
    assert status['verificationCode'] == scanned['verificationCode']
    with pytest.raises(HTTPException):
        svc.oauth_callback(state, 'code')
    svc.confirm_binding(db, user, started['sessionId'])
    assert db.query(Model).one().openid == 'oa-openid'


def test_cross_user_or_tenant_cannot_read_or_confirm(binding):
    svc, db, redis, user, Model = binding
    started = svc.start_binding(user)
    scan(svc, started)
    for other in ({**user, 'user_id': 'u2'}, {**user, 'tenant_no': 't2'}):
        with pytest.raises(HTTPException):
            svc.binding_session(other, started['sessionId'])
        with pytest.raises(HTTPException):
            svc.confirm_binding(db, other, started['sessionId'])
    assert db.query(Model).count() == 0


def test_unsubscribed_cannot_confirm(binding, monkeypatch):
    svc, db, redis, user, Model = binding
    monkeypatch.setattr(svc.client, 'subscribed', lambda oid: False)
    started = svc.start_binding(user)
    scan(svc, started)
    with pytest.raises(HTTPException):
        svc.confirm_binding(db, user, started['sessionId'])
    assert db.query(Model).count() == 0


def test_expired_replaced_session_and_unbind(binding):
    svc, db, redis, user, Model = binding
    first = svc.start_binding(user)
    second = svc.start_binding(user)
    assert svc.binding_session(user, first['sessionId'])['status'] == 'expired'
    scan(svc, second)
    svc.confirm_binding(db, user, second['sessionId'])
    svc.unbind(db, user)
    row = db.query(Model).one()
    assert row.del_flag == 1 and row.openid is None
    with pytest.raises(HTTPException):
        svc.confirm_binding(db, user, second['sessionId'])


def test_cannot_bind_same_wechat_to_another_user_in_tenant(binding):
    svc, db, redis, user, Model = binding
    started = svc.start_binding(user)
    scan(svc, started)
    svc.confirm_binding(db, user, started['sessionId'])
    other = {**user, 'user_id': 'u2'}
    started = svc.start_binding(other)
    scan(svc, started)
    with pytest.raises(HTTPException):
        svc.confirm_binding(db, other, started['sessionId'])
    assert db.query(Model).count() == 1


def test_status_refresh_checks_current_follow_state(binding, monkeypatch):
    svc, db, redis, user, Model = binding
    started = svc.start_binding(user)
    scan(svc, started)
    svc.confirm_binding(db, user, started['sessionId'])
    monkeypatch.setattr(svc.client, 'subscribed', lambda oid: False)
    assert svc.binding_status(db, user) == {'enabled': True, 'bound': True, 'subscribed': False}


def test_start_cannot_replace_session_during_confirmation(binding):
    svc, db, redis, user, Model = binding
    redis.set(svc._owner_key(user) + ':lock', 'confirming', ex=60)
    with pytest.raises(HTTPException):
        svc.start_binding(user)


def test_old_subscription_refresh_cannot_overwrite_new_binding(binding, monkeypatch):
    svc, db, redis, user, Model = binding
    started = svc.start_binding(user)
    scan(svc, started)
    svc.confirm_binding(db, user, started['sessionId'])
    def old_subscription_reply(openid):
        assert openid == 'oa-openid'
        row = db.query(Model).one()
        row.openid, row.subscribed = 'new-openid', 1
        db.commit()
        return False
    monkeypatch.setattr(svc.client, 'subscribed', old_subscription_reply)
    assert svc.binding_status(db, user)['subscribed'] is True
    assert db.query(Model).one().openid == 'new-openid'


def test_callback_instructs_mobile_return_without_binding(binding, monkeypatch):
    from app.api.wechat_notices import callback
    svc, db, _, user, Model = binding
    started = svc.start_binding(user)
    from urllib.parse import urlparse, parse_qs
    state = parse_qs(urlparse(started['authorizeUrl']).query)['state'][0]
    response = callback(state, 'code')
    body = response.body.decode()
    assert '左上角返回' in body
    assert '验证码' in body
    assert '确认绑定' in body
    assert 'oa-openid' not in body
    assert db.query(Model).count() == 0
    assert response.headers['cache-control'] == 'no-store'
    assert response.headers['referrer-policy'] == 'no-referrer'


def test_mobile_web_view_opens_business_domain_before_wechat(binding):
    from app.api.wechat_notices import oauth_launch
    from urllib.parse import parse_qs, urlparse
    svc, _, redis, user, _ = binding
    started = svc.start_binding(user, mode='mobile')
    launch = urlparse(started['authorizeUrl'])
    assert launch.scheme == 'https'
    assert launch.netloc == 'portal.example.test'
    assert launch.path == '/api/public/wechat-notices/oauth/launch'
    assert 'open.weixin.qq.com' not in started['authorizeUrl']
    state = parse_qs(launch.query)['state'][0]
    page = oauth_launch(state)
    body = page.body.decode()
    assert 'location.replace' in body
    assert 'open.weixin.qq.com/connect/oauth2/authorize' in body
    assert state in body
    assert "'unsafe-inline'" not in page.headers['content-security-policy']
    assert redis.get('wechat:oa:state:' + state)
    missing = oauth_launch('a' * 32)
    assert '过期' in missing.body.decode()
    assert 'open.weixin.qq.com' not in missing.body.decode()
    invalid = oauth_launch('javascript:alert(1)')
    assert '无效' in invalid.body.decode()
    assert 'open.weixin.qq.com' not in invalid.body.decode()


def test_mobile_callback_only_returns_to_original_authenticated_client(binding):
    from app.api.wechat_notices import callback, start, StartBinding
    from urllib.parse import urlparse, parse_qs
    svc, db, redis, user, Model = binding
    response = start(body=StartBinding(mode='mobile'), user=user)
    started = response['data']
    assert svc._load(started['sessionId'])['mode'] == 'mobile'
    state = parse_qs(urlparse(started['authorizeUrl']).query)['state'][0]
    page = callback(state, 'code')
    body = page.body.decode()
    assert '正在返回小程序' in body
    assert '记下验证码' not in body
    assert 'oa-openid' not in body
    assert started['sessionId'] not in body
    assert 'jweixin-1.6.0.js' in body
    assert 'nonce-' in page.headers['content-security-policy']
    assert "'unsafe-inline'" not in page.headers['content-security-policy']
    assert db.query(Model).count() == 0
    with pytest.raises(HTTPException):
        svc.confirm_binding(db, {**user, 'user_id': 'u2'}, started['sessionId'])
    assert svc.confirm_binding(db, user, started['sessionId'])['bound'] is True


def test_mobile_unsubscribed_returns_without_success_or_binding(binding, monkeypatch):
    from app.api.wechat_notices import callback
    from urllib.parse import urlparse, parse_qs
    svc, db, _, user, Model = binding
    monkeypatch.setattr(svc.client, 'subscribed', lambda oid: False)
    started = svc.start_binding(user, mode='mobile')
    state = parse_qs(urlparse(started['authorizeUrl']).query)['state'][0]
    page = callback(state, 'code')
    assert '请先关注' in page.body.decode()
    assert '已开通' not in page.body.decode()
    assert db.query(Model).count() == 0


def test_binding_start_mode_defaults_to_desktop_and_rejects_invalid_mode(binding):
    from app.api.wechat_notices import StartBinding, start
    from pydantic import ValidationError
    svc, _, _, user, _ = binding
    result = start(body=None, user=user)
    assert svc._load(result['data']['sessionId'])['mode'] == 'desktop'
    with pytest.raises(ValidationError):
        StartBinding(mode='https://evil.test')