113 lines
7.9 KiB
Python
113 lines
7.9 KiB
Python
import pytest
|
|
from test_support.external_qa import require_external_qa
|
|
import json
|
|
from pathlib import Path
|
|
from yuqei_sdk.contract_intake import (ContractIntakeRequest, ContractIntakeOutput,
|
|
QUESTION_COMPLETION_VERSION, complete_required_questions, validate_intake_output)
|
|
|
|
def request(**changes):
|
|
data={"text":"费用为100元，开始日期2026-09-10。", "template_snapshot":{"id":"service","content_sha256":"a"*64,
|
|
"input_schema":{"version":"template-input.v1","fields":{"费用":{"data_type":"money"},"开始日期":{"data_type":"date"},"启用":{"data_type":"choice","options":["是","否"]}},"rules":[]},
|
|
"required_variables":["费用"],"optional_variables":["开始日期","启用"]},"existing_values":{}}
|
|
data.update(changes); return ContractIntakeRequest.model_validate(data)
|
|
|
|
def output(proposals):
|
|
return ContractIntakeOutput.model_validate({"schema_version":"contract-intake.v1","proposals":proposals,"questions":[],"applicability":"compatible","warnings":[]})
|
|
|
|
def test_grounded_typed_proposal_passes():
|
|
result=output([{"field":"费用","value_json":"100","source_excerpt":"费用为100元","state":"extracted","explanation":"原文明示"}])
|
|
assert validate_intake_output(request(), result)==[]
|
|
|
|
@pytest.mark.parametrize("value,error", [("NaN","合法 JSON"),("true","有限数值"),("101","来源摘录")])
|
|
def test_invalid_values_are_explicit(value,error):
|
|
excerpt="不存在" if value=="101" else "费用为100元"
|
|
result=output([{"field":"费用","value_json":value,"source_excerpt":excerpt,"state":"extracted","explanation":"测试"}])
|
|
assert error in " ".join(validate_intake_output(request(), result))
|
|
|
|
def test_existing_value_conflict_is_resolved_by_apply_flow_not_schema_rejected():
|
|
result=output([{"field":"费用","value_json":"100","source_excerpt":"费用为100元","state":"extracted","explanation":"测试"}])
|
|
assert validate_intake_output(request(existing_values={"费用":99}), result)==[]
|
|
assert validate_intake_output(request(existing_values={"费用":""}), result)==[]
|
|
|
|
def test_required_question_completion_treats_empty_existing_as_missing_without_inventing_fact():
|
|
intake=request(existing_values={"费用":""})
|
|
raw=ContractIntakeOutput.model_validate({**output([]).model_dump(),"questions":[]})
|
|
normalized, added, origins=complete_required_questions(intake,raw)
|
|
assert added==["费用"] and normalized.proposals==raw.proposals
|
|
assert normalized.questions[0].model_dump()=={"field":"费用","question":"请补充模板必填字段：费用。","reason":"missing"}
|
|
assert origins==[{"field":"费用","reason":"missing","origin":"template_required"}]
|
|
assert QUESTION_COMPLETION_VERSION=="contract-intake-question-completion.v1"
|
|
assert validate_intake_output(intake,normalized)==[]
|
|
|
|
def test_required_question_completion_skips_excluded_and_does_not_mask_bad_model_questions():
|
|
excluded=ContractIntakeOutput.model_validate({**output([]).model_dump(),"applicability":"excluded"})
|
|
assert complete_required_questions(request(),excluded)[1]==[]
|
|
wrong=ContractIntakeOutput.model_validate({**output([{"field":"费用","value_json":"[100,200]","source_excerpt":"费用为100元",
|
|
"state":"ambiguous","explanation":"有两个金额"}]).model_dump(),
|
|
"questions":[{"field":"费用","question":"请确认","reason":"missing"}]})
|
|
normalized, added, _=complete_required_questions(request(),wrong)
|
|
assert added==[]
|
|
assert "歧义事实必须提供关联问题" in " ".join(validate_intake_output(request(),normalized))
|
|
|
|
def test_duplicate_model_questions_remain_invalid_after_completion():
|
|
raw=ContractIntakeOutput.model_validate({**output([]).model_dump(),"questions":[
|
|
{"field":"费用","question":"请补充费用","reason":"missing"},
|
|
{"field":"费用","question":"费用是多少","reason":"missing"}]})
|
|
normalized, _, _=complete_required_questions(request(),raw)
|
|
assert "重复问题" in " ".join(validate_intake_output(request(),normalized))
|
|
|
|
def test_whitespace_text_rejected():
|
|
with pytest.raises(ValueError): request(text=" ")
|
|
|
|
def test_ambiguous_money_candidates_are_each_typed_but_extracted_array_is_rejected():
|
|
proposal={"field":"费用","value_json":"[50000,60000]","source_excerpt":"费用为100元","state":"ambiguous","explanation":"两个候选"}
|
|
ambiguous=ContractIntakeOutput.model_validate({**output([proposal]).model_dump(),"questions":[{"field":"费用","question":"请选择金额","reason":"ambiguous"}]})
|
|
assert validate_intake_output(request(), ambiguous)==[]
|
|
extracted=output([{**proposal,"state":"extracted"}])
|
|
assert "有限数值" in " ".join(validate_intake_output(request(), extracted))
|
|
|
|
def test_ambiguous_candidates_still_reject_nonfinite_member():
|
|
proposal={"field":"费用","value_json":"[50000,1e999]","source_excerpt":"费用为100元","state":"ambiguous","explanation":"测试"}
|
|
result=ContractIntakeOutput.model_validate({**output([proposal]).model_dump(),"questions":[{"field":"费用","question":"请选择金额","reason":"ambiguous"}]})
|
|
assert "必须有限" in " ".join(validate_intake_output(request(), result))
|
|
|
|
@pytest.mark.parametrize("value_json", ["[]", "[null]", "[\"\"]", "[true]", "[1e999]"])
|
|
def test_ambiguous_money_rejects_empty_or_invalid_candidates(value_json):
|
|
proposal={"field":"费用","value_json":value_json,"source_excerpt":"费用为100元","state":"ambiguous","explanation":"测试"}
|
|
result=ContractIntakeOutput.model_validate({**output([proposal]).model_dump(),"questions":[{"field":"费用","question":"请选择金额","reason":"ambiguous"}]})
|
|
assert validate_intake_output(request(), result)
|
|
|
|
def test_postfix_regression_replays_run01_failure_without_mutating_artifact():
|
|
require_external_qa(
|
|
"template-prompt-audit-20260828/tp06-11-intake/run-01/INTAKE-003-ambiguous-injection.json",
|
|
"template-prompt-audit-20260828/tp06-11-intake/frozen-v1.json",
|
|
)
|
|
root=Path(__file__).resolve().parents[2]
|
|
artifact_path=root/"docs/qa/template-prompt-audit-20260828/tp06-11-intake/run-01/INTAKE-003-ambiguous-injection.json"
|
|
frozen_path=root/"docs/qa/template-prompt-audit-20260828/tp06-11-intake/frozen-v1.json"
|
|
before=artifact_path.read_bytes(); artifact=json.loads(before); frozen=json.loads(frozen_path.read_text(encoding="utf-8"))
|
|
case=next(item for item in frozen["cases"] if item["id"]==artifact["case_id"])
|
|
replay_errors=validate_intake_output(ContractIntakeRequest.model_validate(case["request"]),
|
|
ContractIntakeOutput.model_validate(artifact["job"]["result"]["output"]))
|
|
assert replay_errors==[] # postfix_regression: original assessment remains immutable and failed.
|
|
assert artifact_path.read_bytes()==before and artifact["assessment"]["passed"] is False
|
|
|
|
def test_question_completion_replays_known_postfix_missing_failure_without_rewriting_evidence():
|
|
require_external_qa(
|
|
"template-prompt-audit-20260828/tp06-12/intake-postfix-run-01/INTAKE-002-missing.json",
|
|
"template-prompt-audit-20260828/tp06-11-intake/frozen-v1.json",
|
|
)
|
|
root=Path(__file__).resolve().parents[2]
|
|
artifact_path=root/"docs/qa/template-prompt-audit-20260828/tp06-12/intake-postfix-run-01/INTAKE-002-missing.json"
|
|
frozen_path=root/"docs/qa/template-prompt-audit-20260828/tp06-11-intake/frozen-v1.json"
|
|
before=artifact_path.read_bytes(); artifact=json.loads(before); frozen=json.loads(frozen_path.read_text(encoding="utf-8"))
|
|
case=next(item for item in frozen["cases"] if item["id"]==artifact["case_id"])
|
|
intake=ContractIntakeRequest.model_validate(case["request"])
|
|
raw=ContractIntakeOutput.model_validate(artifact["job"]["result"]["output"])
|
|
normalized, added, origins=complete_required_questions(intake,raw)
|
|
assert len(added)==15
|
|
assert all(item["origin"]=="template_required" for item in origins if item["field"] in added)
|
|
assert validate_intake_output(intake,normalized)==[]
|
|
assert raw.proposals==normalized.proposals
|
|
assert artifact_path.read_bytes()==before and artifact["assessment"]["passed"] is False
|