wechat-robot-skills/skills/docx/tests/test_document_workflows.py

231 lines
11 KiB
Python

"""Regression checks for the merged document workflows; no Docker or network required."""
from __future__ import annotations
import importlib
import json
import shutil
import sys
import tempfile
import unittest
from copy import deepcopy
from pathlib import Path
from unittest.mock import patch
sys.dont_write_bytecode = True
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
from docx import Document
from docx.oxml import OxmlElement
from docx.oxml.ns import qn
from reportlab.pdfgen.canvas import Canvas
import _docx_common as common
import compile_typst
class DocumentWorkflows(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory(prefix="docx-merge-test-")
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name).resolve()
self.addCleanup(patch.stopall)
patch.object(common, "WORD_OUTPUT_ROOT", self.root).start()
patch.object(compile_typst, "WORD_OUTPUT_ROOT", self.root).start()
self.source = self.root / "source.docx"
def call(self, script, *args):
with patch.object(sys, "argv", [script, *map(str, args)]):
return importlib.import_module(script).main()
def fixture(self):
doc = Document()
p = doc.add_paragraph()
p.add_run("BEFORE ")
p.add_run("HEL").bold = True
p.add_run("LO")
p.add_run(" AFTER").italic = True
doc.save(self.source)
return doc
def edit(self, operations, *args):
output = self.root / "edited.docx"
result = self.call("edit_document", "--input", self.source, "--output", output,
"--spec", json.dumps({"operations": operations}), *args)
return result, Document(output)
def test_cross_run_preserves_unmodified_styles_and_escapes_text(self):
self.fixture()
result, doc = self.edit([{"type": "replace_text", "find": "HELLO", "replace": "A & <B>"}])
p = doc.paragraphs[0]
self.assertEqual(p.text, "BEFORE A & <B> AFTER")
self.assertTrue(next(r for r in p.runs if "A &" in r.text).bold)
self.assertTrue(next(r for r in p.runs if " AFTER" in r.text).italic)
validation = self.call("validate_document", "--input", result["path"])
self.assertEqual(validation["status"], "valid")
def test_single_and_split_matches_are_both_replaced(self):
doc = self.fixture()
doc.add_paragraph("HELLO HELLO")
doc.save(self.source)
result, doc = self.edit([{"type": "replace_text", "find": "HELLO", "replace": "NEW"}])
self.assertEqual(result["operation_results"][0]["replacement_count"], 3)
self.assertNotIn("HELLO", "".join(p.text for p in doc.paragraphs))
def test_tracked_replacement_has_correct_nesting_and_accepted_text(self):
self.fixture()
result, doc = self.edit([{"type": "replace_text", "find": "HELLO", "replace": "A & <B>"}],
"--track-changes", "--author", "审阅 & A")
self.assertEqual(result["tracked_replacement_count"], 1)
self.assertFalse(doc.element.xpath(".//w:r/w:ins | .//w:r/w:del"))
self.assertEqual(len(doc.element.xpath(".//w:p/w:ins")), 1)
self.assertEqual(len(doc.element.xpath(".//w:p/w:del")), 1)
for node in doc.element.xpath(".//w:ins | .//w:del"):
self.assertEqual(node.get(qn("w:author")), "审阅 & A")
accepted = deepcopy(doc.element)
for node in accepted.xpath(".//w:del"):
node.getparent().remove(node)
for node in accepted.xpath(".//w:ins"):
parent = node.getparent()
index = parent.index(node)
for child in list(node):
parent.insert(index, child)
index += 1
parent.remove(node)
self.assertEqual("".join(n.text or "" for n in accepted.xpath(".//w:t")), "BEFORE A & <B> AFTER")
self.assertTrue(next(r for r in doc.paragraphs[0].runs if " AFTER" in r.text).italic)
self.assertEqual(self.call("validate_document", "--input", result["path"])["status"], "valid")
def test_multiple_tracked_matches_in_one_run(self):
doc = Document()
doc.add_paragraph("old old old")
doc.save(self.source)
result, doc = self.edit([{"type": "replace_text", "find": "old", "replace": "new"}], "--track-changes")
self.assertEqual(result["tracked_replacement_count"], 3)
self.assertEqual(len(doc.element.xpath(".//w:p/w:ins")), 3)
ids = [node.get(qn("w:id")) for node in doc.element.xpath(".//w:ins | .//w:del")]
self.assertEqual(len(ids), len(set(ids)))
def test_tracked_edits_reject_structural_operations(self):
self.fixture()
with self.assertRaisesRegex(ValueError, "仅支持 replace_text"):
self.edit([{"type": "append_blocks", "blocks": [{"type": "paragraph", "text": "x"}]}], "--track-changes")
self.assertFalse((self.root / "edited.docx").exists())
def test_tracked_edits_reject_bookmark_crossing(self):
doc = self.fixture()
mark = OxmlElement("w:bookmarkStart")
mark.set(qn("w:id"), "7")
mark.set(qn("w:name"), "target")
doc.paragraphs[0]._p.insert(2, mark)
doc.save(self.source)
with self.assertRaisesRegex(ValueError, "书签"):
self.edit([{"type": "replace_text", "find": "HELLO", "replace": "new"}], "--track-changes")
self.assertFalse((self.root / "edited.docx").exists())
def test_patent_preset_has_independent_headers_and_page_number_restarts(self):
output = self.root / "patent.docx"
spec = {"claims": [{"number": 1, "text": "一种方法 & 装置", "dependent": False}],
"specification": {"field": "领域", "detailed": ["实现 <描述>"]}, "abstract": "摘要"}
result = self.call("create_document", "--preset", "patent", "--output", output, "--spec", json.dumps(spec))
doc = Document(output)
self.assertEqual(len(doc.sections), 3)
self.assertEqual([s.header.paragraphs[0].text for s in doc.sections], ["权利要求书", "说明书", "摘要"])
for section in doc.sections:
self.assertEqual(section._sectPr.find(qn("w:pgNumType")).get(qn("w:start")), "1")
self.assertFalse(section.header.is_linked_to_previous)
self.assertFalse(section.footer.is_linked_to_previous)
self.assertTrue(any(p.style.name == "Heading 2" for p in doc.paragraphs))
self.assertEqual(self.call("validate_document", "--input", result["path"])["status"], "valid")
def test_invalid_patent_numbering_does_not_publish_a_file(self):
output = self.root / "patent.docx"
with self.assertRaisesRegex(ValueError, "连续编号"):
self.call("create_document", "--preset", "patent", "--output", output,
"--spec", json.dumps({"claims": [{"number": 2, "text": "wrong"}]}))
self.assertFalse(output.exists())
def test_generic_document_api_still_supports_tables_and_sections(self):
output = self.root / "generic.docx"
spec = {"header": {"text": "Default"}, "blocks": [
{"type": "heading", "text": "Report"},
{"type": "table", "rows": [["Key", "Value"], ["A", "B"]]},
{"type": "section_break"}, {"type": "paragraph", "text": "Second"}],
"sections": [{"index": 1, "header": {"text": "Second header"}}]}
self.call("create_document", "--output", output, "--spec", json.dumps(spec))
doc = Document(output)
self.assertEqual(doc.tables[0].cell(1, 1).text, "B")
self.assertEqual([s.header.paragraphs[0].text for s in doc.sections], ["Default", "Second header"])
def test_output_root_is_enforced(self):
with self.assertRaisesRegex(ValueError, "必须输出到"):
common.output_file(str(self.root.parent / "outside.docx"))
def test_html_extraction_decodes_entities_and_omits_scripts(self):
path = self.root / "input.html"
path.write_text('<p>A &amp; B</p><script>secret()</script><style>body{}</style><p>正文</p>')
result = self.call("extract_source", "--input", path)
self.assertEqual(result["text"], "A & B\n正文")
self.assertTrue(result["usable_for_summary"])
def test_text_cursor_preserves_every_character(self):
path = self.root / "long.txt"
text = "原始文本ABC & <> " * 90
path.write_text(text)
offset = 0
pieces = []
while True:
result = self.call("extract_source", "--input", path, "--max-chars", 256, "--start-offset", offset)
pieces.append(result["text"])
if not result["has_more"]:
break
self.assertGreater(result["next_offset"], offset)
offset = result["next_offset"]
self.assertEqual("".join(pieces), text)
def test_pdf_two_column_order_and_page_cursor(self):
path = self.root / "columns.pdf"
canvas = Canvas(str(path), pagesize=(600, 800))
for y, left, right in [(730, "LEFT FIRST", "RIGHT FIRST"), (710, "LEFT SECOND", "RIGHT SECOND")]:
canvas.drawString(50, y, left)
canvas.drawString(350, y, right)
canvas.showPage()
canvas.drawString(50, 730, "NEXT PAGE")
canvas.save()
result = self.call("extract_source", "--input", path)
self.assertEqual(result["columns"], 2)
self.assertLess(result["text"].index("LEFT SECOND"), result["text"].index("RIGHT FIRST"))
self.assertEqual(result["next_page"], 2)
result = self.call("extract_source", "--input", path, "--page", 2)
self.assertIn("NEXT PAGE", result["text"])
self.assertFalse(result["has_more"])
def test_empty_pdf_is_not_reported_as_reliable_text(self):
path = self.root / "scan.pdf"
canvas = Canvas(str(path))
canvas.rect(50, 50, 50, 50, fill=1)
canvas.showPage()
canvas.save()
result = self.call("extract_source", "--input", path)
self.assertFalse(result["usable_for_summary"])
self.assertTrue(result["needs_ocr"])
def test_typst_template_keeps_user_content_as_json(self):
text = '#read("/etc/passwd") & [not markup]'
spec = {"name": "测试", "sections": [{"title": "经历", "items": [text]}]}
source = compile_typst.resume_source(spec, self.root)
self.assertNotIn(text, source.read_text())
self.assertEqual(json.loads((self.root / "resume.json").read_text())["sections"][0]["items"][0], text)
@unittest.skipUnless(shutil.which("typst"), "Typst CLI is not installed in this local test runtime")
def test_typst_template_compiles_to_pdf(self):
spec = {"name": "示例", "contact": ["example@example.com"],
"sections": [{"title": "经历", "items": ["测试内容 & <text>"]}]}
output = self.root / "resume.pdf"
result = self.call("compile_typst", "--template", "resume", "--output", output, "--spec", json.dumps(spec))
self.assertGreater(result["page_count"], 0)
if __name__ == "__main__":
unittest.main(verbosity=2)