ai-kefu-admin/tests/test_wall_execution_tenant.py

"""C-1(终审审查)回归锁: 墙的执行侧必须走租户管道。

修复前缺陷: conversation.py 执行侧(过滤/替换/提醒)全走模块级
DEFAULT_WALL(jiancai 出厂),runtime.compiled_wall 全库只有 AIService 的
$DEFLECTION_POOL 一处消费——auto 租户会话实际跑的是 jc 墙(多拦/替换
文案串租户)。本文件用两租户真 runtime + 脚本化假 AI 端到端断言:

- 价格墙: jc 独有 pattern([€£]\\s?\\d,auto 词表无此组)命中的句子在
  jiancai 被拦+替换(收尾来自 jc pivots 池),在 auto 放行逐字流出;
- 零号码墙: 同一邮箱违规句两租户都拦,替换句精确等于各自墙的
  no_contact["web"](jc "within 12 hours" vs auto "replies fast!");
- 提醒侧: contact_ask 触发消息的注入 context 携带各自墙的
  no_contact["web"](web 模板占位 {no_contact_web} 回填)。
"""
import asyncio

from sqlalchemy import select

from app.engine.models import Message, MessageRole
from app.engine.services.conversation import ConversationManager
from tests.conftest import build_auto_runtime, build_jiancai_runtime

# 判别句选取依据(两墙 diff 实证,勿随手改文案——判别力依赖词表差异):
# - PRICE_REPLY 撞 jc 独有的 [€£]\s?\d 组(auto 词表无),零号码两墙都不拦;
# - EMAIL_REPLY 撞零号码墙 any_email(两墙同拦,替换文案不同);
# - 末句无尾随空白 → 以 pending 尾巴形态裁决,走同一条最终检查路径。
PRICE_REPLY = "Thanks for asking. The total is €500 for one set."
EMAIL_REPLY = "Sure, you can also email me at sales@example.com anytime."
USER_MSG = "do you ship to Nigeria?"


class _ScriptedAI:
    """固定回复脚本桩: 小块 token 流出 reply(走真实 split_sentences/
    逐句门控路径),捕获注入的 context(提醒侧断言用)。"""

    def __init__(self, reply):
        self._reply = reply
        self.captured_context = None

    async def stream_chat(self, messages, context=""):
        self.captured_context = context
        for i in range(0, len(self._reply), 7):
            yield self._reply[i:i + 7]

    async def extract_info(self, messages, existing_info):
        return {}

    async def evaluate(self, profile, recent_messages):
        return {"intent_level": "pending", "reasoning": "",
                "contact_info": {}, "info_sufficient": False}


def _send(async_db, runtime, reply, user_msg=USER_MSG):
    """一轮会话 → (落库的助手回复, 注入 context)。落库文本与流出统一
    (stream_process_message 契约),取 Message 表即可代表客户所见。"""

    async def run():
        async with async_db() as db:
            ai = _ScriptedAI(reply)
            mgr = ConversationManager(db, runtime=runtime, ai_service=ai)
            conv = await mgr.create_conversation(source=None)  # standalone
            async for _ in mgr.stream_process_message(conv.session_id,
                                                      user_msg):
                pass
            msgs = (await db.execute(
                select(Message).where(Message.conversation_id == conv.id)
                .order_by(Message.created_at))).scalars().all()
            replies = [m.content for m in msgs
                       if m.role == MessageRole.assistant]
            return (replies[-1] if replies else ""), ai.captured_context
    return asyncio.run(run())


def test_price_wall_pattern_is_tenant_scoped(async_db):
    """jc 独有 [€£] 价格 pattern: jiancai 拦截替换(€ 数字不流出,收尾 =
    jc pivots 池首句);auto 放行逐字流出——同一引擎代码,墙随租户。"""
    jc = build_jiancai_runtime()
    au = build_auto_runtime()
    # 判别前提(词表趋同护栏): 该句必须只撞 jc 墙,否则本测试失去判别力
    from app.engine.services.boundary_filter import violates_boundary
    assert violates_boundary(PRICE_REPLY, wall=jc.compiled_wall) is True
    assert violates_boundary(PRICE_REPLY, wall=au.compiled_wall) is False

    jc_reply, _ = _send(async_db, jc, PRICE_REPLY)
    assert "€500" not in jc_reply
    # 已流出干净前缀("Thanks for asking.")→ 违规后接 pivot 过渡句收尾
    assert jc_reply.endswith(jc.compiled_wall.pivots["en"][0])

    au_reply, _ = _send(async_db, au, PRICE_REPLY)
    assert au_reply == PRICE_REPLY  # auto 墙无此 pattern → 整句放行


def test_no_contact_replacement_copy_from_tenant_pool(async_db):
    """零号码墙(两租户同拦同一邮箱句): 替换句精确等于各自墙的
    no_contact["web"](standalone 源+未留资 → web 变体)。修复前 auto
    会话收的是 jc 文案(12 hours),本测试即失败。"""
    jc = build_jiancai_runtime()
    au = build_auto_runtime()
    assert (jc.compiled_wall.no_contact["web"]
            != au.compiled_wall.no_contact["web"])  # 判别前提

    jc_reply, _ = _send(async_db, jc, EMAIL_REPLY)
    au_reply, _ = _send(async_db, au, EMAIL_REPLY)
    assert jc_reply == jc.compiled_wall.no_contact["web"]
    assert au_reply == au.compiled_wall.no_contact["web"]


def test_contact_ask_reminder_uses_tenant_wall_text(async_db):
    """提醒侧同锁: contact_ask 触发消息的注入 context 携带各自墙的
    no_contact["web"] 文案(web 模板 {no_contact_web} 回填),互不渗漏。"""
    jc = build_jiancai_runtime()
    au = build_auto_runtime()
    jc_web = jc.compiled_wall.no_contact["web"]
    au_web = au.compiled_wall.no_contact["web"]

    _, jc_ctx = _send(async_db, jc, "ok.", user_msg="how can I contact you?")
    _, au_ctx = _send(async_db, au, "ok.", user_msg="how can I contact you?")
    assert jc_web in jc_ctx
    assert au_web in au_ctx
    assert au_web not in jc_ctx
    assert jc_web not in au_ctx