ZNJJ-api-server/test/test_eval_catalog.py

"""Selection tests use real case files and make no API/model calls."""

from collections import Counter
from contextlib import nullcontext
import json
import os
from pathlib import Path
import subprocess
import sys

import pytest
import yaml

from evals.catalog import load_suites, select_cases
from evals.run import CaseTranscript, DEFAULT_CASES_DIR, load_cases

ROOT = Path(__file__).resolve().parents[1]


@pytest.fixture
def catalog():
    return load_cases(DEFAULT_CASES_DIR, [])


def test_full_keeps_all_cases_and_modes(catalog):
    selected = select_cases(catalog, suites=["full"])
    assert selected == catalog
    assert {c["clientMode"] for c in selected} == {"direct", "browser_addon"}


def test_overlapping_suites_execute_each_case_once(catalog):
    selected = select_cases(catalog, suites=["feedback", "collection", "feedback"])
    expected = set(c["name"] for c in select_cases(catalog, suites=["collection"]))
    assert len(selected) == len(expected)
    assert {c["name"] for c in selected} == expected


def test_filter_intersection_and_exclusion(catalog):
    selected = select_cases(catalog, suites=["collection"],
                            groups=["collection_skip_known_info"], mode="direct",
                            names=["single", "dual"], exclude_names=["single", "unclear"])
    assert selected
    assert all(c["clientMode"] == "direct" and "dual" in c["name"]
               and "unclear" not in c["name"] for c in selected)
    assert all(c["_group"] == "collection_skip_known_info" for c in selected)


@pytest.mark.parametrize("kwargs", [
    {"groups": ["not-a-group"]},
    {"suites": ["not-a-suite"]},
    {"names": ["not-a-case"]},
    {"groups": ["complex_vehicle"], "suites": ["photo"]},
])
def test_bad_or_empty_selection_fails(catalog, kwargs):
    with pytest.raises(ValueError):
        select_cases(catalog, **kwargs)


def test_suites_have_live_patterns_and_modes(catalog):
    for name, suite in load_suites().items():
        assert {c["clientMode"] for c in select_cases(catalog, suites=[name])} == {
            "direct", "browser_addon",
        }
        for pattern in suite["include"]:
            assert any(Path(c["_relative_path"]).match(pattern) for c in catalog), (name, pattern)
    assert len(select_cases(catalog, suites=["smoke"])) < len(catalog)


def test_directory_group_and_relative_path(catalog):
    for case in catalog:
        assert case["_group"] == Path(case["_relative_path"]).parent.as_posix()
        assert "\\" not in case["_relative_path"]


def test_custom_directory_and_duplicate_names(tmp_path):
    case = {"name": "custom", "clientMode": "direct", "steps": [
        {"input": "开始", "expect": {"stageCode": "1001"}},
    ]}
    group = tmp_path / "custom_group"
    group.mkdir()
    (group / "a.yaml").write_text(yaml.safe_dump(case), encoding="utf-8")
    selected = select_cases(load_cases(tmp_path, []), groups=["custom_group"])
    assert selected[0]["_relative_path"] == "custom_group/a.yaml"
    (group / "b.yaml").write_text(yaml.safe_dump(case), encoding="utf-8")
    with pytest.raises(ValueError, match="unique"):
        load_cases(tmp_path, [])


@pytest.mark.parametrize("command", [
    ["-m", "evals.run"],
    ["evals/run.py"],
    ["-m", "evals.prompt_compare"],
])
def test_cli_preview_has_no_provider_or_output_requirement(command, tmp_path):
    env = {**os.environ, "PYTHONIOENCODING": "utf-8", "LLM_PROVIDER": "fake",
           "AGENT_BACKEND": "fastgpt"}
    result = subprocess.run(
        [sys.executable, *command, "--suite", "feedback", "--group", "complex_vehicle",
         "--mode", "direct", "--list"], cwd=ROOT, env=env,
        capture_output=True, text=True, encoding="utf-8", timeout=30,
    )
    assert result.returncode == 0, result.stderr
    assert "direct_complex_three_immediate" in result.stdout
    assert "browser_addon_complex_three_immediate" not in result.stdout
    assert "Selected 1 of" in result.stdout


def test_no_identical_inputs_and_assertions_within_a_mode(catalog):
    fingerprints = Counter(json.dumps({"mode": case["clientMode"], "steps": case["steps"]},
                                     sort_keys=True, ensure_ascii=False) for case in catalog)
    assert all(count == 1 for count in fingerprints.values())


@pytest.mark.parametrize("passed,exit_code", [(True, 0), (False, 1)])
def test_execution_uses_the_preview_selection(monkeypatch, passed, exit_code):
    from evals import run

    calls = []
    monkeypatch.setattr(sys, "argv", ["evals.run", "--suite", "feedback", "--suite", "collection",
                                    "--group", "complex_vehicle", "--mode", "direct", "--no-save"])
    monkeypatch.setattr(run.httpx, "Client", lambda **kwargs: nullcontext(None))

    def fake_run(client, case):
        calls.append(case["name"])
        return CaseTranscript(case["name"], case["_path"], case["clientMode"], "test", passed)

    monkeypatch.setattr(run, "run_case", fake_run)
    assert run.main() == exit_code
    assert calls == ["direct_complex_three_immediate"]


def test_cli_unknown_group_is_configuration_error():
    result = subprocess.run(
        [sys.executable, "-m", "evals.run", "--group", "unknown", "--list"],
        cwd=ROOT, capture_output=True, text=True, encoding="utf-8", timeout=30,
    )
    assert result.returncode == 2
    assert "Unknown groups" in result.stderr