From d8a7d3858c42399041ba3e82a42eda8041734654 Mon Sep 17 00:00:00 2001 From: qiaoqichao Date: Tue, 29 Sep 2026 14:20:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20=E6=96=B0=E5=A2=9E=20S1=20=E5=A5=91?= =?UTF-8?q?=E7=BA=A6=E5=B1=82=EF=BC=88=E6=9D=90=E6=96=99=E9=9B=86=E5=90=88?= =?UTF-8?q?=E5=AF=BC=E5=85=A5=E4=B8=8E=E4=BA=8B=E5=AE=9E=E7=A1=AE=E8=AE=A4?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit T02 首个子卡 = S1 契约层。新增 backend/contracts/prescreen.py（设计指定的 唯一 owner），定义 SourceLocator / ImportItem / ImportBatch / FactConflict 与共用约束、错误码与错误信封。 落实的规格条目： - BR-22 逐文件保留来源/定位/解析版本与状态 - BR-23 多来源冲突保留候选并要求显式处理（resolved 需说明+选择+人） - AC-23 反例：绝对路径(C:/、/、//share)、上级引用(..)、超限、空集合、 failed 缺 error_code、parsed 缺 document_id、parsed 带 error_code - AC-24 反例：候选值全同不构成冲突；未知用 null 而非 0/空串； candidates 必须与 candidate_evidence_ids 同序同集合 - 01 §5 错误码与既有 translate() 的 HTTP 语义对齐；E-17 为运行级、HTTP 200 - 03 §9.1 时间必须带时区；禁止未知字段（E-07） D3 定案（工程限制，非业务阈值）：单文件 ≤10 MiB 沿用既有 LIMIT；单批 ≤20 文件 / ≤20 MiB。依据 03 §1 末「管理员只能降低初期限制，放宽须重新验证」， 取保守方向即规格许可；放宽需另立验证。 依赖方向：contracts 不导入 services，标准字段白名单校验留在 service 层。 测试：新增 test_prescreen_contracts.py 38 项，全绿；scripts/verify.py 9/9 PASS。 Co-Authored-By: Claude Code --- backend/contracts/prescreen.py | 309 ++++++++++++++++++++++++++ backend/test_prescreen_contracts.py | 321 ++++++++++++++++++++++++++++ 2 files changed, 630 insertions(+) create mode 100644 backend/contracts/prescreen.py create mode 100644 backend/test_prescreen_contracts.py diff --git a/backend/contracts/prescreen.py b/backend/contracts/prescreen.py new file mode 100644 index 0000000..e045685 --- /dev/null +++ b/backend/contracts/prescreen.py @@ -0,0 +1,309 @@ +"""兴邦预审 v2 · 首个实施切片（S1）契约：材料集合导入与事实确认。 + +唯一职责 owner：[docs/v2/03-detailed-design.md](../../../docs/v2/03-detailed-design.md) §1/§9.2 指定的 +`backend/contracts/prescreen.py`。本文件是该 owner 的**首版实现**，字段命名以本文件为准； +`claude_use/T00/04-first-slice-contracts.md` 是提案记录，不是第二份契约定义。 + +S1 范围（T00/04 §4/§5）： + SourceLocator · ImportItem · ImportBatch · FactConflict + 共用约束 + 错误信封 + +**不在本文件**（属后续任务包，不得在此提前定义）：CaseTimeline、DecisionEvent、 +ExperienceCard、KnowledgeCard、ReviewPlan、DimensionSpec、ScreeningRun、Judgment、 +Simulation、Evaluation、TrajectoryStep、ReviewOutcome、ImprovementCandidate、 +DatasetManifest、ProbabilityView。 + +依赖方向：本模块**不导入** `services/`。标准字段白名单（`assistant_materials.FIELDS`） +的校验属 service 层，避免 contracts → services 的反向依赖。 +""" + +from __future__ import annotations + +import re +from datetime import datetime +from enum import StrEnum +from typing import Annotated, Literal + +from pydantic import AfterValidator, BaseModel, ConfigDict, Field, model_validator + +# ---------------------------------------------------------------- 共用约束 +ID_MAX = 128 +PATH_MAX = 512 +NOTE_MAX = 2000 +EVIDENCE_MAX = 100 +DESC_MAX = 2000 + +# P0 初始工程限制（**非业务阈值**，见 docs/v2/03-detailed-design.md §1 末）。 +# 单文件上限沿用既有 assistant_materials.LIMIT = 10 MiB。 +# 单批总量取保守值：现有实现把整份文件 base64 存进 SQLite 单行， +# 设计文档曾提 100 MiB/批 ≈ 133 MiB base64 入库，可直接造成演示故障 +# （claude_use/T00/04 §3.1）。**降低**无需重新验证；**放宽**须重新验证并更新规格。 +MAX_ITEM_BYTES = 10 * 1024 * 1024 +MAX_BATCH_BYTES = 20 * 1024 * 1024 +MAX_BATCH_FILES = 20 + +_SHA256 = re.compile(r"^[0-9a-f]{64}$") +_WINDOWS_DRIVE = re.compile(r"^[A-Za-z]:") + + +def _aware(value: datetime) -> datetime: + """时间必须带时区（docs/v2/03-detailed-design.md §9.1）。""" + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError("时间必须带时区（ISO-8601）") + return value + + +AwareDatetime = Annotated[datetime, AfterValidator(_aware)] +Identifier = Annotated[str, Field(min_length=1, max_length=ID_MAX)] + + +class Contract(BaseModel): + """所有 S1 对象共用：禁止未知字段（对应 E-07 INVALID_REQUEST）。""" + + model_config = ConfigDict(extra="forbid") + + +class DataDomain(StrEnum): + demo = "demo" + production = "production" + + +class Ref(Contract): + """通用版本引用（docs/v2/03-detailed-design.md §9.1）。""" + + id: Identifier + version: str = Field(min_length=1, max_length=ID_MAX) + + +# ---------------------------------------------------------------- 错误契约 +class ErrorCode(StrEnum): + """S1 涉及的错误码。取值与 docs/v2/01-requirements.md §5 一一对应，不另造。""" + + NOT_FOUND = "NOT_FOUND" + ACCESS_DENIED = "ACCESS_DENIED" + INVALID_REQUEST = "INVALID_REQUEST" + INVALID_EVIDENCE = "INVALID_EVIDENCE" + REVISION_CONFLICT = "REVISION_CONFLICT" + IDEMPOTENCY_CONFLICT = "IDEMPOTENCY_CONFLICT" + MATERIAL_PARSE_FAILED = "MATERIAL_PARSE_FAILED" + IMPORT_BATCH_INVALID = "IMPORT_BATCH_INVALID" + FACT_CONFLICT_UNRESOLVED = "FACT_CONFLICT_UNRESOLVED" + + +# HTTP 语义沿用既有 translate()（backend/api/project_assistant.py:36-46）， +# 不另立一套。E-17 是**运行级结果标识**，不是 HTTP 错误（01 §5 末）。 +S1_ERROR_HTTP: dict[ErrorCode, int] = { + ErrorCode.NOT_FOUND: 404, + ErrorCode.ACCESS_DENIED: 403, + ErrorCode.INVALID_REQUEST: 422, + ErrorCode.INVALID_EVIDENCE: 422, + ErrorCode.REVISION_CONFLICT: 409, + ErrorCode.IDEMPOTENCY_CONFLICT: 409, + ErrorCode.MATERIAL_PARSE_FAILED: 422, + ErrorCode.IMPORT_BATCH_INVALID: 422, + ErrorCode.FACT_CONFLICT_UNRESOLVED: 200, +} + +# 403/422 不自动重试；409 需客户端改载荷或刷新。见 03 §5。 +S1_ERROR_RETRYABLE: dict[ErrorCode, bool] = {code: False for code in ErrorCode} + + +class ErrorBody(Contract): + code: ErrorCode + message: str = Field(min_length=1, max_length=DESC_MAX) + retryable: bool + request_id: str = Field(min_length=1, max_length=ID_MAX) + + +class ErrorResponse(Contract): + """新 prescreen 路由的**独立**错误信封（T00/04 §2 决策 D2）。 + + 既有路由仍是 FastAPI 默认 `{"detail": ...}`，本次不改，避免影响 + backend/api/project_assistant.py 的 translate() 及其全部消费方。 + """ + + error: ErrorBody + + +# ---------------------------------------------------------------- 来源定位 +class SourceLocator(Contract): + """证据在原件中的定位。定位**不等于**真实性核验（01 §1）。 + + kind 决定必填字段，且三种寻址方式互斥： + pdf → page + table → sheet + cell_range + manual → note + """ + + kind: Literal["pdf", "table", "manual"] + document_id: Identifier + document_version: int = Field(ge=1) + page: int | None = Field(default=None, ge=1) + sheet: str | None = Field(default=None, min_length=1, max_length=200) + cell_range: str | None = Field(default=None, min_length=1, max_length=200) + note: str | None = Field(default=None, min_length=1, max_length=PATH_MAX) + excerpt: str | None = Field(default=None, max_length=4000) + + _REQUIRED: dict[str, tuple[str, ...]] = { + "pdf": ("page",), + "table": ("sheet", "cell_range"), + "manual": ("note",), + } + _ADDRESS_FIELDS = ("page", "sheet", "cell_range", "note") + + @model_validator(mode="after") + def _check_kind_fields(self) -> SourceLocator: + required = self._REQUIRED[self.kind] + missing = [name for name in required if getattr(self, name) is None] + if missing: + raise ValueError(f"{self.kind} 定位缺少必填字段：{', '.join(missing)}") + extra = [name for name in self._ADDRESS_FIELDS + if name not in required and getattr(self, name) is not None] + if extra: + raise ValueError(f"{self.kind} 定位不允许字段：{', '.join(extra)}") + return self + + +# ---------------------------------------------------------------- 导入清单 +class ImportItem(Contract): + """集合中的一个文件条目。`relative_path` 仅表达用户选择的文件集合， + 绝不作服务器写入路径（docs/v2/14-experience-loop.md §3）。 + """ + + id: Identifier + document_id: Identifier | None = None # 解析成功后由既有 upload 链回填 + relative_path: str = Field(min_length=1, max_length=PATH_MAX) + sha256: str = Field(pattern=_SHA256.pattern) + size_bytes: int = Field(ge=0, le=MAX_ITEM_BYTES) + mime_type: str = Field(min_length=1, max_length=200) + status: Literal["queued", "parsing", "parsed", "failed"] + error_code: ErrorCode | None = None + observed_at: AwareDatetime | None = None # 材料**发生**时间；未知为 null + received_at: AwareDatetime + + @model_validator(mode="after") + def _check_path_and_status(self) -> ImportItem: + # 路径安全：不允许绝对路径、盘符或上级引用。校验用规范化副本， + # 但**保留原值**，使客户端提交的内容与落库内容一致。 + normalized = self.relative_path.replace("\\", "/") + if normalized.startswith("/") or _WINDOWS_DRIVE.match(normalized): + raise ValueError("relative_path 不得为绝对路径") + if any(part == ".." for part in normalized.split("/")): + raise ValueError("relative_path 不得包含上级引用") + + if self.status == "failed" and self.error_code is None: + raise ValueError("failed 的条目必须有 error_code") + if self.status == "parsed" and self.error_code is not None: + raise ValueError("parsed 的条目不得带 error_code") + if self.status == "parsed" and self.document_id is None: + raise ValueError("parsed 的条目必须已关联 document_id") + return self + + +class ImportBatch(Contract): + """一次「文件集合」导入的容器对象。 + + 状态机（T00/04 §6.1）： + queued → parsing → ready | partial | failed + queued | parsing → cancelled + partial | failed → queued（retry，仅重置失败项，revision++） + """ + + id: Identifier + project_id: Identifier + data_domain: DataDomain + revision: int = Field(ge=1) + status: Literal["queued", "parsing", "ready", "partial", "failed", "cancelled"] + created_at: AwareDatetime + created_by: Identifier + parser_version: str = Field(min_length=1, max_length=64) + items: list[ImportItem] = Field(min_length=1, max_length=MAX_BATCH_FILES) + # 不在此处加 le=MAX_BATCH_BYTES：字段级约束会先于模型级校验触发， + # 使用户看到 pydantic 的英文默认文案而非下面的中文提示。 + total_bytes: int = Field(ge=0) + warnings: list[str] = Field(default_factory=list, max_length=EVIDENCE_MAX) + + @model_validator(mode="after") + def _check_consistency(self) -> ImportBatch: + if len({item.id for item in self.items}) != len(self.items): + raise ValueError("同一批次内 item id 必须唯一") + # 单项上限已由 ImportItem.size_bytes 的 le 约束保证，此处不重复检查。 + total = sum(item.size_bytes for item in self.items) + if total != self.total_bytes: + raise ValueError("total_bytes 必须等于各 item size_bytes 之和") + if total > MAX_BATCH_BYTES: + raise ValueError( + f"单批总量 {total} 超过上限 {MAX_BATCH_BYTES} 字节；请拆分后重试" + ) + return self + + +# ---------------------------------------------------------------- 事实冲突 +class ConflictCandidate(Contract): + """并列展示的候选值之一。未知用 null，**不用 0 或空串代替**（03 §9.1）。""" + + evidence_id: Identifier + value: str | None = None + source: str = Field(min_length=1, max_length=PATH_MAX) + locator: str = Field(min_length=1, max_length=PATH_MAX) + unit: str | None = Field(default=None, max_length=64) + currency: str | None = Field(default=None, max_length=16) + period: str | None = Field(default=None, max_length=64) + + +class FactConflict(Contract): + """同一字段在多来源上给出不同值时的待处理对象（BR-23 / E-17 / AC-24）。 + + 只允许 unresolved → resolved。再次冲突时**新建**对象并用 supersedes_id + 指向旧对象，不覆写历史。 + """ + + id: Identifier + project_id: Identifier + data_domain: DataDomain + revision: int = Field(ge=1) + # 标准字段名（assistant_materials.FIELDS.values()）。白名单校验在 service 层， + # 见本模块顶部「依赖方向」说明。 + field: str = Field(min_length=1, max_length=ID_MAX) + candidate_evidence_ids: list[Identifier] = Field(min_length=2, max_length=EVIDENCE_MAX) + candidates: list[ConflictCandidate] = Field(min_length=2, max_length=EVIDENCE_MAX) + status: Literal["unresolved", "resolved"] + resolution_note: str | None = Field(default=None, min_length=1, max_length=NOTE_MAX) + selected_evidence_ids: list[Identifier] = Field(default_factory=list, max_length=EVIDENCE_MAX) + resolved_by: Identifier | None = None + resolved_at: AwareDatetime | None = None + created_at: AwareDatetime + supersedes_id: Identifier | None = None + + @model_validator(mode="after") + def _check_candidates_and_resolution(self) -> FactConflict: + ids = self.candidate_evidence_ids + if len(set(ids)) != len(ids): + raise ValueError("candidate_evidence_ids 不得重复") + if [c.evidence_id for c in self.candidates] != ids: + raise ValueError("candidates 必须与 candidate_evidence_ids 同序同集合") + # 「至少两个不同值」：全部候选值相同说明没有分歧，不应建冲突对象。 + if len({c.value for c in self.candidates}) < 2: + raise ValueError("候选值全部相同，不构成事实冲突") + + if self.status == "resolved": + missing = [name for name, value in ( + ("resolution_note", self.resolution_note), + ("resolved_by", self.resolved_by), + ("resolved_at", self.resolved_at), + ) if value is None] + if missing: + raise ValueError(f"resolved 必须提供：{', '.join(missing)}") + if not self.selected_evidence_ids: + raise ValueError("resolved 必须提供 selected_evidence_ids") + unknown = set(self.selected_evidence_ids) - set(ids) + if unknown: + raise ValueError(f"选择的值不属于该冲突的候选：{sorted(unknown)}") + else: + stray = [name for name, value in ( + ("resolution_note", self.resolution_note), + ("resolved_by", self.resolved_by), + ("resolved_at", self.resolved_at), + ) if value is not None] + if stray or self.selected_evidence_ids: + raise ValueError("unresolved 不得带解决字段") + return self diff --git a/backend/test_prescreen_contracts.py b/backend/test_prescreen_contracts.py new file mode 100644 index 0000000..f4aa919 --- /dev/null +++ b/backend/test_prescreen_contracts.py @@ -0,0 +1,321 @@ +"""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, + 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_is_not_an_http_error(self): + # 01 §5 末：运行级错误 HTTP 200，只表示结果可读 + self.assertEqual(S1_ERROR_HTTP[ErrorCode.FACT_CONFLICT_UNRESOLVED], 200) + + 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()