proposa-ai/scripts/tests/test_pipeline_guards.py

import inspect
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

from models import (
    AgentState,
    BidOutline,
    Chapter,
    ChapterType,
    ParsedDocument,
    ProjectData,
    ReviewIssue,
    ScoringCriterion,
    TenderAnalysis,
)
from workflow import export_bid_document
from step3_outlining import _build_outline_cache_fingerprint
from step4_writing import write_content
from step5_reviewing.fixer import _Fixer, _insert_into_existing_section
from step5_reviewing.reviewer import (
    _Reviewer,
    _balanced_outline_excerpt,
    _canonical_outline_content,
    _chapter_effective_content,
)
from step6_exporting.docx_builder import (
    _fill_chapters_into_template,
    _try_assemble_from_template,
)


class OutlineCacheTests(unittest.TestCase):
    def test_fingerprint_changes_with_project_and_scoring(self):
        analysis_a = TenderAnalysis(
            project_name="项目A",
            scoring_criteria=[
                ScoringCriterion("SC-1", "技术", "需求理解", "分析重点", 10)
            ],
        )
        analysis_b = TenderAnalysis(
            project_name="项目B",
            scoring_criteria=[
                ScoringCriterion("SC-1", "技术", "需求理解", "分析重点", 10)
            ],
        )
        pd_a = ProjectData(project_id="A", project_name="项目A")
        pd_b = ProjectData(project_id="B", project_name="项目B")
        fingerprint_a = _build_outline_cache_fingerprint(analysis_a, pd_a)
        fingerprint_b = _build_outline_cache_fingerprint(analysis_b, pd_b)
        self.assertNotEqual(fingerprint_a, fingerprint_b)

        analysis_a.scoring_criteria[0].description = "分析重点和难点"
        fingerprint_changed = _build_outline_cache_fingerprint(analysis_a, pd_a)
        self.assertNotEqual(fingerprint_a, fingerprint_changed)


class ReviewCorpusTests(unittest.TestCase):
    def test_demand_chapter_requires_local_scoring_dimensions(self):
        demand = Chapter(
            id="3", title="需求理解", generated_content="项目概况与一般特点分析"
        )
        scoring_node = Chapter(
            id="4.1", title="需求理解（6分）", level=2,
            related_criteria=["SC-02"], generated_content="第四章已有完整响应",
        )
        outline = BidOutline(project_name="测试", chapters=[
            demand,
            Chapter(id="4", title="基本服务方案", children=[scoring_node]),
        ])
        criterion = ScoringCriterion(
            "SC-02", "技术方案", "需求理解",
            "分析服务定位和预期目标，分析重点难点并提出应对或改进措施。", 6,
        )
        reviewer = _Reviewer.__new__(_Reviewer)

        issues = reviewer._check_demand_understanding_scoring_coverage(
            outline, [criterion]
        )

        self.assertEqual(len(issues), 1)
        self.assertEqual(issues[0].chapter_id, "3")
        self.assertEqual(issues[0].issue_type, "demand_scoring_coverage")

    def test_restructure_parent_content_satisfies_legacy_empty_children(self):
        outline = BidOutline(
            project_name="测试项目",
            chapters=[
                Chapter(
                    id="3",
                    title="需求理解",
                    chapter_type=ChapterType.TECHNICAL,
                    generated_content="根据评分项重构后的完整需求理解正文",
                    children=[
                        Chapter(
                            id="3.1",
                            title="旧模板子节",
                            chapter_type=ChapterType.TECHNICAL,
                        )
                    ],
                )
            ],
        )

        reviewer = _Reviewer.__new__(_Reviewer)
        self.assertEqual(reviewer._check_empty_chapters(outline), [])

    def test_parent_and_children_are_not_double_counted(self):
        child = Chapter(id="3.1", title="子节", generated_content="子节正文")
        parent = Chapter(
            id="3",
            title="需求理解",
            generated_content="父章已汇总子节正文",
            children=[child],
        )
        outline = BidOutline(project_name="测试", chapters=[parent])
        self.assertEqual(_chapter_effective_content(parent), "父章已汇总子节正文")
        corpus = _canonical_outline_content(outline)
        self.assertEqual(corpus.count("子节正文"), 1)

    def test_balanced_excerpt_contains_late_chapters(self):
        outline = BidOutline(
            project_name="测试",
            chapters=[
                Chapter(id=str(i), title=f"章节{i}", generated_content=(str(i) * 4000))
                for i in range(1, 9)
            ],
        )
        excerpt = _balanced_outline_excerpt(outline, max_chars=12000)
        self.assertIn("【1 章节1】", excerpt)
        self.assertIn("【8 章节8】", excerpt)


class FixerTests(unittest.TestCase):
    def test_demand_scoring_patch_stays_in_chapter_three_without_new_heading(self):
        class FakeLlm:
            def generate(self, **_kwargs):
                return "一、新增标题\n结合场馆开放特点制定分区分时保障方案。"

        demand = Chapter(
            id="3", title="需求理解",
            generated_content="一、项目概况\n原有正文",
        )
        scoring_node = Chapter(
            id="4.1", title="需求理解（6分）", level=2,
            related_criteria=["SC-02"], generated_content="第四章原文",
        )
        outline = BidOutline(project_name="测试", chapters=[
            demand,
            Chapter(id="4", title="基本服务方案", children=[scoring_node]),
        ])
        analysis = TenderAnalysis(
            project_name="测试",
            scoring_criteria=[ScoringCriterion(
                "SC-02", "技术方案", "需求理解",
                "分析服务定位和预期目标，分析重点难点并提出应对或改进措施。", 6,
            )],
        )
        issue = ReviewIssue(
            chapter_id="3", severity="error",
            issue_type="demand_scoring_coverage",
            description="需求理解章未完整覆盖评分项 [SC-02] 需求理解",
        )
        fixer = object.__new__(_Fixer)
        fixer.llm = FakeLlm()
        fixer.project_data = None

        fixer._fix_demand_understanding_criterion(outline, issue, analysis)

        self.assertIn("服务定位", demand.generated_content)
        self.assertIn("预期目标", demand.generated_content)
        self.assertIn("重点与难点", demand.generated_content)
        self.assertIn("应对或改进措施", demand.generated_content)
        self.assertNotIn("一、新增标题", demand.generated_content)
        self.assertEqual(scoring_node.generated_content, "第四章原文")

    def test_missing_criterion_patch_targets_deepest_mapped_heading(self):
        class FakeLlm:
            def generate(self, **_kwargs):
                return "一、模型自造标题\n补充后的针对性响应正文"

        scoring_node = Chapter(
            id="4.1.1",
            title="（一）保洁作业流程（8分）",
            level=3,
            structure_locked=True,
            related_criteria=["SC-1"],
        )
        parent = Chapter(
            id="4.1", title="一、保洁服务", level=2,
            related_criteria=["SC-1"], children=[scoring_node],
        )
        top = Chapter(
            id="4", title="基本服务方案", level=1,
            chapter_type=ChapterType.TECHNICAL,
            generated_content="原有顶层正文",
            related_criteria=["SC-1"], children=[parent],
        )
        outline = BidOutline(project_name="测试", chapters=[top])
        analysis = TenderAnalysis(
            project_name="测试",
            scoring_criteria=[
                ScoringCriterion(
                    "SC-1", "基本服务", "保洁作业流程", "流程完整", 8
                )
            ],
        )
        issue = ReviewIssue(
            chapter_id="*", severity="error", issue_type="missing_criterion",
            description="[LLM深度检查] 评分项 [SC-1] 保洁作业流程 未充分覆盖",
        )
        fixer = object.__new__(_Fixer)
        fixer.llm = FakeLlm()
        fixer.project_data = None
        fixer._appended_chapters = set()

        fixer._fix_missing_criterion(outline, issue, analysis)

        self.assertEqual(top.generated_content, "原有顶层正文")
        self.assertIn("补充后的针对性响应正文", scoring_node.generated_content)
        self.assertNotIn("一、模型自造标题", scoring_node.generated_content)

    def test_scoring_patch_is_inserted_under_matching_existing_heading(self):
        content = (
            "一、保洁服务\n原保洁正文\n"
            "二、绿化服务\n原绿化正文"
        )

        result = _insert_into_existing_section(
            content,
            "补充保洁作业流程和质量检查要求。",
            ["保洁", "作业流程"],
        )

        self.assertLess(result.index("补充保洁"), result.index("二、绿化服务"))
        self.assertEqual(
            [line for line in result.splitlines() if line.startswith(("一、", "二、"))],
            ["一、保洁服务", "二、绿化服务"],
        )

    def test_word_expansion_only_targets_restructure_chapter(self):
        class FakeLlm:
            def generate(self, **_kwargs):
                return "一、新增章节标题\n新增的实施细节正文"

        fixer = object.__new__(_Fixer)
        fixer.llm = FakeLlm()
        fixer.project_data = None
        preserved = Chapter(
            id="6", title="项目经理", generated_content="模板结构正文"
        )
        scoring = Chapter(
            id="3.1", title="一、需求理解", level=2,
            generated_content="原有评分响应正文",
            related_criteria=["SC-02"], structure_locked=True,
        )
        restructure = Chapter(
            id="3", title="需求理解", generated_content="模板需求理解正文",
            children=[scoring],
        )
        locked = Chapter(
            id="4", title="基本服务方案", generated_content="原有方案正文"
        )
        outline = BidOutline(
            project_name="测试", chapters=[preserved, restructure, locked]
        )
        analysis = TenderAnalysis(project_name="测试")

        fixer._fix_word_count(outline, 1000, analysis)

        self.assertEqual(preserved.generated_content, "模板结构正文")
        self.assertEqual(locked.generated_content, "原有方案正文")
        self.assertEqual(restructure.generated_content, "模板需求理解正文")
        self.assertTrue(scoring.generated_content.startswith("原有评分响应正文"))
        self.assertIn("新增的实施细节正文", scoring.generated_content)
        self.assertNotIn("一、新增章节标题", scoring.generated_content)


class PipelinePolicyTests(unittest.TestCase):
    def test_step4_no_longer_calls_full_chapter_alignment(self):
        self.assertNotIn(
            "_llm_apply_requirement_alignment(", inspect.getsource(write_content)
        )

    def test_step6_formal_path_has_no_native_table_structure_mutations(self):
        source = inspect.getsource(_try_assemble_from_template)
        self.assertEqual(source.count("_resolve_remaining_table_placeholders("), 1)
        self.assertNotIn("_dedupe_identical_tables(", source)
        self.assertNotIn("_append_missing_rejections_to_commitment_tables(", source)
        self.assertNotIn("_append_compliance_checklist(", source)
        self.assertNotIn("_center_document_tables(", source)
        chapter_source = inspect.getsource(_fill_chapters_into_template)
        self.assertNotIn("_append_unpaired_tender_tables(", chapter_source)

    def test_workflow_preserves_requested_final_path_and_ends_at_step6(self):
        with tempfile.TemporaryDirectory() as temp_dir:
            final_path = str(Path(temp_dir) / "用户指定结果.docx")
            project_data = ProjectData(
                project_id="P-1",
                project_name="测试项目",
                reference_bids=[
                    ParsedDocument(
                        file_path="reference.docx",
                        file_name="reference.docx",
                        content="",
                    )
                ],
            )
            state = AgentState(
                project_name="测试项目",
                project_data=project_data,
                tender_analysis=TenderAnalysis(project_name="测试项目"),
                outline=BidOutline(
                    project_name="测试项目",
                    chapters=[Chapter(id="3", title="需求理解", generated_content="正文")],
                ),
                output_path=final_path,
            )

            def fake_export(**kwargs):
                self.assertEqual(kwargs["output_path"], final_path)
                self.assertEqual(kwargs["reference_bid_path"], "reference.docx")
                return kwargs["output_path"]

            fake_cfg = SimpleNamespace(
                output_dir=temp_dir,
                save_chapter_docs=False,
            )
            with patch("workflow.get_config", return_value=fake_cfg), \
                    patch("workflow.export_to_docx", side_effect=fake_export):
                result = export_bid_document(state)

        self.assertEqual(result["output_path"], final_path)
        self.assertEqual(result["current_step"], "export")


if __name__ == "__main__":
    unittest.main()