"""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( "%s" % ( index, "".join( f'{cell}' for col, cell in enumerate(row) ), ) for index, row in enumerate(rows, 1) ) parts = { "xl/workbook.xml": f'', "xl/_rels/workbook.xml.rels": '' '', "xl/worksheets/sheet1.xml": f'{sheet_rows}', } 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'正文第一段' f'甲' f'乙' ) buffer = io.BytesIO() with zipfile.ZipFile(buffer, "w") as archive: archive.writestr("word/document.xml", f"{body}") 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()