qiaoqichaoandClaude Code
669e969c41
T03 卡的要求是"复用上传/确认入口，增加导入清单、定位和冲突候选"。核对后： 导入清单与冲突候选已由 T02/S1 交付，但**"定位"只定义了契约、没有接上**—— parse() 仍只产出自由字符串 locator，SourceLocator 类型化投影是空的。 本次补上这条链路（契约与实现不一致是我的遗留）。 - assistant_materials.parse()：每条 chunk 增加 source_locator（kind + 坐标）， 按 T00/04 §5.3 的映射：pdf→page、xlsx 单元格/行与 csv/txt/docx 表格→sheet+cell_range、 docx 正文段→manual(note) - material() 补 document_id/document_version 后**按契约校验**；漏了定位或定位不合法 直接抛错，不落半成品进库 - **locator 自由字符串保留**：enqueue() 依赖它算证据 ID 的稳定哈希，改它会破坏 BR-13 实现中修掉一个自己引入的 bug：类型化定位的辅助函数初版命名为 table，与 docx 分支里 的表格计数器 table 同名，该分支内 helper 被整数遮蔽（'int' object is not callable）。 已改名 sheet_loc。 AC-08 补测：跨项目候选引用被 supplement 拒（"不属于项目"），同项目可用作对照， 未知 candidate_id 同样被拒。既有 test_project_assistant 只覆盖了跨项目文档下载。 测试：新增 test_sourcelocator.py 13 项，含真实 XLSX/DOCX fixture（zipfile 构造）， 覆盖五种格式的定位派生与契约校验。全量 248 项 OK（235 → 248）；verify.py 10/10 PASS。 Co-Authored-By: Claude Code <noreply@anthropic.com>
211 lines
10 KiB
Python
211 lines
10 KiB
Python
"""T03 类型化来源定位测试：parse()/material() 必须产出契约合法的 SourceLocator。
|
||
|
||
背景：T02 定义了 `contracts.prescreen.SourceLocator`，但当时**没有接上**
|
||
（parse() 只产出自由字符串 locator）。本文件锁定这条链路：
|
||
parse() 产出 kind + 坐标 → material() 补 document_id/document_version → 契约校验
|
||
|
||
同时锁定 `locator` 自由字符串**必须保留**：enqueue() 用它算证据 ID 的稳定哈希，
|
||
改动它会破坏 BR-13 的版本冻结。
|
||
"""
|
||
|
||
import io
|
||
import unittest
|
||
import zipfile
|
||
from pathlib import Path
|
||
|
||
from contracts.prescreen import SourceLocator
|
||
from services.assistant_materials import material, parse
|
||
|
||
|
||
def xlsx_bytes(rows=(("field", "value"), ("facility.amount", "100"))):
|
||
"""构造最小可解析 XLSX（inlineStr，无需 sharedStrings）。"""
|
||
ns = 'xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main"'
|
||
rel = 'xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships"'
|
||
sheet_rows = "".join(
|
||
"<row r=\"%d\">%s</row>" % (
|
||
index,
|
||
"".join(
|
||
f'<c r="{chr(65 + col)}{index}" t="inlineStr"><is><t>{cell}</t></is></c>'
|
||
for col, cell in enumerate(row)
|
||
),
|
||
)
|
||
for index, row in enumerate(rows, 1)
|
||
)
|
||
parts = {
|
||
"xl/workbook.xml": f'<workbook {ns}><sheets><sheet name="Sheet1" sheetId="1" r:id="rId1" {rel}/></sheets></workbook>',
|
||
"xl/_rels/workbook.xml.rels": '<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
|
||
'<Relationship Id="rId1" Target="worksheets/sheet1.xml"/></Relationships>',
|
||
"xl/worksheets/sheet1.xml": f'<worksheet {ns}><sheetData>{sheet_rows}</sheetData></worksheet>',
|
||
}
|
||
buffer = io.BytesIO()
|
||
with zipfile.ZipFile(buffer, "w") as archive:
|
||
for name, body in parts.items():
|
||
archive.writestr(name, body)
|
||
return buffer.getvalue()
|
||
|
||
|
||
def docx_bytes():
|
||
"""构造最小可解析 DOCX：一段正文 + 一张表。"""
|
||
ns = 'xmlns:w="http://schemas.openxmlformats.org/wordprocessingml/2006/main"'
|
||
body = (
|
||
f'<w:p><w:r><w:t>正文第一段</w:t></w:r></w:p>'
|
||
f'<w:tbl><w:tr><w:tc><w:p><w:r><w:t>甲</w:t></w:r></w:p></w:tc>'
|
||
f'<w:tc><w:p><w:r><w:t>乙</w:t></w:r></w:p></w:tc></w:tr></w:tbl>'
|
||
)
|
||
buffer = io.BytesIO()
|
||
with zipfile.ZipFile(buffer, "w") as archive:
|
||
archive.writestr("word/document.xml", f"<w:document {ns}><w:body>{body}</w:body></w:document>")
|
||
return buffer.getvalue()
|
||
|
||
|
||
class TypedLocatorTests(unittest.TestCase):
|
||
def locators(self, content, filename):
|
||
document = material("DEMO-A", filename, content, "tester")
|
||
return document, [chunk["source_locator"] for chunk in document["chunks"]]
|
||
|
||
def test_csv_rows_become_table_locators(self):
|
||
content = "field,value\nfacility.amount,100\n".encode("utf-8-sig")
|
||
document, locators = self.locators(content, "流水.csv")
|
||
first = locators[0]
|
||
self.assertEqual(first["kind"], "table")
|
||
self.assertEqual(first["sheet"], "流水") # 文件名去扩展名
|
||
self.assertEqual(first["cell_range"], "第1行")
|
||
self.assertEqual(first["document_id"], document["id"])
|
||
self.assertEqual(first["document_version"], 1)
|
||
|
||
def test_txt_rows_become_table_locators(self):
|
||
content = "第一行\n第二行\n".encode("utf-8-sig")
|
||
_, locators = self.locators(content, "说明.txt")
|
||
self.assertEqual(locators[1]["cell_range"], "第2行")
|
||
self.assertEqual(locators[1]["kind"], "table")
|
||
|
||
def test_xlsx_cells_and_rows_become_table_locators(self):
|
||
_, locators = self.locators(xlsx_bytes(), "台账.xlsx")
|
||
kinds = {item["kind"] for item in locators}
|
||
self.assertEqual(kinds, {"table"})
|
||
# 单元格级定位：A1 必须是 sheet + cell_range，而不是行号
|
||
cell = next(item for item in locators if item["cell_range"] == "A1")
|
||
self.assertEqual(cell["sheet"], "Sheet1")
|
||
# 行级定位也保留
|
||
self.assertTrue(any(item["cell_range"].endswith("行") for item in locators))
|
||
|
||
def test_docx_paragraph_is_manual_and_table_is_table(self):
|
||
_, locators = self.locators(docx_bytes(), "说明.docx")
|
||
by_kind = {}
|
||
for item in locators:
|
||
by_kind.setdefault(item["kind"], []).append(item)
|
||
self.assertIn("manual", by_kind, "DOCX 正文段应归 manual（三值枚举里没有段落子类型）")
|
||
self.assertIn("table", by_kind, "DOCX 表格应归 table")
|
||
self.assertEqual(by_kind["manual"][0]["note"], "正文第1段")
|
||
self.assertEqual(by_kind["table"][0]["sheet"], "表1")
|
||
|
||
def test_every_locator_validates_against_the_contract(self):
|
||
# material() 已按契约校验；这里再独立复核一遍，防止校验被绕过
|
||
for content, name in [(xlsx_bytes(), "台账.xlsx"), (docx_bytes(), "说明.docx"),
|
||
("a,b\n".encode("utf-8-sig"), "t.csv")]:
|
||
with self.subTest(name=name):
|
||
_, locators = self.locators(content, name)
|
||
self.assertTrue(locators)
|
||
for item in locators:
|
||
SourceLocator(**item) # 不合法会抛 ValidationError
|
||
|
||
def test_free_string_locator_is_preserved(self):
|
||
"""locator 必须保留：enqueue() 依赖它计算证据 ID 的稳定哈希（BR-13）。"""
|
||
content = "field,value\nfacility.amount,100\n".encode("utf-8-sig")
|
||
document = material("DEMO-A", "流水.csv", content, "tester")
|
||
strings = [chunk["locator"] for chunk in document["chunks"]]
|
||
self.assertIn("第1行", strings)
|
||
self.assertIn("第2行", strings)
|
||
# 每个 chunk 同时有两种形态
|
||
for chunk in document["chunks"]:
|
||
self.assertIn("locator", chunk)
|
||
self.assertIn("source_locator", chunk)
|
||
self.assertIsInstance(chunk["locator"], str)
|
||
|
||
def test_parse_without_material_fillup_has_no_document_id(self):
|
||
"""parse() 本身不知道 document_id——这正是补齐放在 material() 的原因。"""
|
||
chunks, _ = parse("a\n".encode("utf-8-sig"), "t.txt")
|
||
raw = chunks[0]["source_locator"]
|
||
self.assertNotIn("document_id", raw)
|
||
self.assertEqual(raw["kind"], "table")
|
||
|
||
def test_document_version_is_recorded_not_assumed(self):
|
||
content = "a,b\n".encode("utf-8-sig")
|
||
document = material("DEMO-A", "v.csv", content, "tester", version=7)
|
||
self.assertEqual(document["version"], 7)
|
||
self.assertEqual(document["chunks"][0]["source_locator"]["document_version"], 7)
|
||
|
||
def test_locator_cannot_cross_documents(self):
|
||
"""两个文档的定位互不串用：document_id 是各自的。"""
|
||
a = material("DEMO-A", "a.csv", "x,y\n".encode("utf-8-sig"), "tester")
|
||
b = material("DEMO-A", "b.csv", "x,y\n".encode("utf-8-sig"), "tester")
|
||
self.assertNotEqual(a["id"], b["id"])
|
||
self.assertEqual(a["chunks"][0]["source_locator"]["document_id"], a["id"])
|
||
self.assertEqual(b["chunks"][0]["source_locator"]["document_id"], b["id"])
|
||
# 文档 a 的定位不可能出现在文档 b 里
|
||
self.assertNotIn(a["id"], [c["source_locator"]["document_id"] for c in b["chunks"]])
|
||
|
||
def test_missing_typed_locator_is_rejected(self):
|
||
"""防御性断言：若某条解析路径漏了 source_locator，material() 必须失败而不是静默产半成品。"""
|
||
import services.assistant_materials as materials
|
||
from unittest.mock import patch
|
||
original = materials.parse
|
||
|
||
def stripped(content, filename):
|
||
chunks, warnings = original(content, filename)
|
||
for chunk in chunks:
|
||
chunk.pop("source_locator", None)
|
||
return chunks, warnings
|
||
|
||
with patch.object(materials, "parse", stripped):
|
||
with self.assertRaises(ValueError) as ctx:
|
||
materials.material("DEMO-A", "t.csv", "a,b\n".encode("utf-8-sig"), "tester")
|
||
self.assertIn("类型化定位", str(ctx.exception))
|
||
|
||
|
||
class ReferenceValidationTests(unittest.TestCase):
|
||
"""AC-08：无效或跨项目引用必须被拒，且不暴露无权原文。"""
|
||
|
||
def setUp(self):
|
||
import tempfile
|
||
from services.project_assistant import ProjectAssistantService
|
||
self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
|
||
self.service = ProjectAssistantService(Path(self.temp.name) / "assistant.sqlite3")
|
||
|
||
def tearDown(self):
|
||
self.temp.cleanup()
|
||
|
||
def test_candidate_from_another_project_is_rejected(self):
|
||
payload = "field,value,currency,unit,period,scope\nfacility.amount,100,CNY,万元,2025,合并\n".encode("utf-8-sig")
|
||
document = self.service.upload("DEMO-A", "流水.csv", payload, "demo-manager")
|
||
self.assertTrue(document["candidates"], "解析应产出候选")
|
||
candidate_id = document["candidates"][0]["id"]
|
||
|
||
with self.assertRaises(ValueError) as ctx:
|
||
self.service.supplement("DEMO-B", "demo-manager", {}, [],
|
||
self.service.get_project("DEMO-B")["revision"],
|
||
"跨项目引用测试", candidate_ids=[candidate_id])
|
||
self.assertIn("不属于项目", str(ctx.exception))
|
||
|
||
def test_candidate_reference_is_project_scoped(self):
|
||
"""同项目可用、跨项目不可用——证明拒绝来自项目范围而不是候选不存在。"""
|
||
payload = "field,value,currency,unit,period,scope\nfacility.amount,100,CNY,万元,2025,合并\n".encode("utf-8-sig")
|
||
document = self.service.upload("DEMO-A", "流水.csv", payload, "demo-manager")
|
||
candidate_id = document["candidates"][0]["id"]
|
||
|
||
project = self.service.get_project("DEMO-A")
|
||
result = self.service.supplement("DEMO-A", "demo-manager", {}, [], project["revision"],
|
||
"同项目确认来源", candidate_ids=[candidate_id])
|
||
self.assertEqual(result["project_id"], "DEMO-A")
|
||
self.assertIn("facility.amount", result["changed_fields"])
|
||
|
||
def test_unknown_candidate_id_is_also_rejected(self):
|
||
project = self.service.get_project("DEMO-A")
|
||
with self.assertRaises(ValueError):
|
||
self.service.supplement("DEMO-A", "demo-manager", {}, [], project["revision"],
|
||
"伪造引用", candidate_ids=["candidate-does-not-exist"])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|