188 lines
9.1 KiB
Python
188 lines
9.1 KiB
Python
 
|
"""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()
|