193 lines
9.4 KiB
Python
193 lines
9.4 KiB
Python
"""任务状态判定：审批归类、用户回答的肯定 / 否定、产物占位文字、终态（取消优先于已提交）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from backend.orchestrator import orchestrator as orc
|
||
|
||
|
||
class TestClassifyApproval:
|
||
@pytest.mark.parametrize("text", ["已通过", "审批通过", "同意", "已审批", "办结", "批准", "已生效", "归档"])
|
||
def test_approved(self, text):
|
||
assert orc._classify_approval(text) == "approved"
|
||
|
||
@pytest.mark.parametrize("text", ["已驳回", "退回", "拒绝", "否决", "作废", "撤销"])
|
||
def test_rejected(self, text):
|
||
assert orc._classify_approval(text) == "rejected"
|
||
|
||
@pytest.mark.parametrize("text", ["待审批", "未审批", "审批中", "待审核", "审核中", "处理中", "已提交"])
|
||
def test_pending(self, text):
|
||
assert orc._classify_approval(text) == "pending"
|
||
|
||
@pytest.mark.parametrize("text", ["审批不通过", "未通过", "审核不通过"])
|
||
def test_negated_approval_is_rejected_not_approved(self, text):
|
||
# 「审批不通过」同时含「通过」和「不通过」，驳回词必须优先，否则驳回会被当成通过
|
||
assert orc._classify_approval(text) == "rejected"
|
||
|
||
@pytest.mark.parametrize("text", ["", " ", None, "页面上什么状态都没写"])
|
||
def test_unknown(self, text):
|
||
assert orc._classify_approval(text) == "unknown"
|
||
|
||
def test_pending_beats_approved_when_both_present(self):
|
||
# 「已提交，等待审批通过」还没批下来，不能记成 approved
|
||
assert orc._classify_approval("已提交，等待审批通过") == "pending"
|
||
|
||
|
||
class TestAnswerPolarity:
|
||
@pytest.mark.parametrize("answer", ["是", "好的", "可以", "行", "去办吧", "嗯", "对", "同意", "确认提交", "已处理", "继续", "ok", "YES"])
|
||
def test_affirmative(self, answer):
|
||
assert orc._is_affirmative(answer) is True
|
||
assert orc._is_negative(answer) is False
|
||
|
||
@pytest.mark.parametrize("answer", ["否", "不用了", "别办", "取消", "算了", "先不办", "no"])
|
||
def test_negative(self, answer):
|
||
assert orc._is_negative(answer) is True
|
||
assert orc._is_affirmative(answer) is False
|
||
|
||
def test_cancel_phrase_from_the_confirm_card(self):
|
||
# 确认卡片上「取消，不要提交」按钮送出的原话
|
||
assert orc._is_negative("取消，不要提交") is True
|
||
assert orc._is_affirmative("取消，不要提交") is False
|
||
|
||
@pytest.mark.parametrize("answer", ["不错，就这样", "不用担心，继续"])
|
||
def test_polite_phrases_are_not_refusals(self, answer):
|
||
# 「不错」「不用担心」以「不」开头但表示同意，不能当成拒绝
|
||
assert orc._is_negative(answer) is False
|
||
|
||
@pytest.mark.parametrize("answer", ["课程是高等数学", "听课地点在教学楼", "我来接管"])
|
||
def test_unrelated_answer_is_neither(self, answer):
|
||
# 回归：「是 / 好 / 行 / 办 / 去 / 对」这类单字词曾经在句中任意位置命中，
|
||
# 补充资料的回答会被当成「同意先办前置事项」，把任务带去办别的业务
|
||
assert orc._is_affirmative(answer) is False
|
||
assert orc._is_negative(answer) is False
|
||
|
||
def test_single_character_yes_still_works_at_the_start(self):
|
||
for answer in ("是", "好", "行", "办吧", "去办", "对"):
|
||
assert orc._is_affirmative(answer) is True
|
||
|
||
|
||
class TestMeaningfulArtifact:
|
||
@pytest.mark.parametrize("value", ["已提交成功", "计科2301班", "0", "查课听课记录"])
|
||
def test_real_values(self, value):
|
||
assert orc._meaningful(value) is True
|
||
|
||
@pytest.mark.parametrize("value", ["未获取", "未找到入口", "无法读取表单", "失败", "暂无", "N/A", "none", "null", "", " ", None])
|
||
def test_placeholder_values_do_not_count(self, value):
|
||
# Agent 有时会把「未获取到表单字段」写进产物，不能据此判定这一步成功了
|
||
assert orc._meaningful(value) is False
|
||
|
||
|
||
class TestCriteriaMet:
|
||
def test_any_declared_artifact_is_enough(self):
|
||
step = {"success_artifacts_any": ["form_fields", "menus"]}
|
||
assert orc._criteria_met(step, {"menus": "门户有 8 个模块"}) is True
|
||
|
||
def test_placeholder_artifact_does_not_satisfy(self):
|
||
step = {"success_artifacts_any": ["form_fields"]}
|
||
assert orc._criteria_met(step, {"form_fields": "未获取"}) is False
|
||
|
||
def test_no_declared_artifacts_means_not_met(self):
|
||
assert orc._criteria_met({}, {"anything": "x"}) is False
|
||
|
||
def test_unrelated_artifacts_do_not_satisfy(self):
|
||
step = {"success_artifacts_any": ["form_fields"]}
|
||
assert orc._criteria_met(step, {"menus": "x"}) is False
|
||
|
||
|
||
class TestVerdictOverrides:
|
||
"""Agent 的 is_successful 不稳定，客观产物要能纠正它的自评。"""
|
||
|
||
STEP = {"step_id": "fill", "success_artifacts_any": ["submitted_record", "result_message"]}
|
||
|
||
def test_success_is_left_alone(self):
|
||
verdict, summary, note = orc.apply_verdict_overrides("success", True, self.STEP, {}, "ok")
|
||
assert (verdict, summary, note) == ("success", "ok", None)
|
||
|
||
def test_user_cancel_is_not_a_failure(self):
|
||
# 回归：这一步没声明 success_artifacts_any，取消后 Agent 自评失败就直接落到「办理未完成」；
|
||
# 同一个取消动作曾经一次记成已完成、一次记成办理未完成
|
||
step = {"step_id": "fill"} # 没有 success_artifacts_any
|
||
verdict, _, note = orc.apply_verdict_overrides("failed", True, step, {"cancelled": "取消，不要提交"}, "未提交")
|
||
assert verdict == "success"
|
||
assert note[0] == "用户取消"
|
||
|
||
def test_cancel_override_works_even_with_declared_criteria(self):
|
||
verdict, _, _ = orc.apply_verdict_overrides("failed", True, self.STEP, {"cancelled": "取消"}, "未提交")
|
||
assert verdict == "success"
|
||
|
||
def test_placeholder_cancel_value_does_not_override(self):
|
||
verdict, _, note = orc.apply_verdict_overrides("failed", True, self.STEP, {"cancelled": "无法确定"}, "失败")
|
||
assert verdict == "failed" and note is None
|
||
|
||
def test_declared_criteria_still_override_a_self_assessed_failure(self):
|
||
artifacts = {"result_message": "提交成功"}
|
||
verdict, summary, note = orc.apply_verdict_overrides("failed", True, self.STEP, artifacts, "没把握")
|
||
assert verdict == "success"
|
||
assert "产物已满足本步完成判据" in summary
|
||
assert note[0] == "判定覆盖"
|
||
|
||
def test_genuine_failure_is_not_rescued(self):
|
||
verdict, _, note = orc.apply_verdict_overrides("failed", True, self.STEP, {"menus": "找了一圈"}, "没找到入口")
|
||
assert verdict == "failed" and note is None
|
||
|
||
def test_agent_that_never_finished_is_not_rescued(self):
|
||
# 跑满步数 / 中途报错（agent_done=False）时不能靠产物翻案
|
||
verdict, _, _ = orc.apply_verdict_overrides("failed", False, self.STEP, {"cancelled": "取消"}, "超步数")
|
||
assert verdict == "failed"
|
||
|
||
|
||
class TestFinalStatus:
|
||
SUBMIT_STEP = {"step_id": "fill", "on_success_status": "submitted"}
|
||
|
||
def test_submitted_when_submit_artifact_present(self):
|
||
assert orc._final_status([self.SUBMIT_STEP], {"submitted_record": "课程：高等数学"}) == "submitted"
|
||
|
||
def test_result_message_also_counts_as_submitted(self):
|
||
assert orc._final_status([self.SUBMIT_STEP], {"result_message": "提交成功"}) == "submitted"
|
||
|
||
def test_cancelled_wins_over_submit_artifacts(self):
|
||
# 用户在确认卡点了取消；即使 Agent 顺手存了 submitted_record 也不能记成已提交
|
||
artifacts = {"cancelled": "用户说不要提交", "submitted_record": "课程：高等数学"}
|
||
assert orc._final_status([self.SUBMIT_STEP], artifacts) == "completed"
|
||
|
||
def test_submit_step_without_artifacts_is_only_completed(self):
|
||
assert orc._final_status([self.SUBMIT_STEP], {"summary": "填完了没提交"}) == "completed"
|
||
|
||
def test_read_only_step_never_reports_submitted(self):
|
||
steps = [{"step_id": "explore", "success_artifacts_any": ["menus"]}]
|
||
assert orc._final_status(steps, {"menus": "8 个模块", "submitted_record": "x"}) == "completed"
|
||
|
||
def test_custom_submitted_artifact_keys_are_honoured(self):
|
||
step = {"on_success_status": "submitted", "submitted_artifacts_any": ["ticket_no"]}
|
||
assert orc._final_status([step], {"ticket_no": "A123"}) == "submitted"
|
||
assert orc._final_status([step], {"submitted_record": "x"}) == "completed"
|
||
|
||
|
||
class TestTrackingConfig:
|
||
def test_reads_business_tracking_block(self, business):
|
||
cfg = orc._tracking_cfg(business)
|
||
assert cfg["interval_minutes"] == 60
|
||
assert cfg["max_checks"] == 24
|
||
assert cfg["step"]["step_id"] == "track_status" # 没写 step_id 时给默认值
|
||
|
||
def test_business_without_tracking(self):
|
||
assert orc._tracking_cfg({"id": "x"}) is None
|
||
|
||
def test_tracking_without_step_is_ignored(self):
|
||
assert orc._tracking_cfg({"tracking": {"interval_minutes": 30}}) is None
|
||
|
||
|
||
def test_next_check_is_in_the_future():
|
||
from datetime import datetime
|
||
|
||
nxt = datetime.strptime(orc._next_check_str(60), "%Y-%m-%d %H:%M:%S")
|
||
assert nxt > datetime.now()
|
||
|
||
|
||
def test_next_check_never_schedules_in_the_past():
|
||
from datetime import datetime
|
||
|
||
nxt = datetime.strptime(orc._next_check_str(0), "%Y-%m-%d %H:%M:%S")
|
||
assert nxt > datetime.now()
|