2026Technology-Competition/tests/test_impact_agent.py

"""ImpactAgent 变更点定位测试（MVP 确定性规则）。"""
import pytest

from genesis.data_models import (
    CellValue, ChangeType, ControllerInfo, EntityInfo, ExcelTable,
    ExistingSystemInfo, Provenance, ServiceInfo, SheetType, StructuredSource,
)
from genesis.impact.impact_agent import ImpactAgent


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="samples/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"] == "samples/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