iomgaa
dea8a7d3f6
从 JSON 目录 glob *.json 加载题目，stem 作 video_id。 legacy schema 无 difficulty 字段时赋 _LEGACY_DEFAULT_DIFFICULTY 常量。 options/source_nodes 转 tuple 配合 frozen dataclass。 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
130 lines
4.9 KiB
Python
130 lines
4.9 KiB
Python
"""app/question_gen/loader.py 单元测试。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.question_gen.loader import load_benchmark
|
|
from core.types import GeneratedQuestion
|
|
|
|
|
|
@pytest.fixture()
|
|
def benchmark_dir(tmp_path: Path) -> Path:
|
|
"""创建包含 benchmark JSON 的临时目录。"""
|
|
data = [
|
|
{
|
|
"question_id": "1-1",
|
|
"task_type": "Action Reasoning",
|
|
"question": "What happened?",
|
|
"options": ["A. X", "B. Y", "C. Z", "D. W"],
|
|
"answer": "A",
|
|
},
|
|
{
|
|
"question_id": "1-2",
|
|
"task_type": "OCR Problems",
|
|
"question": "What text is shown?",
|
|
"options": ["A. Hello", "B. World", "C. Foo", "D. Bar"],
|
|
"answer": "B",
|
|
},
|
|
]
|
|
(tmp_path / "video_abc.json").write_text(json.dumps(data), encoding="utf-8")
|
|
return tmp_path
|
|
|
|
|
|
class TestLoadBenchmark:
|
|
def test_loads_questions_from_json(self, benchmark_dir: Path) -> None:
|
|
questions = load_benchmark(benchmark_dir)
|
|
assert len(questions) == 2
|
|
|
|
def test_video_id_from_filename(self, benchmark_dir: Path) -> None:
|
|
questions = load_benchmark(benchmark_dir)
|
|
assert all(q.video_id == "video_abc" for q in questions)
|
|
|
|
def test_fields_mapped_correctly(self, benchmark_dir: Path) -> None:
|
|
questions = load_benchmark(benchmark_dir)
|
|
q = questions[0]
|
|
assert q.question_id == "1-1"
|
|
assert q.task_type == "Action Reasoning"
|
|
assert q.question == "What happened?"
|
|
assert q.options == ("A. X", "B. Y", "C. Z", "D. W")
|
|
assert q.answer == "A"
|
|
|
|
def test_options_is_tuple(self, benchmark_dir: Path) -> None:
|
|
questions = load_benchmark(benchmark_dir)
|
|
assert isinstance(questions[0].options, tuple)
|
|
|
|
def test_source_nodes_is_empty_tuple(self, benchmark_dir: Path) -> None:
|
|
questions = load_benchmark(benchmark_dir)
|
|
assert questions[0].source_nodes == ()
|
|
|
|
def test_difficulty_defaults_to_medium_for_legacy(self, benchmark_dir: Path) -> None:
|
|
questions = load_benchmark(benchmark_dir)
|
|
assert questions[0].difficulty == "medium"
|
|
|
|
def test_difficulty_from_json_when_present(self, tmp_path: Path) -> None:
|
|
data = [
|
|
{
|
|
"question_id": "2-1",
|
|
"task_type": "OCR Problems",
|
|
"question": "Q?",
|
|
"options": ["A. 1", "B. 2", "C. 3", "D. 4"],
|
|
"answer": "C",
|
|
"difficulty": "hard",
|
|
}
|
|
]
|
|
(tmp_path / "vid.json").write_text(json.dumps(data), encoding="utf-8")
|
|
questions = load_benchmark(tmp_path)
|
|
assert questions[0].difficulty == "hard"
|
|
|
|
def test_empty_directory_returns_empty_list(self, tmp_path: Path) -> None:
|
|
questions = load_benchmark(tmp_path)
|
|
assert questions == []
|
|
|
|
def test_sorted_by_filename(self, tmp_path: Path) -> None:
|
|
for name in ["z_video.json", "a_video.json"]:
|
|
data = [
|
|
{
|
|
"question_id": f"{name}-1",
|
|
"task_type": "T",
|
|
"question": "Q?",
|
|
"options": ["A", "B", "C", "D"],
|
|
"answer": "A",
|
|
}
|
|
]
|
|
(tmp_path / name).write_text(json.dumps(data), encoding="utf-8")
|
|
questions = load_benchmark(tmp_path)
|
|
assert questions[0].video_id == "a_video"
|
|
assert questions[1].video_id == "z_video"
|
|
|
|
def test_returns_generated_question_instances(self, benchmark_dir: Path) -> None:
|
|
questions = load_benchmark(benchmark_dir)
|
|
assert all(isinstance(q, GeneratedQuestion) for q in questions)
|
|
|
|
def test_loads_real_benchmark(self) -> None:
|
|
"""使用真实 benchmark 数据验证加载正确性。"""
|
|
real_dir = Path("store/questions/benchmarks/Video-MME")
|
|
if not real_dir.exists():
|
|
pytest.skip("真实 benchmark 数据不存在")
|
|
questions = load_benchmark(real_dir)
|
|
assert len(questions) > 0
|
|
for q in questions:
|
|
assert isinstance(q, GeneratedQuestion)
|
|
assert len(q.options) == 4
|
|
assert q.answer in ("A", "B", "C", "D")
|
|
|
|
def test_malformed_json_raises(self, tmp_path: Path) -> None:
|
|
"""非法 JSON 文件应抛出 json.JSONDecodeError。"""
|
|
(tmp_path / "bad.json").write_text("not valid json{{{", encoding="utf-8")
|
|
with pytest.raises(json.JSONDecodeError):
|
|
load_benchmark(tmp_path)
|
|
|
|
def test_missing_required_field_raises(self, tmp_path: Path) -> None:
|
|
"""缺少必需字段（如 question_id）应抛出 KeyError。"""
|
|
data = [{"task_type": "T", "question": "Q?", "options": ["A"], "answer": "A"}]
|
|
(tmp_path / "vid.json").write_text(json.dumps(data), encoding="utf-8")
|
|
with pytest.raises(KeyError):
|
|
load_benchmark(tmp_path)
|