"""S1 契约层测试：材料集合导入与事实确认。 覆盖 T00/04 §9 的 AC-23 / AC-24 可机械断言部分，以及契约自身的边界。 每条规则都有反例——只测通过路径不能证明约束存在。 """ import unittest from datetime import datetime, timezone from pydantic import ValidationError from contracts.prescreen import ( MAX_BATCH_BYTES, MAX_BATCH_FILES, MAX_ITEM_BYTES, ConflictCandidate, DataDomain, ErrorCode, ErrorResponse, FactConflict, ImportBatch, ImportItem, Ref, S1_ERROR_HTTP, S1_ERROR_RETRYABLE, SourceLocator, ) NOW = datetime(2026, 9, 29, 12, 0, tzinfo=timezone.utc) def locator(**over): base = dict(kind="pdf", document_id="doc-1", document_version=1, page=3) base.update(over) return base def item(**over): base = dict( id="item-1", relative_path="materials/a.pdf", sha256="a" * 64, size_bytes=1024, mime_type="application/pdf", status="queued", received_at=NOW, ) base.update(over) return base def batch(**over): base = dict( id="batch-1", project_id="DEMO-A", data_domain="demo", revision=1, status="queued", created_at=NOW, created_by="demo-manager", parser_version="assistant_materials@1", items=[item()], total_bytes=1024, ) base.update(over) return base def conflict(**over): ids = ["ev-1", "ev-2"] base = dict( id="fc-1", project_id="DEMO-A", data_domain="demo", revision=1, field="facility.amount", candidate_evidence_ids=list(ids), candidates=[ ConflictCandidate(evidence_id="ev-1", value="1000", source="a.csv", locator="第2行"), ConflictCandidate(evidence_id="ev-2", value="1200", source="b.csv", locator="第5行"), ], status="unresolved", created_at=NOW, ) base.update(over) return base class SharedConstraintTests(unittest.TestCase): def test_unknown_field_is_rejected(self): # E-07：所有输入禁止未知字段 with self.assertRaises(ValidationError): ImportItem(**item(), unexpected="x") def test_naive_datetime_is_rejected(self): with self.assertRaises(ValidationError): ImportItem(**item(received_at=datetime(2026, 9, 29, 12, 0))) def test_ref_requires_both_fields(self): self.assertEqual(Ref(id="r-1", version="v1").version, "v1") with self.assertRaises(ValidationError): Ref(id="r-1") def test_data_domain_enum_values(self): self.assertEqual({d.value for d in DataDomain}, {"demo", "production"}) with self.assertRaises(ValidationError): ImportBatch(**batch(data_domain="staging")) class SourceLocatorTests(unittest.TestCase): def test_each_kind_accepts_its_own_fields(self): self.assertEqual(SourceLocator(**locator()).page, 3) self.assertEqual( SourceLocator(**locator(kind="table", page=None, sheet="Sheet1", cell_range="C4")).sheet, "Sheet1") self.assertEqual( SourceLocator(**locator(kind="manual", page=None, note="正文第2段")).note, "正文第2段") def test_missing_required_field_per_kind(self): for over, missing in [ (dict(page=None), "page"), (dict(kind="table", page=None), "sheet"), (dict(kind="table", page=None, sheet="S"), "cell_range"), (dict(kind="manual", page=None), "note"), ]: with self.subTest(missing=missing): with self.assertRaises(ValidationError) as ctx: SourceLocator(**locator(**over)) self.assertIn(missing, str(ctx.exception)) def test_cross_kind_fields_are_rejected(self): # 三种寻址方式互斥，不允许混填 with self.assertRaises(ValidationError): SourceLocator(**locator(kind="table", sheet="S", cell_range="C4")) # page 也在 with self.assertRaises(ValidationError): SourceLocator(**locator(note="顺带写个说明")) def test_page_must_be_positive(self): with self.assertRaises(ValidationError): SourceLocator(**locator(page=0)) class ImportItemTests(unittest.TestCase): def test_valid_item(self): self.assertEqual(ImportItem(**item()).status, "queued") def test_absolute_paths_rejected(self): for bad in ["/etc/passwd", "C:/Windows/x.pdf", "C:\\Windows\\x.pdf", "//share/x.pdf"]: with self.subTest(path=bad): with self.assertRaises(ValidationError): ImportItem(**item(relative_path=bad)) def test_parent_reference_rejected(self): with self.assertRaises(ValidationError): ImportItem(**item(relative_path="../../secret.pdf")) with self.assertRaises(ValidationError): ImportItem(**item(relative_path="a/../../b.pdf")) def test_backslash_relative_path_is_kept_verbatim(self): # 校验用规范化副本，但保留原值：落库内容须与客户端提交一致 parsed = ImportItem(**item(relative_path="materials\\sub\\a.pdf")) self.assertEqual(parsed.relative_path, "materials\\sub\\a.pdf") def test_sha256_shape(self): with self.assertRaises(ValidationError): ImportItem(**item(sha256="A" * 64)) # 大写不合法 with self.assertRaises(ValidationError): ImportItem(**item(sha256="a" * 63)) def test_single_file_over_limit_rejected(self): with self.assertRaises(ValidationError): ImportItem(**item(size_bytes=MAX_ITEM_BYTES + 1)) def test_failed_requires_error_code_and_parsed_forbids_it(self): with self.assertRaises(ValidationError): ImportItem(**item(status="failed")) self.assertEqual( ImportItem(**item(status="failed", error_code=ErrorCode.MATERIAL_PARSE_FAILED)).status, "failed") with self.assertRaises(ValidationError): ImportItem(**item(status="parsed", document_id="doc-1", error_code=ErrorCode.MATERIAL_PARSE_FAILED)) def test_parsed_requires_document_id(self): # AC-23：解析成功必须能定位到既有 document with self.assertRaises(ValidationError): ImportItem(**item(status="parsed")) self.assertEqual( ImportItem(**item(status="parsed", document_id="doc-9")).document_id, "doc-9") def test_observed_at_may_be_null_and_is_not_faked(self): # BR-02：缺发生时间保留 null，不用导入时间冒充 parsed = ImportItem(**item()) self.assertIsNone(parsed.observed_at) self.assertIsNotNone(parsed.received_at) class ImportBatchTests(unittest.TestCase): def test_valid_batch(self): self.assertEqual(ImportBatch(**batch()).total_bytes, 1024) def test_total_bytes_must_match_sum(self): with self.assertRaises(ValidationError): ImportBatch(**batch(total_bytes=999)) def test_duplicate_item_ids_rejected(self): with self.assertRaises(ValidationError): ImportBatch(**batch( items=[item(), item()], total_bytes=2048)) def test_empty_items_rejected(self): # E-16：空集合不合法 with self.assertRaises(ValidationError): ImportBatch(**batch(items=[], total_bytes=0)) def test_over_file_count_rejected(self): with self.assertRaises(ValidationError): ImportBatch(**batch(items=[], total_bytes=0)) # min_length=1 已拦截 items = [item(id=f"i-{n}", size_bytes=1) for n in range(MAX_BATCH_FILES + 1)] with self.assertRaises(ValidationError): ImportBatch(**batch(items=items, total_bytes=len(items))) def test_batch_total_limit_enforced(self): # D3 决策：单批 ≤ 20 MiB。每个文件合法但合计超限必须拒绝。 each = MAX_BATCH_BYTES // 3 + 1 items = [item(id=f"i-{n}", size_bytes=each) for n in range(3)] self.assertLessEqual(each, MAX_ITEM_BYTES) with self.assertRaises(ValidationError) as ctx: ImportBatch(**batch(items=items, total_bytes=each * 3)) self.assertIn("拆分", str(ctx.exception)) def test_status_enum_bounded(self): with self.assertRaises(ValidationError): ImportBatch(**batch(status="done")) class FactConflictTests(unittest.TestCase): def test_valid_unresolved(self): parsed = FactConflict(**conflict()) self.assertEqual(parsed.status, "unresolved") self.assertEqual(parsed.selected_evidence_ids, []) def test_fewer_than_two_candidates_rejected(self): with self.assertRaises(ValidationError): FactConflict(**conflict(candidate_evidence_ids=["ev-1"])) def test_identical_values_are_not_a_conflict(self): # AC-24 反例：两个来源给出同一个值不应建冲突对象 with self.assertRaises(ValidationError) as ctx: FactConflict(**conflict(candidates=[ ConflictCandidate(evidence_id="ev-1", value="1000", source="a", locator="L1"), ConflictCandidate(evidence_id="ev-2", value="1000", source="b", locator="L2"), ])) self.assertIn("不构成", str(ctx.exception)) def test_null_value_counts_as_distinct(self): # 未知用 null，且与已有数值构成分歧 parsed = FactConflict(**conflict(candidates=[ ConflictCandidate(evidence_id="ev-1", value="1000", source="a", locator="L1"), ConflictCandidate(evidence_id="ev-2", value=None, source="b", locator="L2"), ])) self.assertIsNone(parsed.candidates[1].value) def test_candidates_must_align_with_id_list(self): with self.assertRaises(ValidationError): FactConflict(**conflict(candidates=[ ConflictCandidate(evidence_id="ev-2", value="1000", source="a", locator="L1"), ConflictCandidate(evidence_id="ev-1", value="1200", source="b", locator="L2"), ])) with self.assertRaises(ValidationError): FactConflict(**conflict(candidate_evidence_ids=["ev-1", "ev-1"], candidates=[ ConflictCandidate(evidence_id="ev-1", value="1000", source="a", locator="L1"), ConflictCandidate(evidence_id="ev-1", value="1200", source="b", locator="L2"), ])) def test_resolved_requires_note_selection_and_actor(self): good = conflict(status="resolved", resolution_note="以合同为准", selected_evidence_ids=["ev-2"], resolved_by="demo-manager", resolved_at=NOW) self.assertEqual(FactConflict(**good).selected_evidence_ids, ["ev-2"]) for drop in ("resolution_note", "resolved_by", "resolved_at", "selected_evidence_ids"): with self.subTest(drop=drop): broken = dict(good) broken.pop(drop) with self.assertRaises(ValidationError): FactConflict(**broken) def test_selection_must_come_from_candidates(self): with self.assertRaises(ValidationError) as ctx: FactConflict(**conflict(status="resolved", resolution_note="n", selected_evidence_ids=["ev-999"], resolved_by="demo-manager", resolved_at=NOW)) self.assertIn("不属于", str(ctx.exception)) def test_unresolved_must_not_carry_resolution(self): with self.assertRaises(ValidationError): FactConflict(**conflict(resolution_note="提前写了结论")) def test_supersedes_chain_is_representable(self): # 再次冲突新建对象并指向旧对象，不覆写历史 parsed = FactConflict(**conflict(id="fc-2", supersedes_id="fc-1")) self.assertEqual(parsed.supersedes_id, "fc-1") class ErrorContractTests(unittest.TestCase): def test_every_code_has_an_http_status(self): self.assertEqual(set(S1_ERROR_HTTP), set(ErrorCode)) def test_status_semantics_match_existing_translate(self): # 既有 translate() 的语义：KeyError→404, Conflict→409, PermissionError→403, ValueError→422 self.assertEqual(S1_ERROR_HTTP[ErrorCode.NOT_FOUND], 404) self.assertEqual(S1_ERROR_HTTP[ErrorCode.ACCESS_DENIED], 403) self.assertEqual(S1_ERROR_HTTP[ErrorCode.REVISION_CONFLICT], 409) self.assertEqual(S1_ERROR_HTTP[ErrorCode.IDEMPOTENCY_CONFLICT], 409) self.assertEqual(S1_ERROR_HTTP[ErrorCode.IMPORT_BATCH_INVALID], 422) def test_e17_confirm_refusal_is_422(self): # E-17 有两个面：确认动作被拒 = 422（客户端须先裁定）； # 运行级结果 = HTTP 200 且冲突保存在 degradations（那个面不走错误信封）。 self.assertEqual(S1_ERROR_HTTP[ErrorCode.FACT_CONFLICT_UNRESOLVED], 422) def test_all_codes_are_non_retryable(self): # 03 §5：403/422 不自动重试；409 需客户端改载荷或刷新 self.assertFalse(any(S1_ERROR_RETRYABLE.values())) def test_envelope_shape(self): body = ErrorResponse.model_validate({"error": { "code": "IMPORT_BATCH_INVALID", "message": "材料集合不符合导入要求", "retryable": False, "request_id": "req-1"}}) self.assertEqual(body.error.code, ErrorCode.IMPORT_BATCH_INVALID) def test_no_credentials_in_contract_module(self): # 契约层不得出现凭据字段（E-10 / BR-16 的静态底线） import contracts.prescreen as module with open(module.__file__, encoding="utf-8") as handle: source = handle.read() for banned in ("api_key", "password", "token", "secret"): self.assertNotIn(banned, source.lower()) if __name__ == "__main__": unittest.main()