120 lines
6.0 KiB
Python
120 lines
6.0 KiB
Python
"""合规层种子数据的集成验证（真实 MySQL，不使用 mock）。
|
||
|
||
断言口径与 `app/service/agent/governance.py` L62-65 的取数条件**逐字一致**：
|
||
`status='active' AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL`。
|
||
用别的口径断言会得出"种子已生效"的假结论——判据一改，governance 仍查不到规则。
|
||
"""
|
||
|
||
import pytest
|
||
from sqlalchemy import text
|
||
|
||
from app.core.contracts import AgentDefinition, RequestContext
|
||
from app.infrastructure.db import SessionFactory
|
||
from app.service.agent.governance import PlatformGovernance
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
#: docs/02 §10.2 逐字要求的 7 个零容忍词（监管口径，不得增减）。
|
||
NEGATIVE_WORDS = ("保本", "稳赚", "无风险", "保证收益", "预期收益率", "年化收益率", "安全")
|
||
|
||
#: `agent_reply_template.scene` 的 CHECK 取值集里，本次要覆盖的 6 类固定话术场景。
|
||
EXPECTED_SCENES = {
|
||
"disclaimer",
|
||
"clarification",
|
||
"low_confidence",
|
||
"compliance_block",
|
||
"transfer",
|
||
"system_busy",
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_all_seven_negative_words_are_active_and_reviewed() -> None:
|
||
async with SessionFactory() as session:
|
||
rows = (await session.execute(text(
|
||
"SELECT word_pattern FROM agent_negative_word "
|
||
"WHERE status='active' AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL"
|
||
))).scalars().all()
|
||
patterns = " ".join(rows)
|
||
for word in NEGATIVE_WORDS:
|
||
assert word in patterns, f"负面词未生效：{word}"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_reply_templates_cover_all_six_scenes() -> None:
|
||
async with SessionFactory() as session:
|
||
rows = (await session.execute(text(
|
||
"SELECT template_code, scene FROM agent_reply_template "
|
||
"WHERE status='active' AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL"
|
||
))).all()
|
||
scenes = {row.scene for row in rows}
|
||
assert len(rows) >= 6, f"话术模板不足 6 条：{len(rows)}"
|
||
assert len(scenes) >= 6, f"场景不足 6 类：{scenes}"
|
||
# 比"数量够"更严的一步：6 类场景必须正好是 DDL CHECK 取值集里要覆盖的那 6 类。
|
||
# 只数数量的话，把 6 条都塞进同一个 scene 也能过——那已经不是"6 类固定话术"了。
|
||
assert scenes == EXPECTED_SCENES, f"话术场景与 DDL 取值不符：{scenes}"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_negative_rules_are_gated_to_customer_service_only() -> None:
|
||
"""规则必须按 `applicable_agents` 门控，且只对客服 Agent 生效。
|
||
|
||
为什么不能只查 SQL：光断言"规则在库里存在且 active"**抓不到门控被破坏**——门控只在
|
||
`governance.py` L70-74 里生效，而那条判据是 `if agents and definition.agent_type not in
|
||
agents: continue`，两种坏取值各有各的坏法（均为实测结论）：
|
||
- `applicable_agents` 指向别的 Agent（如 `["advisor"]`）→ 客服一条都加载不到（服务静默失效）；
|
||
- 置空 `[]` 或 `NULL` → 判据的 `if agents` 为假、`continue` 不执行 → 规则**溢出到所有
|
||
Agent**（advisor 也加载到 11 条），是 fail-open 方向。
|
||
所以正面断言（客服加载数 == 库里生效数）与反面断言（其他 Agent 必须为 0）缺一不可：
|
||
只留正面断言会漏掉 `[]`/`NULL` 的溢出。三种坏取值下本测试都会失败（已逐一实测）。
|
||
"""
|
||
context = RequestContext(user_id="9001", trace_id="t", roles=("customer",))
|
||
service = PlatformGovernance()
|
||
|
||
customer_service = await service.resolve(
|
||
AgentDefinition(agent_type="customer_service", version="1"), context)
|
||
other_agent = await service.resolve(
|
||
AgentDefinition(agent_type="advisor", version="1"), context)
|
||
|
||
async with SessionFactory() as session:
|
||
seeded = await session.scalar(text(
|
||
"SELECT COUNT(*) FROM agent_negative_word "
|
||
"WHERE status='active' AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL"
|
||
))
|
||
assert seeded >= len(NEGATIVE_WORDS), f"生效规则少于 7 条硬要求：{seeded}"
|
||
assert len(customer_service.negative_rules) == seeded, (
|
||
f"客服 Agent 应加载全部 {seeded} 条规则，实际 {len(customer_service.negative_rules)} 条")
|
||
assert customer_service.negative_rules, "客服 Agent 一条规则都没加载"
|
||
assert other_agent.negative_rules == (), (
|
||
f"其他 Agent 不应加载这些规则，实际加载了 {other_agent.negative_rules}")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_own_reply_templates_do_not_trip_the_seeded_rules() -> None:
|
||
"""话术不得被自己的规则拦截——否则 Task 2 接入话术时会形成自绊循环。
|
||
|
||
治理层对 `contains` 是朴素子串匹配（`governance.py` L127-128），**没有否定式豁免**：
|
||
写过「非保本」的合规话术会命中「保本」规则。目前 `governance.py` L131 命中后替换的是
|
||
写死的句子、并不读 `safe_reply_template_code`，所以自绊尚未暴露；等 Task 2 把话术接进
|
||
治理层，"替换文本又被规则过滤"的循环就会出现。这条测试把该事实固定住。
|
||
"""
|
||
async with SessionFactory() as session:
|
||
rules = (await session.execute(text(
|
||
"SELECT word_pattern FROM agent_negative_word "
|
||
"WHERE status='active' AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL"
|
||
))).scalars().all()
|
||
templates = (await session.execute(text(
|
||
"SELECT template_code, content_text FROM agent_reply_template "
|
||
"WHERE status='active' AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL"
|
||
))).all()
|
||
|
||
assert rules, "没有任何生效规则，这条自绊检查会退化成永远通过"
|
||
assert templates, "没有任何生效话术"
|
||
trips = [
|
||
(row.template_code, word)
|
||
for row in templates
|
||
for word in rules
|
||
if word in (row.content_text or "")
|
||
]
|
||
assert trips == [], f"话术命中了自己的禁用词（自绊）：{trips}"
|