import pytest from types import SimpleNamespace from docx import Document from genesis.data_models import ( ControllerInfo, EntityInfo, ExistingSystemInfo, ImpactReport, ParsedTemplate, ChapterMarker, ServiceInfo, ) from genesis.writer.orchestrator import WriteOrchestrator from genesis.writer.models import ChapterContent class FakeEngine: def __init__(self): self.last_variables = None def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2): self.last_variables = variables # 返回一个固定章节内容（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) # ---------- Impact Agent MVP：影响调查结果作为生成主上下文（2026-08-23） ---------- def _ss_with_existing(template_path): from genesis.data_models import CellValue, ExcelTable, Provenance, SheetType 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"]}, ) existing = ExistingSystemInfo( controller_layer=[ControllerInfo("OrderController", "OrderController", "OrderController.java", "/trade/order", [], "trade-order/OrderController.java")], service_layer=[], entity_layer=[EntityInfo("OrderDO", "OrderDO", "OrderDO.java", "trade_order", ["id"], "trade-order/OrderDO.java")], api_endpoints=[], source_path="samples/existing-system", ) headers = ["機能ID", "機能名", "変更区分", "既存対応"] rows = [{ "機能ID": CellValue("F002", Provenance("f.xlsx", "機能一覧", 3, "A", "機能ID")), "機能名": CellValue("订单状态查询扩展", Provenance("f.xlsx", "機能一覧", 3, "B", "機能名")), "変更区分": CellValue("変更", Provenance("f.xlsx", "機能一覧", 3, "G", "変更区分")), "既存対応": CellValue("OrderController", Provenance("f.xlsx", "機能一覧", 3, "H", "既存対応")), }] tables = [ExcelTable(name="機能一覧", detected_type=SheetType.FUNCTION, extraction_method="openpyxl", headers=headers, rows=rows)] return SimpleNamespace(template=parsed, existing_system=existing, tables=tables, impact_report=None) def test_generate_auto_runs_impact_when_existing_system(tmp_path): """门控：existing_system 非 None 且未显式传 impact_report → 自动跑影响调查，注入生成上下文。""" tpl = tmp_path / "tpl.docx" out = tmp_path / "out.docx" _make_template(str(tpl)) engine = FakeEngine() WriteOrchestrator().generate( _ss_with_existing(str(tpl)), str(out), samples_dir="nonexistent_dir_xyz", engine=engine, ) impact = engine.last_variables["impact"] assert "project_type=enhancement" in impact assert "OrderController" in impact def test_generate_without_existing_system_impact_empty(tmp_path): tpl = tmp_path / "tpl.docx" out = tmp_path / "out.docx" _make_template(str(tpl)) engine = FakeEngine() WriteOrchestrator().generate( _ss(str(tpl)), str(out), samples_dir="nonexistent_dir_xyz", engine=engine, ) assert engine.last_variables["impact"] == "" def test_generate_explicit_impact_report_used(tmp_path): from genesis.data_models import ChangeAnalysis, ChangeElement, ChangeType tpl = tmp_path / "tpl.docx" out = tmp_path / "out.docx" _make_template(str(tpl)) ca = ChangeAnalysis(project_type="enhancement", new_elements=[], modified_elements=[], deleted_elements=[], unchanged_elements=[], warnings=[]) report = ImpactReport(metadata={"version": "v1"}, change_analysis=ca, summary={"new": 0}) engine = FakeEngine() ss = _ss_with_existing(str(tpl)) WriteOrchestrator().generate(ss, str(out), samples_dir="nonexistent_dir_xyz", engine=engine, impact_report=report) assert engine.last_variables["impact"] != "" # 显式传入时也回填 structured_source，便于 QA/日志读取 assert ss.impact_report is report def test_generate_missing_template_path_raises(tmp_path): parsed = ParsedTemplate(file_name=None, sections=[], placeholders={}, styles={"used": []}) with pytest.raises(ValueError, match="template_path"): WriteOrchestrator().generate( SimpleNamespace(template=parsed), str(tmp_path / "out.docx"), samples_dir="nonexistent_dir_xyz", engine=FakeEngine(), )