ai-kefu-admin/tests/test_touchstone.py

"""试金石(wall/test,Phase 2 Task 6):逐句墙判定预览。

复用 test_wall_execution_tenant 的同款姿势——boundary_filter 纯函数 +
jc/auto 两墙的词表差异做判别。判定优先级与 conversation.py 执行侧一致:
价格墙 → 零号码墙 → 再索要墙(仅已留资)。

红线:不落库、不 bump 版本(reset/编辑器共用的预览入口)。
"""
from app.engine.services import boundary_filter
from app.platform.tenant_source import FileTenantSource
from tests.conftest import SEED_DIR
from tests.test_tenant_admin_api import _auth, _create

WALL_TEST = "/api/platform/tenants/{slug}/wall/test"

# 判别句选取依据(两墙 diff 实证,勿随手改文案——判别力依赖词表差异):
# - PRICE 句撞 jc 独有的 [€£]\s?\d 组(auto 词表无此组);
# - EMAIL 句撞零号码墙 any_email(替换句 = jc no_contact["web"]);
# - CLEAN 句三墙全放行。
PRICE = "The total is €500 for one set."
EMAIL = "Sure, you can also email me at sales@example.com anytime."
CLEAN = "We ship worldwide every day."
REASK = "What's your WhatsApp number?"

AUTO_WALL = FileTenantSource(SEED_DIR).read_tenant("auto").snapshot["wall"]
JC_WALL = FileTenantSource(SEED_DIR).read_tenant("jiancai").snapshot["wall"]


def _post(client, slug, payload):
    return client.post(WALL_TEST.format(slug=slug),
                       headers=_auth(client), json=payload)


def _config_version(client, slug):
    r = client.get(f"/api/platform/tenants/{slug}/config",
                   headers=_auth(client))
    assert r.status_code == 200
    return r.json()["version"]


def test_touchstone_verdicts(super_client):
    _create(super_client, slug="demo")
    version_before = _config_version(super_client, "demo")

    r = _post(super_client, "demo", {"sentences": [PRICE, EMAIL, CLEAN]})
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["wall_source"] == "current"
    results = body["results"]
    assert len(results) == 3

    # 价格墙: blocked + 替换话术来自 jc deflections 池(en 首句)
    assert results[0]["action"] == "blocked"
    assert results[0]["reason"] == "boundary"
    assert results[0]["replacement"] == boundary_filter.DEFLECTIONS_EN[0]
    # 零号码墙: replaced + 替换句精确等于 jc no_contact["web"]
    assert results[1]["action"] == "replaced"
    assert results[1]["reason"] == "no_contact"
    assert results[1]["replacement"] == boundary_filter.NO_CONTACT_WEB
    # 干净句: pass,无替换
    assert results[2]["action"] == "pass"
    assert results[2]["reason"] is None
    assert results[2]["replacement"] is None

    # 不落库不 bump 版本
    assert _config_version(super_client, "demo") == version_before


def test_touchstone_has_contact_and_source_variants(super_client):
    """已留资/app 渠道两开关照搬执行侧语义:替换句换池/换变体,再索要墙启用。"""
    _create(super_client, slug="demo")
    # has_contact=True: 价格墙替换句来自 deflections_contact 池
    r = _post(super_client, "demo", {"sentences": [PRICE, EMAIL, REASK],
                                     "has_contact": True})
    results = r.json()["results"]
    assert results[0]["replacement"] == boundary_filter.DEFLECTIONS_EN_CONTACT[0]
    # 零号码墙替换句换 already 变体(已留资不再索要)
    assert results[1]["replacement"] == boundary_filter.NO_CONTACT_ALREADY
    # 再索要句被丢(has_contact 才启用): blocked 无替换
    assert results[2]["action"] == "blocked"
    assert results[2]["reason"] == "reask"
    assert results[2]["replacement"] is None

    # source=app: 零号码墙替换句用 app 变体
    r = _post(super_client, "demo", {"sentences": [EMAIL], "source": "app"})
    assert r.json()["results"][0]["replacement"] == boundary_filter.NO_CONTACT_APP
    # has_contact=False(默认): 再索要句放行(墙未启用)
    r = _post(super_client, "demo", {"sentences": [REASK]})
    assert r.json()["results"][0]["action"] == "pass"


def test_touchstone_wall_override(super_client):
    """body 带 wall 覆盖 = reset/编辑器的预览:同句换墙判定翻转,不动库。"""
    _create(super_client, slug="demo")
    version_before = _config_version(super_client, "demo")

    # 覆盖为 auto 墙(无 € 组): 价格句放行
    r = _post(super_client, "demo",
              {"sentences": [PRICE], "wall": AUTO_WALL})
    assert r.status_code == 200
    body = r.json()
    assert body["wall_source"] == "override"
    assert body["results"][0]["action"] == "pass"

    # 覆盖墙坏 regex → 400 带字段路径与 regex 字样
    bad = dict(JC_WALL)
    bad["price_patterns"] = ["([unclosed"]
    r = _post(super_client, "demo", {"sentences": [PRICE], "wall": bad})
    assert r.status_code == 400
    assert "price_patterns[0]" in r.json()["detail"]
    assert "regex" in r.json()["detail"]

    # 覆盖实验全程不 bump 版本
    assert _config_version(super_client, "demo") == version_before


def test_touchstone_input_validation(super_client):
    _create(super_client, slug="demo")
    # sentences 必须是非空字符串列表
    assert _post(super_client, "demo", {"sentences": []}).status_code == 400
    assert _post(super_client, "demo",
                 {"sentences": "not a list"}).status_code == 400
    assert _post(super_client, "demo",
                 {"sentences": ["ok", 42]}).status_code == 400
    assert _post(super_client, "demo", {}).status_code == 400
    # source 只认 web/app;has_contact 必须 bool;未知键拒绝
    assert _post(super_client, "demo",
                 {"sentences": ["ok."], "source": "sms"}).status_code == 400
    assert _post(super_client, "demo",
                 {"sentences": ["ok."],
                  "has_contact": "yes"}).status_code == 400
    assert _post(super_client, "demo",
                 {"sentences": ["ok."], "surprise": 1}).status_code == 400
    # 未知租户 404
    assert _post(super_client, "ghost",
                 {"sentences": ["ok."]}).status_code == 404