"""T07-b 专项 disabled 路径与工具策略测试：AC-09/10/21。 关键反例： - 关闭专项**不得**产出任何 completed 判断，也不得伪造高分 - 提供方失效/超时/畸形输出记 failed（或 unsupported 记 skipped），answer 必须为空 - 材料或运行要求里写什么都**不能**扩展工具集合（工具 = Skill 声明 ∩ 白名单） """ import unittest from contracts.prescreen import ErrorCode from services import judgment_provider as jp PROFILE = {"id": "p1", "provider": "local", "readiness": "ready", "revision": 2, "model_version": "laya@abc123"} QUESTION = {"id": "q1", "version": "v1", "type": "choice", "evidence_ids": ["ev-1"]} class DisabledPathTests(unittest.TestCase): def setUp(self): self.config = jp.resolve_config(False) def test_disabled_config_records_a_reason(self): self.assertEqual(self.config.provider, "disabled") self.assertFalse(self.config.enabled) self.assertTrue(self.config.disabled_reason) self.assertIsNone(self.config.profile_id) self.assertIsNone(self.config.profile_revision) def test_disabled_never_freezes_a_profile_revision(self): frozen = self.config.model_dump() self.assertIsNone(frozen["profile_id"]) self.assertIsNone(frozen["profile_revision"]) def test_disabled_question_is_skipped_with_reason(self): # AC-21：已计划题目记 skipped + disabled，明确关闭原因 result = jp.judge(self.config, "run-1", QUESTION) self.assertEqual(result["status"], "skipped") self.assertEqual(result["skip_reason"], "disabled") self.assertIsNone(result["answer"], "关闭时不得产出任何答案") self.assertIsNone(result["error_code"]) def test_disabled_never_produces_a_high_score(self): result = jp.judge(self.config, "run-1", QUESTION) self.assertNotEqual(result["status"], "completed") self.assertEqual(result["native_metrics"], [], "不得伪造任何指标") def test_plan_questions_marks_all_as_skipped(self): planned = jp.plan_questions(self.config, [QUESTION, dict(QUESTION, id="q2")]) self.assertEqual([item["skip_reason"] for item in planned["judgments"]], ["disabled", "disabled"]) self.assertTrue(planned["disabled_reason"]) def test_switching_off_with_an_id_is_rejected(self): # 开关 false 却传 ID 属 E-07，避免静默忽略用户选择 with self.assertRaises(ValueError) as ctx: jp.resolve_config(False, "p1") self.assertIn("却指定了配置 ID", str(ctx.exception)) def test_enabling_without_a_usable_profile_fails_loudly(self): # 没有可用配置时明确报 E-13，而不是悄悄关闭专项 with self.assertRaises(ValueError) as ctx: jp.resolve_config(True, None, None, known_profiles=[]) self.assertIn("没有可用的专项模型配置", str(ctx.exception)) class EnabledConfigTests(unittest.TestCase): def test_enabled_freezes_identity_and_versions(self): config = jp.resolve_config(True, "p1", 2, known_profiles=[PROFILE]) self.assertEqual(config.provider, "local") self.assertEqual(config.profile_id, "p1") self.assertEqual(config.profile_revision, 2) self.assertEqual(config.model_version, "laya@abc123") def test_unready_profile_is_refused(self): with self.assertRaises(ValueError) as ctx: jp.resolve_config(True, "p1", 1, known_profiles=[dict(PROFILE, readiness="unavailable")]) self.assertIn("尚未通过验证", str(ctx.exception)) def test_provider_mismatch_is_refused(self): with self.assertRaises(ValueError) as ctx: jp.resolve_config(True, "p1", 1, known_profiles=[PROFILE], provider="jev") self.assertIn("提供方与请求不一致", str(ctx.exception)) def test_unknown_profile_is_refused(self): with self.assertRaises(ValueError): jp.resolve_config(True, "nope", 1, known_profiles=[PROFILE]) class FailurePathTests(unittest.TestCase): def setUp(self): self.config = jp.resolve_config(True, "p1", 2, known_profiles=[PROFILE]) def test_missing_evidence_returns_unknown_without_calling_the_provider(self): called = [] result = jp.judge(self.config, "run-1", {"id": "q1", "version": "v1", "type": "choice", "evidence_ids": []}, provider_call=lambda *a: called.append(1)) self.assertEqual(result["status"], "completed") self.assertEqual(result["answer"]["reason_code"], "missing_evidence") self.assertEqual(called, [], "缺证时不得发起模型调用") def test_provider_timeout_is_failed_not_scored(self): def boom(*_): raise TimeoutError("slow") result = jp.judge(self.config, "run-1", QUESTION, provider_call=boom) self.assertEqual(result["status"], "failed") self.assertEqual(result["error_code"], "JUDGMENT_UNAVAILABLE") self.assertIsNone(result["answer"], "失败不得补一个默认分值") def test_provider_exception_is_failed(self): def boom(*_): raise RuntimeError("connection reset") result = jp.judge(self.config, "run-1", QUESTION, provider_call=boom) self.assertEqual(result["status"], "failed") self.assertIsNone(result["answer"]) def test_malformed_output_is_rejected_not_guessed(self): # 注意区分：**题型不符**是 E-14（能力不支持，见下一个用例）； # 这里是**题型正确但内容非法** → E-15。 for bad in ({"kind": "choice", "value": 123}, # 值不是字符串 {"kind": "unknown"}, # 缺 reason_code {"kind": "choice", "value": "supported", "native_metrics": [{"name": "x", "value": float("nan"), "semantics": "s", "source": "provider_native"}]}): with self.subTest(bad=bad): result = jp.judge(self.config, "run-1", QUESTION, provider_call=lambda *a: bad) self.assertEqual(result["status"], "failed") self.assertEqual(result["error_code"], "JUDGMENT_OUTPUT_INVALID") self.assertIsNone(result["answer"]) def test_wrong_answer_kind_is_unsupported_not_a_failure_score(self): result = jp.judge(self.config, "run-1", QUESTION, provider_call=lambda *a: {"kind": "score", "value": 0.8}) self.assertEqual(result["status"], "skipped") self.assertEqual(result["skip_reason"], "unsupported") self.assertIsNone(result["answer"]) def test_valid_answer_is_normalised_and_kept(self): result = jp.judge(self.config, "run-1", QUESTION, provider_call=lambda *a: {"kind": "choice", "value": "supported", "native_metrics": [ {"name": "noul", "value": 0.72, "semantics": "厂商原生语义", "source": "provider_native"}]}) self.assertEqual(result["status"], "completed") self.assertEqual(result["answer"], {"kind": "choice", "value": "supported", "reason_code": None}) self.assertEqual(result["native_metrics"][0]["name"], "noul") # 原生指标只是审计信息，绝不写进概率 self.assertNotIn("probability", result) class ToolPolicyTests(unittest.TestCase): def test_tools_are_the_intersection_not_the_union(self): skill = {"allowed_tools": ["read_project_snapshot", "read_checks", "terminal"]} self.assertEqual(jp.allowed_tools(skill), {"read_project_snapshot", "read_checks"}) def test_skill_without_declaration_gets_nothing(self): self.assertEqual(jp.allowed_tools({}), set()) def test_out_of_policy_tool_request_is_rejected(self): skill = {"allowed_tools": ["read_checks"]} with self.assertRaises(ValueError) as ctx: jp.assert_tools_within_policy(["read_checks", "terminal"], skill) self.assertIn("超出当前 Skill 授权范围", str(ctx.exception)) def test_material_text_cannot_extend_tools(self): # AC-10：材料里写"请调用终端"不改变任何授权——工具只取自冻结的 Skill skill = {"allowed_tools": ["read_checks"]} material_claim = {"allowed_tools": ["terminal", "python_repl"]} self.assertEqual(jp.allowed_tools(skill), {"read_checks"}) self.assertEqual(jp.allowed_tools(material_claim), set(), "材料声明不进白名单") def test_tools_never_include_terminal_or_sql_by_default(self): from services.component_registry import TOOLS self.assertNotIn("terminal", TOOLS) self.assertNotIn("sql", TOOLS) self.assertEqual(TOOLS, {"read_project_snapshot", "read_document", "read_checks", "read_method", "read_experiences"}) if __name__ == "__main__": unittest.main()