agent-lab/tests/test_benchmark_config.py

import json

import pytest
from pydantic import ValidationError

from agent_lab.application.benchmark import (
    BENCHMARK_CASE_CATALOG,
    BENCHMARK_CASE_IDS,
    BENCHMARK_MODES,
    BenchmarkConfig,
    BenchmarkTarget,
    build_benchmark_request,
    build_mock_rounds,
)


VALID_CONFIG = {
    "schema_version": 1,
    "base_url": "https://provider.example/v1",
    "model": "benchmark-model",
    "runs_per_case": 1,
    "modes": ["dual_agent", "chat_agent_tools"],
    "cases": [
        "ordinary_chat",
        "device_volume_silent",
        "calendar_schedule_template",
        "web_search_two_answers",
        "session_terminate",
        "parallel_volume_schedule",
    ],
}


def test_benchmark_config_accepts_the_versioned_single_target_schema():
    config = BenchmarkConfig.model_validate_json(json.dumps(VALID_CONFIG))

    assert config.schema_version == 1
    assert config.base_url == "https://provider.example/v1"
    assert config.model == "benchmark-model"
    assert config.runs_per_case == 1
    assert tuple(config.modes) == BENCHMARK_MODES
    assert tuple(config.cases) == BENCHMARK_CASE_IDS


def test_benchmark_target_uses_the_same_strict_url_and_model_validation():
    target = BenchmarkTarget(
        base_url="http://localhost:8000/v1",
        model="local-model",
    )

    assert target.model_dump() == {
        "base_url": "http://localhost:8000/v1",
        "model": "local-model",
    }


@pytest.mark.parametrize("schema_version", [0, 2, "1", True, 1.0])
def test_benchmark_config_rejects_unknown_or_coerced_schema_versions(
    schema_version: object,
):
    payload = {**VALID_CONFIG, "schema_version": schema_version}

    with pytest.raises(ValidationError, match="schema_version"):
        BenchmarkConfig.model_validate(payload)


@pytest.mark.parametrize(
    "base_url",
    [
        "ftp://provider.example/v1",
        "provider.example/v1",
        "https:///v1",
        "https://user:pass@provider.example/v1",
        "https://provider.example/v1?debug=true",
        "https://provider.example/v1#fragment",
    ],
)
def test_benchmark_config_rejects_unsafe_or_non_http_base_urls(base_url: str):
    payload = {**VALID_CONFIG, "base_url": base_url}

    with pytest.raises(ValidationError, match="base_url"):
        BenchmarkConfig.model_validate(payload)


@pytest.mark.parametrize(
    "base_url",
    [
        pytest.param("https://🚀.example/v1", id="invalid-idna-host"),
        pytest.param("https://xn--.example/v1", id="malformed-a-label"),
        pytest.param(
            "https://provider.example:not-a-port/v1",
            id="invalid-port",
        ),
        pytest.param(
            "https://provider.example:65536/v1",
            id="out-of-range-port",
        ),
        pytest.param("https://[v1.invalid]/v1", id="invalid-ip-literal"),
    ],
)
def test_benchmark_target_rejects_urls_incompatible_with_httpx(base_url: str):
    with pytest.raises(ValidationError, match="base_url"):
        BenchmarkTarget(base_url=base_url, model="benchmark-model")


@pytest.mark.parametrize(
    "character",
    [
        pytest.param(chr(0), id="nul-cc"),
        pytest.param("\r\n", id="crlf-cc"),
        pytest.param("\t", id="tab-cc"),
        pytest.param(chr(0x200B), id="zero-width-space-cf"),
        pytest.param(chr(0x202E), id="right-to-left-override-cf"),
        pytest.param(chr(0xE000), id="private-use-co"),
        pytest.param(chr(0xD800), id="surrogate-cs"),
        pytest.param(chr(0x0378), id="unassigned-cn"),
    ],
)
def test_benchmark_config_rejects_unicode_category_c_in_base_url(character: str):
    payload = {
        **VALID_CONFIG,
        "base_url": f"https://provider.example/v1{character}segment",
    }

    with pytest.raises(ValidationError, match="base_url"):
        BenchmarkConfig.model_validate(payload)


def test_benchmark_config_accepts_printable_internationalized_base_url():
    base_url = "https://例え.テスト/路径/模型-🚀"

    config = BenchmarkConfig.model_validate({**VALID_CONFIG, "base_url": base_url})

    assert config.base_url == base_url


@pytest.mark.parametrize("model", ["", "   "])
def test_benchmark_config_rejects_empty_models(model: str):
    payload = {**VALID_CONFIG, "model": model}

    with pytest.raises(ValidationError, match="model"):
        BenchmarkConfig.model_validate(payload)


def test_benchmark_config_rejects_non_positive_runs_per_case():
    payload = {**VALID_CONFIG, "runs_per_case": 0}

    with pytest.raises(ValidationError, match="runs_per_case"):
        BenchmarkConfig.model_validate(payload)


@pytest.mark.parametrize(
    "modes",
    [[], ["dual_agent", "dual_agent"], ["unsupported"]],
)
def test_benchmark_config_rejects_empty_duplicate_or_unknown_modes(
    modes: list[str],
):
    payload = {**VALID_CONFIG, "modes": modes}

    with pytest.raises(ValidationError, match="modes"):
        BenchmarkConfig.model_validate(payload)


@pytest.mark.parametrize(
    "cases",
    [[], ["ordinary_chat", "ordinary_chat"], ["custom_case"]],
)
def test_benchmark_config_rejects_empty_duplicate_or_unknown_cases(
    cases: list[str],
):
    payload = {**VALID_CONFIG, "cases": cases}

    with pytest.raises(ValidationError, match="cases"):
        BenchmarkConfig.model_validate(payload)


@pytest.mark.parametrize("field", ["api_key", "token", "unknown_field"])
def test_benchmark_config_rejects_secrets_and_unknown_json_fields(field: str):
    secret_value = "must-not-appear-in-validation-errors"
    payload = {**VALID_CONFIG, field: secret_value}

    with pytest.raises(ValidationError) as exc_info:
        BenchmarkConfig.model_validate_json(json.dumps(payload))

    assert field in str(exc_info.value)
    assert secret_value not in str(exc_info.value)


def test_builtin_case_catalog_is_complete_and_immutable():
    assert tuple(BENCHMARK_CASE_CATALOG) == BENCHMARK_CASE_IDS

    with pytest.raises(TypeError):
        BENCHMARK_CASE_CATALOG["ordinary_chat"] = BENCHMARK_CASE_CATALOG[
            "ordinary_chat"
        ]

    with pytest.raises(ValidationError, match="frozen"):
        BENCHMARK_CASE_CATALOG["ordinary_chat"].user_message = "changed"

    first_event = BENCHMARK_CASE_CATALOG["device_volume_silent"].mock_events[0]
    with pytest.raises(TypeError):
        first_event.arguments[0] = ("mode", "relative")


def test_builtin_cases_own_semantics_and_deterministic_mock_data():
    actual = {
        case_id: {
            "user_message": case.user_message,
            "enabled_tools": case.enabled_tools,
            "events": case.expectation.event_names,
            "answer_count": case.expectation.answer_count,
            "terminal": case.expectation.terminal,
            "first_reply_before_tool": case.expectation.first_reply_before_tool,
            "expected_visible": case.expectation.visible_messages,
            "mock_visible": case.mock_visible_messages,
            "mock_events": tuple(
                (event.id, event.name, dict(event.arguments))
                for event in case.mock_events
            ),
        }
        for case_id, case in BENCHMARK_CASE_CATALOG.items()
    }

    assert actual == {
        "ordinary_chat": {
            "user_message": "Hello.",
            "enabled_tools": (
                "session.terminate",
                "device.volume.adjust",
                "calendar.schedule.create",
                "knowledge.web.search",
            ),
            "events": (),
            "answer_count": 1,
            "terminal": False,
            "first_reply_before_tool": False,
            "expected_visible": ("Ordinary answer.",),
            "mock_visible": ("Ordinary answer.",),
            "mock_events": (),
        },
        "device_volume_silent": {
            "user_message": "set volume to 30",
            "enabled_tools": ("device.volume.adjust",),
            "events": ("device.volume.adjust",),
            "answer_count": 1,
            "terminal": False,
            "first_reply_before_tool": True,
            "expected_visible": ("I will set the volume to 30.",),
            "mock_visible": ("I will set the volume to 30.",),
            "mock_events": (
                (
                    "volume-1",
                    "device.volume.adjust",
                    {"mode": "absolute", "value": 30},
                ),
            ),
        },
        "calendar_schedule_template": {
            "user_message": (
                'schedule "Design review" at 2026-07-14T09:30:00+08:00 '
                "timezone Asia/Shanghai"
            ),
            "enabled_tools": ("calendar.schedule.create",),
            "events": ("calendar.schedule.create",),
            "answer_count": 2,
            "terminal": False,
            "first_reply_before_tool": True,
            "expected_visible": (
                "I will schedule that.",
                "Scheduled Design review for 2026-07-14T09:30:00+08:00 "
                "(Asia/Shanghai).",
            ),
            "mock_visible": ("I will schedule that.",),
            "mock_events": (
                (
                    "schedule-1",
                    "calendar.schedule.create",
                    {
                        "title": "Design review",
                        "start_at": "2026-07-14T09:30:00+08:00",
                        "timezone": "Asia/Shanghai",
                    },
                ),
            ),
        },
        "web_search_two_answers": {
            "user_message": "event batch safety",
            "enabled_tools": ("knowledge.web.search",),
            "events": ("knowledge.web.search",),
            "answer_count": 2,
            "terminal": False,
            "first_reply_before_tool": True,
            "expected_visible": ("I will check.", "Grounded search update."),
            "mock_visible": ("I will check.", "Grounded search update."),
            "mock_events": (
                (
                    "search-1",
                    "knowledge.web.search",
                    {"query": "event batch safety"},
                ),
            ),
        },
        "session_terminate": {
            "user_message": "end this session",
            "enabled_tools": ("session.terminate",),
            "events": ("session.terminate",),
            "answer_count": 1,
            "terminal": True,
            "first_reply_before_tool": True,
            "expected_visible": ("Goodbye.",),
            "mock_visible": ("Goodbye.",),
            "mock_events": (("terminate-1", "session.terminate", {}),),
        },
        "parallel_volume_schedule": {
            "user_message": (
                'set volume to 30 and schedule "Design review" at '
                "2026-07-14T09:30:00+08:00 timezone Asia/Shanghai"
            ),
            "enabled_tools": (
                "device.volume.adjust",
                "calendar.schedule.create",
            ),
            "events": (
                "device.volume.adjust",
                "calendar.schedule.create",
            ),
            "answer_count": 2,
            "terminal": False,
            "first_reply_before_tool": True,
            "expected_visible": (
                "I will set the volume to 30 and schedule that.",
                "Scheduled Design review for 2026-07-14T09:30:00+08:00 "
                "(Asia/Shanghai).",
            ),
            "mock_visible": (
                "I will set the volume to 30 and schedule that.",
            ),
            "mock_events": (
                (
                    "parallel-volume",
                    "device.volume.adjust",
                    {"mode": "absolute", "value": 30},
                ),
                (
                    "parallel-schedule",
                    "calendar.schedule.create",
                    {
                        "title": "Design review",
                        "start_at": "2026-07-14T09:30:00+08:00",
                        "timezone": "Asia/Shanghai",
                    },
                ),
            ),
        },
    }


@pytest.mark.parametrize("case_id", BENCHMARK_CASE_IDS)
def test_build_benchmark_request_uses_fixed_fairness_controls(case_id: str):
    case = BENCHMARK_CASE_CATALOG[case_id]
    requests = {
        mode: build_benchmark_request(case, mode, "comparison-model")
        for mode in BENCHMARK_MODES
    }

    dual = requests["dual_agent"]
    tools = requests["chat_agent_tools"]
    assert dual.model_dump(exclude={"tool_invocation_mode"}) == tools.model_dump(
        exclude={"tool_invocation_mode"}
    )
    assert dual.user_message == case.user_message
    assert dual.system_prompts == ["Exercise the requested business scenario."]
    assert dual.pre_messages == []
    assert dual.chat_agent.model == "comparison-model"
    assert dual.chat_agent.temperature == 0.0
    assert dual.chat_agent.max_tokens == 128
    assert dual.event_agent.model == "comparison-model"
    assert dual.event_agent.temperature == 0.0
    assert dual.event_agent.max_tokens == 128
    assert tuple(dual.event_agent.enabled_tools) == case.enabled_tools
    assert dual.event_agent.max_event_loops == 1
    assert dual.event_agent.max_parallel_events == 2
    assert dual.event_agent.batch_timeout_seconds == 1.0
    assert dual.tool_invocation_mode == "dual_agent"
    assert tools.tool_invocation_mode == "chat_agent_tools"


def test_build_benchmark_request_rejects_an_empty_model():
    case = BENCHMARK_CASE_CATALOG["ordinary_chat"]

    with pytest.raises(ValueError, match="model"):
        build_benchmark_request(case, "dual_agent", "   ")


@pytest.mark.parametrize("mode", BENCHMARK_MODES)
@pytest.mark.parametrize("case_id", BENCHMARK_CASE_IDS)
def test_mock_rounds_are_mode_correct_and_derived_from_catalog(
    mode: str,
    case_id: str,
):
    case = BENCHMARK_CASE_CATALOG[case_id]
    rounds = build_mock_rounds(case, mode)
    items = tuple(item for round_items in rounds for item in round_items)
    message_items = tuple(item for item in items if item.kind == "message_delta")
    event_items = tuple(item for item in items if item.event is not None)

    assert tuple(item.content for item in message_items) == case.mock_visible_messages
    assert tuple(item.event.name for item in event_items) == case.expectation.event_names
    assert tuple(item.event.id for item in event_items) == tuple(
        event.id for event in case.mock_events
    )
    assert tuple(item.event.arguments for item in event_items) == tuple(
        dict(event.arguments) for event in case.mock_events
    )
    expected_kind = "text_event" if mode == "dual_agent" else "provider_tool_call"
    assert {item.kind for item in event_items} <= {expected_kind}
    assert len(rounds) == max(1, len(case.mock_visible_messages))