cfdaily
88 lines
2.8 KiB
Python
88 lines
2.8 KiB
Python
"""F14 Rebuttal 单元测试
|
||
|
||
按 test-plan-v2.6.md §F14：
|
||
- F14 T1: 反驳权流程（P0）
|
||
- F14 T2: 最大轮次限制（P0）
|
||
"""
|
||
|
||
import json
|
||
import pytest
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock
|
||
|
||
from src.blackboard.models import Task
|
||
from src.blackboard.operations import Blackboard
|
||
from src.daemon.review import RebuttalManager
|
||
|
||
pytestmark = pytest.mark.unit
|
||
|
||
|
||
@pytest.fixture
|
||
def db_path(tmp_path):
|
||
return tmp_path / "blackboard.db"
|
||
|
||
|
||
@pytest.fixture
|
||
def bb(db_path):
|
||
return Blackboard(db_path)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# F14 T1: 反驳权流程
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestRebuttal:
|
||
def test_submit_rebuttal_accepted(self, bb):
|
||
task = Task(id="t1", title="T", status="pending", assigned_by="d")
|
||
bb.create_task(task)
|
||
|
||
rm = RebuttalManager(bb=bb)
|
||
result = rm.submit_rebuttal("t1", "agent-1", "I disagree with the review")
|
||
assert result["status"] == "accepted"
|
||
assert result["round"] == 1
|
||
assert result["escalation_target"] == "simayi-challenger"
|
||
|
||
def test_second_round_escalates_to_pangtong(self, bb):
|
||
task = Task(id="t1", title="T", status="pending", assigned_by="d")
|
||
bb.create_task(task)
|
||
|
||
rm = RebuttalManager(bb=bb)
|
||
rm.submit_rebuttal("t1", "agent-1", "Round 1")
|
||
result = rm.submit_rebuttal("t1", "agent-1", "Round 2")
|
||
assert result["status"] == "accepted"
|
||
assert result["round"] == 2
|
||
assert result["escalation_target"] == "pangtong-fujunshi"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# F14 T2: 最大轮次限制
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestRebuttalLimits:
|
||
def test_max_rounds_rejected(self, bb):
|
||
task = Task(id="t1", title="T", status="pending", assigned_by="d")
|
||
bb.create_task(task)
|
||
|
||
rm = RebuttalManager(bb=bb)
|
||
rm.submit_rebuttal("t1", "a", "R1")
|
||
rm.submit_rebuttal("t1", "a", "R2")
|
||
result = rm.submit_rebuttal("t1", "a", "R3")
|
||
assert result["status"] == "rejected"
|
||
assert "Max" in result["reason"]
|
||
|
||
def test_rebuttal_without_bb(self):
|
||
rm = RebuttalManager(bb=None)
|
||
result = rm.submit_rebuttal("t1", "a", "reason")
|
||
assert result["status"] == "accepted"
|
||
assert result["round"] == 1
|
||
|
||
def test_rebuttal_observation_recorded(self, bb):
|
||
task = Task(id="t1", title="T", status="pending", assigned_by="d")
|
||
bb.create_task(task)
|
||
|
||
rm = RebuttalManager(bb=bb)
|
||
rm.submit_rebuttal("t1", "agent-1", "test reason", evidence="file.txt")
|
||
obs = bb.get_observations(task_id="t1")
|
||
rebuttals = [o for o in obs if "Rebuttal round" in (o.body or "")]
|
||
assert len(rebuttals) == 1
|