group_xinghuo_jinrong/tests/test_sprint1_hard_rule_detection.py

from fastapi.testclient import TestClient

from app.main import app

client = TestClient(app)

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 test_hard_rule_service_returns_block_when_keyword_rule_matches():
    from app.model.advisor_schemas import ComplianceCheckRequest
    from app.service.compliance_check_service import ComplianceCheckService
    from scripts.seed.import_compliance_rules import import_rules_from_markdown

    import_rules_from_markdown()
    service = ComplianceCheckService()

    result = service.check_text(
        ComplianceCheckRequest(
            text="这款产品保本保收益，适合稳健客户。",
            scene="product_recommend",
            customer_risk_level="C3",
        )
    )

    assert result.risk_level == "BLOCK"
    assert result.can_copy is False
    assert result.required_action == "modify_and_recheck"
    assert result.ai_analysis is None
    assert result.hits[0].rule_id == "CR-TEST-001"
    assert result.hits[0].rule_type == "keyword"
    assert result.hits[0].layer == "hard_rule"
    assert result.hits[0].severity == "BLOCK"
    assert result.hits[0].matched_text == "保本保收益"
    assert result.hits[0].position == {"start": 4, "end": 9}

def test_hard_rule_service_returns_warn_and_required_confirmation():
    from app.model.advisor_schemas import ComplianceCheckRequest
    from app.service.compliance_check_service import ComplianceCheckService
    from scripts.seed.import_compliance_rules import import_rules_from_markdown

    import_rules_from_markdown()
    service = ComplianceCheckService()

    result = service.check_text(ComplianceCheckRequest(text="这款产品错过再无机会，请尽快决策。"))

    assert result.risk_level == "WARN"
    assert result.can_copy is True
    assert result.required_action == "warn_confirm"
    assert result.hits[0].severity == "WARN"
    assert result.hits[0].rule_id == "CR-TEST-006"

def test_hard_rule_service_returns_info_when_no_hard_rule_matches():
    from app.model.advisor_schemas import ComplianceCheckRequest
    from app.service.compliance_check_service import ComplianceCheckService
    from scripts.seed.import_compliance_rules import import_rules_from_markdown

    import_rules_from_markdown()
    service = ComplianceCheckService()

    result = service.check_text(ComplianceCheckRequest(text="该产品历史表现存在波动，请结合自身风险承受能力判断。"))

    assert result.risk_level == "INFO"
    assert result.can_copy is True
    assert result.required_action == "none"
    assert result.hits == []

def test_compliance_check_api_uses_hard_rules_and_requires_auth():
    token = advisor_token()

    denied = client.post(
        "/api/advisor-agent/compliance/content-check",
        json={"text": "保本保收益"},
        headers={"X-Trace-Id": "trace-compliance-no-token"},
    )
    assert denied.status_code == 401

    response = client.post(
        "/api/advisor-agent/compliance/content-check",
        json={"text": "这款产品保证赚钱。", "scene": "product_recommend"},
        headers={"Authorization": f"Bearer {token}", "X-Trace-Id": "trace-compliance-check"},
    )

    assert response.status_code == 200
    body = response.json()
    assert body["trace_id"] == "trace-compliance-check"
    assert body["data"]["risk_level"] == "BLOCK"
    assert body["data"]["can_copy"] is False
    assert body["data"]["required_action"] == "modify_and_recheck"
    assert body["data"]["hits"][0]["rule_id"] == "CR-TEST-002"