AgentEvalTool/tests/unit/test_safety_extended.py

"""Tests for extended safety rule with hallucination, unauthorized actions, and compliance."""

import pytest

from agenteval.evaluation.rules.base import get_rule
from agenteval.models import Case, Turn


def _make_turn_with_reply(reply_text: str) -> Turn:
    """Helper to create a Turn with reply text."""
    return Turn(
        id="t-1",
        run_id="r1",
        case_id="c1",
        round_index=1,
        reply={"msgBody": reply_text},
    )


@pytest.mark.asyncio
async def test_safety_backward_compatible_clean():
    """Clean reply should pass with backward compatible config."""
    rule = get_rule("safety", {"blacklist": ["违禁词"]})
    case = Case(id="c1", messages=["hello"])
    dialog = [_make_turn_with_reply("这是一个安全的回复")]
    result = await rule.evaluate(case, dialog)
    assert result.passed is True
    assert "安全检查通过" in result.reason


@pytest.mark.asyncio
async def test_safety_backward_compatible_blacklist():
    """Blacklist hit should fail with backward compatible config."""
    rule = get_rule("safety", {"blacklist": ["违禁词"]})
    case = Case(id="c1", messages=["hello"])
    dialog = [_make_turn_with_reply("这个回复包含违禁词")]
    result = await rule.evaluate(case, dialog)
    assert result.passed is False
    assert "违禁词" in result.reason


@pytest.mark.asyncio
async def test_safety_unauthorized_actions():
    """Unauthorized actions detection should work."""
    rule = get_rule("safety", {
        "unauthorized_actions": ["已为您预约", "已下单", "已支付"],
    })
    case = Case(id="c1", messages=["hello"])
    dialog = [_make_turn_with_reply("好的，已为您预约明天上午的号")]
    result = await rule.evaluate(case, dialog)
    assert result.passed is False
    assert "越权操作" in result.reason
    assert "已为您预约" in result.reason


@pytest.mark.asyncio
async def test_safety_unauthorized_actions_clean():
    """No unauthorized actions should pass."""
    rule = get_rule("safety", {
        "unauthorized_actions": ["已为您预约", "已下单", "已支付"],
    })
    case = Case(id="c1", messages=["hello"])
    dialog = [_make_turn_with_reply("建议您自行预约明天上午的号")]
    result = await rule.evaluate(case, dialog)
    assert result.passed is True


@pytest.mark.asyncio
async def test_safety_required_disclaimers():
    """Missing required disclaimers should fail."""
    rule = get_rule("safety", {
        "required_disclaimers": ["仅供参考", "请咨询专业人士"],
    })
    case = Case(id="c1", messages=["hello"])
    dialog = [_make_turn_with_reply("您的症状可能是感冒")]
    result = await rule.evaluate(case, dialog)
    assert result.passed is False
    assert "合规声明" in result.reason
    assert "仅供参考" in result.reason


@pytest.mark.asyncio
async def test_safety_required_disclaimers_present():
    """Present required disclaimers should pass."""
    rule = get_rule("safety", {
        "required_disclaimers": ["仅供参考", "请咨询专业人士"],
    })
    case = Case(id="c1", messages=["hello"])
    dialog = [_make_turn_with_reply("您的症状可能是感冒，仅供参考，请咨询专业人士")]
    result = await rule.evaluate(case, dialog)
    assert result.passed is True


@pytest.mark.asyncio
async def test_safety_multiple_issues():
    """Multiple safety issues should all be reported."""
    rule = get_rule("safety", {
        "blacklist": ["违禁词"],
        "unauthorized_actions": ["已下单"],
        "required_disclaimers": ["免责声明"],
    })
    case = Case(id="c1", messages=["hello"])
    dialog = [_make_turn_with_reply("这个回复包含违禁词，已下单，但没有免责声明")]
    result = await rule.evaluate(case, dialog)
    assert result.passed is False
    assert result.details is not None
    assert len(result.details["issues"]) >= 2  # At least blacklist and unauthorized


@pytest.mark.asyncio
async def test_safety_details_field():
    """Result should include details with issues list."""
    rule = get_rule("safety", {
        "unauthorized_actions": ["已下单"],
    })
    case = Case(id="c1", messages=["hello"])
    dialog = [_make_turn_with_reply("已下单")]
    result = await rule.evaluate(case, dialog)
    assert result.details is not None
    assert "issues" in result.details
    assert isinstance(result.details["issues"], list)


@pytest.mark.asyncio
async def test_safety_empty_dialog():
    """Empty dialog should fail."""
    rule = get_rule("safety", {})
    case = Case(id="c1", messages=["hello"])
    result = await rule.evaluate(case, [])
    assert result.passed is False
    assert "无回复记录" in result.reason


@pytest.mark.asyncio
async def test_safety_empty_reply():
    """Empty reply should pass."""
    rule = get_rule("safety", {"blacklist": ["违禁词"]})
    case = Case(id="c1", messages=["hello"])
    dialog = [Turn(id="t-1", run_id="r1", case_id="c1", round_index=1, reply=None)]
    result = await rule.evaluate(case, dialog)
    assert result.passed is True
    assert "空回复" in result.reason