lhl
becd3e1f57
按《参赛成果物提交规范·赛道一》§6 红线： - samples/ 目录改名 sample/（git mv，保留历史） - 10 个中日文样本文件 + docs 参赛手册 PDF 重命名为 ASCII （requirements_*/template_*/rules_*/contestant-handbook.pdf） - tests/test_zh_template.py 硬编码绝对路径 D:\00_project\Genesis 改为相对路径 - 全局更新 21 个活动文件引用；历史日志/审查文档不改（追加说明记录） 全量 pytest 431 passed / 99.15%
204 lines
9.5 KiB
Python
204 lines
9.5 KiB
Python
"""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="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
|