"""任务状态判定：审批归类、用户回答的肯定 / 否定、产物占位文字、终态（取消优先于已提交）。""" 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()