122 lines
5.0 KiB
Python
122 lines
5.0 KiB
Python
"""AI 辅助创建与冲突检测单元测试（mock LLM / 内存库）。"""
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from sqlmodel import Session, SQLModel, create_engine
|
|
|
|
from app.ai_assist import _find_mutex_clauses, _similar, conflict_scan, draft_from_blab
|
|
from app.models import Agent, AgentPromptTemplate, AgentSkill, ModelProvider
|
|
|
|
|
|
def _session():
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False})
|
|
SQLModel.metadata.create_all(engine)
|
|
return Session(engine)
|
|
|
|
|
|
# ---------------- 大白话 → 草稿 ----------------
|
|
|
|
def test_draft_requires_description():
|
|
with _session() as s:
|
|
try:
|
|
draft_from_blab(s, "agent", " ")
|
|
raised = False
|
|
except ValueError as exc:
|
|
raised = True
|
|
assert "描述" in str(exc)
|
|
assert raised
|
|
|
|
|
|
def test_draft_requires_model():
|
|
with _session() as s:
|
|
try:
|
|
draft_from_blab(s, "agent", "我要一个写小红书的助手")
|
|
raised = False
|
|
except ValueError as exc:
|
|
raised = True
|
|
assert "模型" in str(exc) or "Provider" in str(exc)
|
|
assert raised
|
|
|
|
|
|
def test_draft_returns_structured_json():
|
|
with _session() as s:
|
|
s.add(ModelProvider(name="P", base_url="https://x/v1", models=["m1"],
|
|
default_model="m1", enabled=True))
|
|
s.commit()
|
|
fake = {"status": "ok", "json": {"kind": "agent", "name": "小红书文案手", "role": "production",
|
|
"desc": "写小红书笔记", "rules": ["口语化", "带emoji"],
|
|
"advice": "建议挂接小红书知识库"},
|
|
"provider": "P", "model": "m1", "tokens": 500, "duration_ms": 800}
|
|
with patch("app.ai_assist.chat_json", return_value=fake):
|
|
draft = draft_from_blab(s, "agent", "我要一个写小红书笔记的助手，说话口语化，多用emoji")
|
|
assert draft["name"] == "小红书文案手"
|
|
assert draft["kind"] == "agent"
|
|
assert draft["_meta"]["model"] == "m1"
|
|
|
|
|
|
# ---------------- 冲突检测 ----------------
|
|
|
|
def test_mutex_clause_detection():
|
|
assert _find_mutex_clauses("要求口语化，但报告部分要书面", []) == ["口语 ↔ 书面"]
|
|
assert _find_mutex_clauses("全程大白话，一段到底", []) == []
|
|
assert _find_mutex_clauses("既要严肃又要幽默", []) == ["严肃 ↔ 幽默"]
|
|
|
|
|
|
def test_similar_name_warning():
|
|
assert _similar("小红书文案手", "小红书文案助手") > 0.8
|
|
assert _similar("小红书文案手", "数据分析师") < 0.5
|
|
|
|
|
|
def test_conflict_scan_duplicate_name_blocks():
|
|
with _session() as s:
|
|
s.add(AgentSkill(name="口播稿写手", desc="", instructions="写口播稿"))
|
|
s.commit()
|
|
result = conflict_scan(s, "skill", {"name": "口播稿写手", "instructions": "x"})
|
|
assert result["conflicts"] and "同名" in result["conflicts"][0]
|
|
|
|
|
|
def test_conflict_scan_similar_name_warns():
|
|
with _session() as s:
|
|
s.add(AgentSkill(name="口播稿写手", desc="", instructions="写口播稿"))
|
|
s.commit()
|
|
result = conflict_scan(s, "skill", {"name": "口播稿写手2", "instructions": "x"})
|
|
assert not result["conflicts"]
|
|
assert any("很像" in w for w in result["warnings"])
|
|
|
|
|
|
def test_conflict_scan_internal_mutex():
|
|
with _session() as s:
|
|
result = conflict_scan(s, "template", {
|
|
"name": "矛盾模板", "role_prompt": "你要求输出既简短又详细展开"})
|
|
assert any("打架" in w or "互相" in w for w in result["warnings"])
|
|
|
|
|
|
def test_conflict_scan_agent_vs_skill():
|
|
with _session() as s:
|
|
skill = AgentSkill(name="书面报告技能", desc="", instructions="全部输出用书面语，禁止口语")
|
|
s.add(skill); s.commit(); s.refresh(skill)
|
|
result = conflict_scan(s, "agent", {
|
|
"name": "口语主播", "rules": ["全程口语化，像聊天一样"],
|
|
"skill_ids": [skill.id]})
|
|
assert any("书面" in w or "口语" in w for w in result["warnings"]), result["warnings"]
|
|
|
|
|
|
def test_conflict_scan_agent_vs_template():
|
|
with _session() as s:
|
|
template = AgentPromptTemplate(name="书面模板", role_prompt="正式书面输出",
|
|
constraints=["禁止口语化表达"])
|
|
s.add(template); s.commit(); s.refresh(template)
|
|
result = conflict_scan(s, "agent", {
|
|
"name": "口语助手", "rules": ["全程口语化"],
|
|
"prompt_template_id": template.id})
|
|
assert any("模板" in w for w in result["warnings"])
|
|
|
|
|
|
def test_conflict_scan_exclude_self():
|
|
with _session() as s:
|
|
skill = AgentSkill(name="测试技能", desc="", instructions="x")
|
|
s.add(skill); s.commit(); s.refresh(skill)
|
|
result = conflict_scan(s, "skill", {"name": "测试技能", "instructions": "y"},
|
|
exclude_id=skill.id)
|
|
assert not result["conflicts"]
|