zhanghongyu_0626
70aa861983
- Added new modules for advisor compliance, KYC sessions, and script templates, enhancing the advisor agent's capabilities. - Implemented a comprehensive API structure under the `/api/advisor-agent` prefix, ensuring clear organization and access to new features. - Established database models and repositories for compliance rules and KYC sessions, facilitating robust data management. - Integrated exception handling and response models to improve error management and user feedback. - Updated settings to include new configurations for compliance and KYC features, ensuring flexibility and adaptability. This update significantly expands the advisor agent's functionality, providing essential tools for compliance and customer interaction while maintaining a structured API design.
87 lines
3.2 KiB
Python
87 lines
3.2 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
from app.advisor_db import AgentSessionLocal
|
|
from app.main import app
|
|
from app.model.entities_advisor import ComplianceCheckLog
|
|
from app.model.advisor_schemas import ComplianceCheckRequest
|
|
from app.service.compliance_check_service import ComplianceCheckService
|
|
from app.service.compliance_semantic_service import ComplianceSemanticService
|
|
from scripts.seed.import_compliance_rules import import_rules_from_markdown
|
|
|
|
client = TestClient(app)
|
|
|
|
class FakeTimeoutLLMClient:
|
|
def complete(self, prompt: str, *, timeout_seconds: float) -> str:
|
|
raise TimeoutError("semantic timeout")
|
|
|
|
def advisor_token() -> str:
|
|
response = client.post("/api/auth/login", json={"actor_id": "STAFF-10086", "token_type": "staff"})
|
|
assert response.status_code == 200
|
|
return response.json()["data"]["access_token"]
|
|
|
|
def get_check_log(trace_id: str) -> ComplianceCheckLog:
|
|
with AgentSessionLocal() as session:
|
|
log = (
|
|
session.query(ComplianceCheckLog)
|
|
.filter(ComplianceCheckLog.trace_id == trace_id)
|
|
.order_by(ComplianceCheckLog.id.desc())
|
|
.first()
|
|
)
|
|
assert log is not None
|
|
session.expunge(log)
|
|
return log
|
|
|
|
def test_compliance_check_api_writes_hard_rule_log_with_trace_and_hits():
|
|
import_rules_from_markdown()
|
|
trace_id = "trace-s1-05-hard-rule-log"
|
|
token = advisor_token()
|
|
|
|
response = client.post(
|
|
"/api/advisor-agent/compliance/content-check",
|
|
json={
|
|
"text": "这款产品保证赚钱。",
|
|
"scene": "product_recommend",
|
|
"customer_risk_level": "C3",
|
|
},
|
|
headers={"Authorization": f"Bearer {token}", "X-Trace-Id": trace_id},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
log = get_check_log(trace_id)
|
|
|
|
assert log.check_id == body["data"]["check_id"]
|
|
assert log.trace_id == trace_id
|
|
assert log.advisor_id == "STAFF-10086"
|
|
assert log.scene == "product_recommend"
|
|
assert log.input_text == "这款产品保证赚钱。"
|
|
assert len(log.input_hash) == 64
|
|
assert log.risk_level == "BLOCK"
|
|
assert log.hit_count == 1
|
|
assert log.hit_details[0]["rule_id"] == "CR-TEST-002"
|
|
assert log.ai_analysis is None
|
|
assert log.ai_degraded is False
|
|
assert log.latency_ms >= 0
|
|
assert log.customer_risk_level == "C3"
|
|
|
|
def test_compliance_check_service_writes_ai_degraded_log():
|
|
trace_id = "trace-s1-05-ai-degraded-log"
|
|
semantic_service = ComplianceSemanticService(llm_client=FakeTimeoutLLMClient())
|
|
service = ComplianceCheckService(semantic_service=semantic_service)
|
|
|
|
result = service.check_text(
|
|
ComplianceCheckRequest(text="这段话没有直接命中硬规则。", scene="general"),
|
|
trace_id=trace_id,
|
|
advisor_id="ADV-TEST-001",
|
|
)
|
|
log = get_check_log(trace_id)
|
|
|
|
assert result.risk_level == "WARN"
|
|
assert log.check_id == result.check_id
|
|
assert log.risk_level == "WARN"
|
|
assert log.hit_count == 0
|
|
assert log.hit_details == []
|
|
assert log.ai_analysis["degraded"] is True
|
|
assert log.ai_analysis["reason"] == "AI semantic detection degraded: timeout"
|
|
assert log.ai_degraded is True
|