"""ImpactAgent 变更点定位测试（MVP 确定性规则）。""" import pytest from genesis.data_models import ( CellValue, ChangeType, ControllerInfo, EntityInfo, ExcelTable, ExistingSystemInfo, Provenance, RuleDocument, ServiceInfo, SheetType, StructuredSource, ) from genesis.impact.impact_agent import ImpactAgent, impact_report_to_dict def _cv(value) -> CellValue: return CellValue(value=value, provenance=Provenance("f.xlsx", "S", 1, "A", "h")) def _row(*values: str) -> dict[str, CellValue]: headers = ["要素ID", "要素名", "変更区分", "既存対応"] return {h: _cv(v) for h, v in zip(headers, values)} def _table(name: str, stype: SheetType, rows: list[dict]) -> ExcelTable: return ExcelTable(name=name, detected_type=stype, extraction_method="openpyxl", headers=["要素ID", "要素名", "変更区分", "既存対応"], rows=rows) def _existing() -> ExistingSystemInfo: return ExistingSystemInfo( controller_layer=[ ControllerInfo("OrderController", "OrderController", "OrderController.java", "/trade/order", ["/trade/order/{id}"], "trade-order/OrderController.java"), ControllerInfo("RealtimeMarketDataController", "RealtimeMarketDataController", "R.java", "/market-data/realtime", [], "trade-market-data/R.java"), ], service_layer=[ ServiceInfo("OrderServiceImpl", "OrderServiceImpl", "OrderServiceImpl.java", ["createOrder"], "trade-order/OrderServiceImpl.java"), ], entity_layer=[ EntityInfo("OrderDO", "OrderDO", "OrderDO.java", "trade_order", ["id"], "trade-order/OrderDO.java"), EntityInfo("StrategyBacktest", "StrategyBacktest", "SB.java", "ts_strategy_backtest", [], "trade-strategy/SB.java"), ], api_endpoints=[], source_path="sample/existing-system", ) def _ss(existing: ExistingSystemInfo, tables: list[ExcelTable]) -> StructuredSource: return StructuredSource(tables=tables, template=None, rule_docs=[], image_analyses=[], existing_system=existing, comments=[]) def test_classify_new_modified_deleted(): tables = [ _table("機能一覧", SheetType.FUNCTION, [ _row("F001", "止损风控机能", "新規", ""), _row("F002", "订单状态查询扩展", "変更", "OrderController,OrderServiceImpl"), _row("F003", "实时行情显示扩展", "変更", "RealtimeMarketDataController"), _row("F004", "旧回测机能废止", "削除", "StrategyBacktest"), ]), ] report = ImpactAgent().run(_ss(_existing(), tables)) ca = report.change_analysis assert ca.project_type == "enhancement" assert [e.element_id for e in ca.new_elements] == ["F001"] assert ca.new_elements[0].change_type is ChangeType.NEW assert [e.element_id for e in ca.modified_elements] == ["F002", "F003"] m2 = ca.modified_elements[0] assert m2.impacted_existing == ["OrderController", "OrderServiceImpl"] assert m2.evidence # 命中类 source_uri assert [e.element_id for e in ca.deleted_elements] == ["F004"] assert ca.deleted_elements[0].impacted_existing == ["StrategyBacktest"] assert ca.warnings == [] def test_prefix_mapping_matches_impl_and_entity(): tables = [_table("機能一覧", SheetType.FUNCTION, [ _row("F005", "订单查询", "変更", "OrderService"), # 前缀匹配 OrderServiceImpl _row("F006", "订单表变更", "変更", "OrderDO"), ])] ca = ImpactAgent().run(_ss(_existing(), tables)).change_analysis m = {e.element_id: e for e in ca.modified_elements} assert "OrderServiceImpl" in m["F005"].impacted_existing assert m["F006"].impacted_existing == ["OrderDO"] def test_new_with_mapping_is_conflict_warning(): tables = [_table("機能一覧", SheetType.FUNCTION, [_row("F007", "新建订单接口", "新規", "OrderController")])] report = ImpactAgent().run(_ss(_existing(), tables)) el = report.change_analysis.new_elements[0] assert el.status == "conflict" assert any(w.element_id == "F007" for w in report.change_analysis.warnings) def test_modified_without_mapping_warns(): tables = [_table("機能一覧", SheetType.FUNCTION, [_row("F008", "订单变更", "変更", "")])] report = ImpactAgent().run(_ss(_existing(), tables)) assert report.change_analysis.modified_elements[0].status == "warning" assert any(w.element_id == "F008" for w in report.change_analysis.warnings) def test_modified_mapping_no_match_warns(): tables = [_table("機能一覧", SheetType.FUNCTION, [_row("F009", "账户变更", "変更", "AccountController")])] report = ImpactAgent().run(_ss(_existing(), tables)) assert report.change_analysis.modified_elements[0].status == "warning" assert any("AccountController" in w.issue for w in report.change_analysis.warnings) def test_unchanged_count_in_summary(): tables = [_table("機能一覧", SheetType.FUNCTION, [_row("F002", "订单查询", "変更", "OrderController")])] report = ImpactAgent().run(_ss(_existing(), tables)) assert report.summary["unchanged"] >= 2 # 既有 5 类中未被引用者 def test_table_without_change_column_skipped(): tbl = ExcelTable(name="帳票一覧", detected_type=SheetType.REPORT, extraction_method="openpyxl", headers=["帳票ID", "帳票名"], rows=[]) report = ImpactAgent().run(_ss(_existing(), [tbl])) assert report.change_analysis.new_elements == [] assert report.change_analysis.warnings == [] def test_run_without_existing_raises(): ss = StructuredSource(tables=[], template=None, rule_docs=[], image_analyses=[], existing_system=None, comments=[]) with pytest.raises(ValueError, match="既有系统"): ImpactAgent().run(ss) def test_metadata_and_summary_shape(): tables = [_table("機能一覧", SheetType.FUNCTION, [_row("F001", "新功能", "新規", "")])] report = ImpactAgent().run(_ss(_existing(), tables), session_id="sess-1") assert report.metadata["version"] == "v1" assert report.metadata["session_id"] == "sess-1" assert report.metadata["source"] == "sample/existing-system" assert report.summary["total"] == 1 assert report.summary["new"] == 1 # ---------- 覆盖率补全（分支覆盖） ---------- def test_scope_logs_warning(caplog): tables = [_table("機能一覧", SheetType.FUNCTION, [_row("F001", "新功能", "新規", "")])] with caplog.at_level("WARNING", logger="genesis.impact.impact_agent"): ImpactAgent().run(_ss(_existing(), tables), scope={"modules": ["trade-order"]}) assert any("scope" in r.message for r in caplog.records) def test_empty_and_unknown_change_values_skipped(): def cv(v): return CellValue(v, Provenance("f.xlsx", "S", 1, "A", "h")) headers = ["要素ID", "要素名", "変更区分", "既存対応"] rows = [ {"要素ID": cv("F100"), "要素名": cv("空区分"), "変更区分": cv(""), "既存対応": cv("")}, {"要素ID": cv("F101"), "要素名": cv("未知区分"), "変更区分": cv("未知"), "既存対応": cv("")}, {"要素ID": cv("F102"), "要素名": cv("正常"), "変更区分": cv("新規"), "既存対応": cv("")}, ] tbl = ExcelTable("機能一覧", SheetType.FUNCTION, "openpyxl", headers, rows) report = ImpactAgent().run(_ss(_existing(), [tbl])) assert [e.element_id for e in report.change_analysis.new_elements] == ["F102"] assert report.change_analysis.warnings == [] def test_deleted_without_mapping_warns_and_lists(): tables = [_table("機能一覧", SheetType.FUNCTION, [_row("F030", "旧功能", "削除", "")])] report = ImpactAgent().run(_ss(_existing(), tables)) assert report.change_analysis.deleted_elements[0].status == "warning" assert any(w.element_id == "F030" for w in report.change_analysis.warnings) def test_entity_without_table_name_matches_by_class(): existing = _existing() existing.entity_layer.append(EntityInfo("NoTable", "NoTable", "NT.java", None, [], "NT.java")) tables = [_table("DB定義", SheetType.DATABASE, [_row("T100", "无表实体", "変更", "NoTable")])] report = ImpactAgent().run(_ss(existing, tables)) assert report.change_analysis.modified_elements[0].impacted_existing == ["NoTable"] def test_matching_dedup_same_class_via_class_and_table(): tables = [_table("DB定義", SheetType.DATABASE, [ _row("T101", "订单表", "変更", "OrderDO,trade_order"), _row("T102", "订单综合", "変更", "Order,Order"), ])] ca = ImpactAgent().run(_ss(_existing(), tables)).change_analysis by_id = {e.element_id: e for e in ca.modified_elements} assert by_id["T101"].impacted_existing == ["OrderDO"] # 类名+表名 → 同一实体去重 assert by_id["T102"].impacted_existing.count("OrderController") == 1 # 前缀重复去重 def test_cell_out_of_range_returns_empty(): assert ImpactAgent._cell({}, ["A"], 5) == "" def test_impact_report_to_dict_serializable(): from genesis.impact.impact_agent import impact_report_to_dict tables = [_table("機能一覧", SheetType.FUNCTION, [ _row("F001", "止损风控", "新規", ""), _row("F002", "订单查询", "変更", "OrderController"), ])] report = ImpactAgent().run(_ss(_existing(), tables), session_id="s") d = impact_report_to_dict(report) assert d["metadata"]["session_id"] == "s" assert d["change_analysis"]["new_elements"][0]["change_type"] == "新規" assert d["change_analysis"]["modified_elements"][0]["impacted_existing"] == ["OrderController"] assert d["summary"]["modified"] == 1 def _doc(name: str, content: str) -> RuleDocument: return RuleDocument(file_name=name, category="design", markdown_content=content, source_path=name, file_type="word", hash="h") def _ss_with_design(existing, tables, docs) -> StructuredSource: return StructuredSource(tables=tables, template=None, rule_docs=[], image_analyses=[], existing_system=existing, comments=[], design_docs=docs) def test_design_doc_cross_reference_hits_identifier(): docs = [_doc("design_a.docx", "本章描述 OrderController 的限价单处理逻辑与风控。")] tables = [_table("機能一覧", SheetType.FUNCTION, [_row("F001", "新功能", "新規", "")])] report = ImpactAgent().run(_ss_with_design(_existing(), tables, docs), session_id="d") refs = report.design_references ids = {r.identifier for r in refs} assert "OrderController" in ids assert refs[0].doc_name == "design_a.docx" assert "OrderController" in refs[0].snippet # 序列化也包含 assert any(r["identifier"] == "OrderController" for r in impact_report_to_dict(report)["design_references"]) def test_design_doc_cross_reference_empty_when_no_docs(): tables = [_table("機能一覧", SheetType.FUNCTION, [_row("F001", "新功能", "新規", "")])] report = ImpactAgent().run(_ss_with_design(_existing(), tables, []), session_id="d") assert report.design_references == [] def test_design_doc_with_empty_content_skipped(): docs = [_doc("empty.docx", "")] # 无正文 → continue（line 164） tables = [_table("機能一覧", SheetType.FUNCTION, [_row("F001", "新功能", "新規", "")])] report = ImpactAgent().run(_ss_with_design(_existing(), tables, docs), session_id="d") assert report.design_references == [] # 无正文设计文档被跳过，line 165 正常执行