Files
2026Technology-Competition/tests/test_phase5_writer_orchestrator.py
T

111 lines
4.0 KiB
Python

import pytest
from types import SimpleNamespace
from docx import Document
from genesis.data_models import ParsedTemplate, ChapterMarker
from genesis.writer.orchestrator import WriteOrchestrator
from genesis.writer.models import ChapterContent
class FakeEngine:
def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2):
# 返回一个固定章节内容(title + 一个段落块)
return SimpleNamespace(
data={"title": variables["title"], "blocks": [{"type": "paragraph", "text": "自动生成的内容"}]},
status="ok",
)
def _make_template(path):
doc = Document()
doc.add_paragraph("はじめに", style="Heading 1")
doc.add_paragraph("{{section:introduction}}")
doc.save(path)
def _ss(template_path):
parsed = ParsedTemplate(
file_name=template_path,
sections=[
ChapterMarker(type="heading", name="はじめに", level=1),
ChapterMarker(type="placeholder", name="section:introduction", level=0),
],
placeholders={},
styles={"defined": ["Heading 1"], "used": ["Heading 1"]},
)
return SimpleNamespace(template=parsed)
def test_generate_produces_filled_docx(tmp_path):
tpl = tmp_path / "tpl.docx"
out = tmp_path / "out.docx"
_make_template(str(tpl))
orch = WriteOrchestrator()
contents = orch.generate(_ss(str(tpl)), str(out), samples_dir="nonexistent_dir_xyz", engine=FakeEngine())
assert isinstance(contents, list) and len(contents) == 1
# 输出 docx 含注入文本且无残留异常
loaded = Document(str(out))
joined = "\n".join(p.text for p in loaded.paragraphs)
assert "自动生成的内容" in joined
class AsyncFakeEngine:
async def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2):
return SimpleNamespace(
data={"title": variables["title"], "blocks": [{"type": "paragraph", "text": "异步引擎内容"}]},
status="ok",
)
def test_generate_with_async_engine_produces_filled_docx(tmp_path):
tpl = tmp_path / "tpl.docx"
out = tmp_path / "out.docx"
_make_template(str(tpl))
orch = WriteOrchestrator()
contents = orch.generate(
_ss(str(tpl)), str(out), samples_dir="nonexistent_dir_xyz", engine=AsyncFakeEngine()
)
assert isinstance(contents, list) and len(contents) == 1
loaded = Document(str(out))
joined = "\n".join(p.text for p in loaded.paragraphs)
assert "异步引擎内容" in joined
def test_generate_delegates_to_factory_when_engine_none(monkeypatch, tmp_path):
# 门禁真实路径:engine=None 时应委托 build_inference_engine() 构造真实引擎
monkeypatch.setattr("genesis.writer.orchestrator.build_inference_engine", lambda: FakeEngine())
tpl = tmp_path / "tpl.docx"
out = tmp_path / "out.docx"
_make_template(str(tpl))
contents = WriteOrchestrator().generate(
_ss(str(tpl)), str(out), samples_dir="nonexistent_dir_xyz", engine=None
)
assert len(contents) == 1
loaded = Document(str(out))
assert "自动生成的内容" in "\n".join(p.text for p in loaded.paragraphs)
def test_generate_warns_on_unanchored_heading(tmp_path, caplog):
# 防静默丢章:无 {{section:id}} 锚点的章 → 打 WARNING 并列出章名
tpl = tmp_path / "tpl.docx"
out = tmp_path / "out.docx"
doc = Document()
doc.add_paragraph("附録", style="Heading 1")
doc.save(str(tpl))
parsed = ParsedTemplate(
file_name=str(tpl),
sections=[ChapterMarker(type="heading", name="附録", level=1)],
placeholders={},
styles={"defined": ["Heading 1"], "used": ["Heading 1"]},
)
orch = WriteOrchestrator()
with caplog.at_level("WARNING", logger="genesis.writer.orchestrator"):
orch.generate(
SimpleNamespace(template=parsed),
str(out),
samples_dir="nonexistent_dir_xyz",
engine=FakeEngine(),
template_path=str(tpl),
)
assert any("附録" in r.message for r in caplog.records)