iomgaa
3cc8dc9105
Capture loguru warning via project sink pattern and assert the dangling-orphan warning is emitted; rename _sift_disk_pairs to _keep_complete_disk_pairs; drop pure-WHAT docstrings on __init__/pending_orphans while keeping WHY notes.
254 lines
9.0 KiB
Python
254 lines
9.0 KiB
Python
"""pair_atomic_writer 纯 helper 单元测试。
|
||
|
||
覆盖三件核心行为：
|
||
- PairPendingBuffer：single 直通成单元、pair 按 pair_id 收齐才 emit、
|
||
未配对孤儿可查询、绑定不一致（unit_hash 语义代偿）拒绝。
|
||
- write_accepted：全量 tmp + os.replace 原子写，落盘无残留 .tmp。
|
||
- read_accepted：build_units + validate_units 剔除磁盘悬挂孤儿（warn+drop）。
|
||
|
||
测试用真实 GeneratedQuestion 构造（single + 合法 AR pair + 故意缺 mirror 的孤儿），
|
||
跑完整 add → write → read 往返。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
|
||
import pytest
|
||
from loguru import logger
|
||
|
||
from app.harness.pools import _q_to_dict
|
||
from app.question_gen.pair_atomic_writer import (
|
||
PairPendingBuffer,
|
||
read_accepted,
|
||
write_accepted,
|
||
)
|
||
from core.types import GeneratedQuestion
|
||
|
||
|
||
def _make_single(qid: str = "s1") -> GeneratedQuestion:
|
||
"""构造一条真实 single 题（pair_id=None，question_role='single'）。"""
|
||
return GeneratedQuestion(
|
||
question_id=qid,
|
||
video_id="vid_A",
|
||
task_type="Action Reasoning",
|
||
question=f"{qid} 问题文本？",
|
||
options=("A. 甲", "B. 乙", "C. 丙", "D. 丁"),
|
||
answer="B",
|
||
source_nodes=("L2_0", "L3_1"),
|
||
difficulty="medium",
|
||
)
|
||
|
||
|
||
def _make_pair(
|
||
pair_id: str = "p1",
|
||
*,
|
||
video_id: str = "vid_B",
|
||
flip_axis: str = "before_after",
|
||
) -> tuple[GeneratedQuestion, GeneratedQuestion]:
|
||
"""构造一对合法 AR 孪生对（original + mirror，共享 pair_id/flip_axis）。"""
|
||
original = GeneratedQuestion(
|
||
question_id=f"{pair_id}_o",
|
||
video_id=video_id,
|
||
task_type="Action Reasoning",
|
||
question="事件 X 发生在事件 Y 之前吗？",
|
||
options=("A. 是", "B. 否", "C. 无关", "D. 无法判断"),
|
||
answer="A",
|
||
source_nodes=("L2_2",),
|
||
difficulty="hard",
|
||
pair_id=pair_id,
|
||
question_role="pair_original",
|
||
flip_axis=flip_axis,
|
||
)
|
||
mirror = GeneratedQuestion(
|
||
question_id=f"{pair_id}_m",
|
||
video_id=video_id,
|
||
task_type="Action Reasoning",
|
||
question="事件 Y 发生在事件 X 之前吗？",
|
||
options=("A. 是", "B. 否", "C. 无关", "D. 无法判断"),
|
||
answer="B",
|
||
source_nodes=("L2_2",),
|
||
difficulty="hard",
|
||
pair_id=pair_id,
|
||
question_role="pair_mirror",
|
||
flip_axis=flip_axis,
|
||
)
|
||
return original, mirror
|
||
|
||
|
||
def test_single_add_emits_unit_immediately() -> None:
|
||
"""single 题 add 后立即返回 kind='single' 单元，不进 pending。"""
|
||
buf = PairPendingBuffer()
|
||
unit = buf.add(_make_single())
|
||
assert unit is not None
|
||
assert unit.kind == "single"
|
||
assert unit.size == 1
|
||
assert buf.pending_orphans() == []
|
||
|
||
|
||
def test_pair_emits_only_when_both_arrived() -> None:
|
||
"""pair 首个成员进 pending 返回 None，第二个到齐才 emit 一个 pair 单元。"""
|
||
buf = PairPendingBuffer()
|
||
original, mirror = _make_pair()
|
||
|
||
assert buf.add(original) is None
|
||
assert len(buf.pending_orphans()) == 1
|
||
|
||
unit = buf.add(mirror)
|
||
assert unit is not None
|
||
assert unit.kind == "pair"
|
||
assert unit.size == 2
|
||
assert buf.pending_orphans() == []
|
||
|
||
|
||
def test_pair_completes_regardless_of_arrival_order() -> None:
|
||
"""mirror 先到、original 后到也能正确成对。"""
|
||
buf = PairPendingBuffer()
|
||
original, mirror = _make_pair()
|
||
|
||
assert buf.add(mirror) is None
|
||
unit = buf.add(original)
|
||
assert unit is not None and unit.kind == "pair" and unit.size == 2
|
||
|
||
|
||
def test_dangling_original_stays_in_pending() -> None:
|
||
"""只落 original 未落 mirror → 悬挂在 pending，供调用方批次末尾检测。"""
|
||
buf = PairPendingBuffer()
|
||
original, _ = _make_pair()
|
||
assert buf.add(original) is None
|
||
orphans = buf.pending_orphans()
|
||
assert len(orphans) == 1
|
||
assert orphans[0].question_id == original.question_id
|
||
|
||
|
||
def test_inconsistent_pair_binding_rejected_and_partner_recoverable() -> None:
|
||
"""两成员 flip_axis 不一致 → 拒绝；且校验失败后首成员仍留 pending 可取回。"""
|
||
buf = PairPendingBuffer()
|
||
original, _ = _make_pair(flip_axis="before_after")
|
||
_, bad_mirror = _make_pair(flip_axis="left_right")
|
||
|
||
buf.add(original)
|
||
with pytest.raises(ValueError, match="绑定不一致"):
|
||
buf.add(bad_mirror)
|
||
|
||
# Important: 先校验后删除——raise 后首成员未被驱逐，调用方仍能检测到悬挂项
|
||
orphans = buf.pending_orphans()
|
||
assert len(orphans) == 1
|
||
assert orphans[0].question_id == original.question_id
|
||
|
||
|
||
def test_same_role_pair_rejected() -> None:
|
||
"""同 pair_id 两成员角色相同（两个 original）→ fail-fast 拒绝。"""
|
||
buf = PairPendingBuffer()
|
||
o1, _ = _make_pair()
|
||
o2, _ = _make_pair()
|
||
buf.add(o1)
|
||
with pytest.raises(ValueError):
|
||
buf.add(o2)
|
||
|
||
|
||
def test_write_read_roundtrip_keeps_pair_and_single(tmp_path) -> None:
|
||
"""add → write_accepted → read_accepted 往返：single 直通、pair 不拆、内容等价。"""
|
||
buf = PairPendingBuffer()
|
||
single = _make_single()
|
||
original, mirror = _make_pair()
|
||
|
||
units = []
|
||
for q in (single, original, mirror):
|
||
emitted = buf.add(q)
|
||
if emitted is not None:
|
||
units.append(emitted)
|
||
assert buf.pending_orphans() == []
|
||
assert len(units) == 2 # 1 single + 1 pair
|
||
|
||
out = tmp_path / "accepted" / "bank.json"
|
||
write_accepted(out, units)
|
||
|
||
assert out.exists()
|
||
assert not out.with_suffix(".tmp").exists() # 原子写无残留
|
||
|
||
loaded = read_accepted(out)
|
||
kinds = sorted(u.kind for u in loaded)
|
||
assert kinds == ["pair", "single"]
|
||
|
||
loaded_pair = next(u for u in loaded if u.kind == "pair")
|
||
assert loaded_pair.size == 2
|
||
assert {q.question_id for q in loaded_pair.questions} == {
|
||
original.question_id,
|
||
mirror.question_id,
|
||
}
|
||
assert {q.flip_axis for q in loaded_pair.questions} == {"before_after"}
|
||
|
||
|
||
def test_read_drops_disk_dangling_orphan(tmp_path) -> None:
|
||
"""磁盘上某 pair 只有 original（缺 mirror）→ read_accepted warn+drop，不 raise。"""
|
||
single = _make_single()
|
||
original, _ = _make_pair()
|
||
# 手工写入"只落 P 未落 Q"的记录：single + 孤零 original
|
||
records = [_q_to_dict(single), _q_to_dict(original)]
|
||
out = tmp_path / "bank.json"
|
||
out.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
# loguru 不走标准 logging，用项目既定 sink 捕获模式（见 test_inference_pair_aggregate）
|
||
captured: list[str] = []
|
||
sink_id = logger.add(captured.append, level="WARNING", format="{message}")
|
||
try:
|
||
loaded = read_accepted(out)
|
||
finally:
|
||
logger.remove(sink_id)
|
||
|
||
assert len(loaded) == 1
|
||
assert loaded[0].kind == "single"
|
||
assert loaded[0].questions[0].question_id == single.question_id
|
||
# warn+drop 契约的 "warn" 半：悬挂孤儿必须告警，不静默丢弃
|
||
assert any("悬挂孤儿" in msg for msg in captured), "悬挂孤儿未告警（静默丢弃）"
|
||
|
||
|
||
def test_read_raises_on_disk_corrupt_duplicate_role(tmp_path) -> None:
|
||
"""磁盘上同 pair_id 出现两个 pair_original（角色重复）→ 数据损坏，fail-fast raise。"""
|
||
o1, _ = _make_pair()
|
||
o2, _ = _make_pair()
|
||
# 同 pair_id、两条都是 pair_original（size==2 但角色非法）→ 结构损坏
|
||
dup = GeneratedQuestion(
|
||
question_id="p1_o2",
|
||
video_id=o2.video_id,
|
||
task_type=o2.task_type,
|
||
question="重复 original",
|
||
options=o2.options,
|
||
answer=o2.answer,
|
||
source_nodes=o2.source_nodes,
|
||
difficulty=o2.difficulty,
|
||
pair_id=o1.pair_id,
|
||
question_role="pair_original",
|
||
flip_axis=o1.flip_axis,
|
||
)
|
||
records = [_q_to_dict(o1), _q_to_dict(dup)]
|
||
out = tmp_path / "corrupt.json"
|
||
out.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
with pytest.raises(ValueError, match="结构损坏"):
|
||
read_accepted(out)
|
||
|
||
|
||
def test_read_raises_on_disk_binding_mismatch(tmp_path) -> None:
|
||
"""磁盘上合法成对但 flip_axis 绑定不一致 → 显式 raise（不依赖会被 -O 剥除的 assert）。"""
|
||
original, _ = _make_pair(pair_id="p1", flip_axis="before_after")
|
||
_, mirror = _make_pair(pair_id="p1", flip_axis="left_right")
|
||
# 直接写盘绕过 write_accepted 的 from_pair 组装，模拟外部篡改的不一致成对
|
||
records = [_q_to_dict(original), _q_to_dict(mirror)]
|
||
out = tmp_path / "mismatch.json"
|
||
out.write_text(json.dumps(records, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
with pytest.raises(ValueError, match="绑定不一致"):
|
||
read_accepted(out)
|
||
|
||
|
||
def test_write_accepted_rejects_malformed_unit(tmp_path) -> None:
|
||
"""validate_units 闸门：结构非法的 pair 单元（size≠2）落盘前 fail-fast。"""
|
||
from core.types import QuestionUnit
|
||
|
||
original, _ = _make_pair()
|
||
bad_unit = QuestionUnit("pair", "p1", "Action Reasoning", (original,))
|
||
with pytest.raises(ValueError):
|
||
write_accepted(tmp_path / "bad.json", [bad_unit])
|