60 lines
2.5 KiB
Python
60 lines
2.5 KiB
Python
"""微信 Bot 账号隔离回归测试。"""
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlmodel import Session, SQLModel, create_engine
|
|
|
|
from app.auth import feature_permissions, required_permission
|
|
from app.main import _wechat_account_visible, wechat_bot_reply
|
|
from app.models import Agent, AppUser, WeChatBotAccount
|
|
|
|
|
|
def _user(uid: int, org: int, role: str = "editor"):
|
|
return AppUser(id=uid, username=f"u{uid}", role=role,
|
|
organization_id=org, enabled=True)
|
|
|
|
|
|
def test_wechat_accounts_use_digital_human_permission_domain():
|
|
assert required_permission("POST", "/api/wechat/bot/accounts/connect") == "agents:write"
|
|
assert required_permission("GET", "/api/wechat/bot/accounts") == "digital_humans:read"
|
|
assert feature_permissions("POST", "/api/wechat/bot/accounts/connect")[0] == "feature:digital-humans"
|
|
|
|
|
|
def test_wechat_account_visibility_is_organization_scoped():
|
|
row = WeChatBotAccount(owner_user_id=7, organization_id=10)
|
|
assert _wechat_account_visible(row, _user(7, 10))
|
|
assert not _wechat_account_visible(row, _user(7, 11))
|
|
assert not _wechat_account_visible(row, _user(8, 11, "admin"))
|
|
assert _wechat_account_visible(row, _user(8, 10, "admin"))
|
|
|
|
|
|
def test_reply_requires_account_and_never_uses_global_fallback(tmp_path):
|
|
"""没有 account_id 时即使设置了全局环境变量也必须拒绝。"""
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'wechat.db'}")
|
|
SQLModel.metadata.create_all(engine)
|
|
request = SimpleNamespace(headers={})
|
|
with Session(engine) as session:
|
|
with pytest.raises(HTTPException) as exc:
|
|
wechat_bot_reply({"text": "hello", "conversation_id": "c"}, session, request)
|
|
assert exc.value.status_code == 422
|
|
|
|
|
|
def test_reply_rejects_disconnected_account_before_agent_lookup(tmp_path):
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'wechat2.db'}")
|
|
SQLModel.metadata.create_all(engine)
|
|
request = SimpleNamespace(headers={})
|
|
with Session(engine) as session:
|
|
owner = _user(1, 1)
|
|
session.add(owner)
|
|
session.add(WeChatBotAccount(id=3, owner_user_id=1, organization_id=1,
|
|
agent_id=9, status="disconnected",
|
|
bot_secret_hash="x"))
|
|
session.commit()
|
|
with pytest.raises(HTTPException) as exc:
|
|
wechat_bot_reply({"account_id": 3, "text": "hello", "conversation_id": "c"},
|
|
session, request)
|
|
assert exc.value.status_code == 409
|
|
|