wecom-fullstack
b3f31f9ac9
ci/woodpecker/push/main Pipeline failed
- map_connections 读侧以 from/to_objective_id 反查当前 dimensions 树定位节点，改名/重排不再静默断线 - 悬停/拖拽期 map_objectives 未返回时按位置兜底（仅同层同序位有效），并记录 stale_objective_id - 新增 tests/test_conn_key_rename_stability.py（含 2 条保守反错绑负测用例） - 零 DDL、零 health_score 改动（问3=A）
416 lines
20 KiB
Python
416 lines
20 KiB
Python
"""P3 阶段A · 连线键解耦（读侧解析键）验收测试 —— 沙箱，不连生产。
|
||
|
||
判据（对应 P3 单验收标准 2/3）：
|
||
1. 目标改名后 `from_key/to_key` 跟随**同一节点**（key 由 objective_id 反查，不随中文名变）；
|
||
2. 索引位移（删掉前一个目标）后 key 跟随节点到新位置；
|
||
3. 负测：拿掉 `objective_id` 锚点（等价于"未实现解析键"）且名字已改 ⇒ 必须解析失败
|
||
并被计入 `unresolved`，证明判据有区分度、不是空过。
|
||
"""
|
||
import copy
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import StrategicMap, MapObjective, MapConnection
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header
|
||
|
||
DIMS = [
|
||
{"key": "finance", "name": "财务层", "icon": "F", "color": "#F56C6C",
|
||
"objectives": [{"name": "营业收入"}, {"name": "营业利润"}]},
|
||
{"key": "customer", "name": "客户层", "icon": "C", "color": "#409EFF",
|
||
"objectives": [{"name": "客户留存"}]},
|
||
{"key": "process", "name": "流程层", "icon": "P", "color": "#67C23A",
|
||
"objectives": [{"name": "交付效率"}, {"name": "质量闭环"}]},
|
||
{"key": "learning", "name": "学习层", "icon": "L", "color": "#E6A23C",
|
||
"objectives": [{"name": "全员培训"}, {"name": "能力认证"}]},
|
||
]
|
||
|
||
|
||
def _seed(db: Session) -> StrategicMap:
|
||
m = StrategicMap(entity_id=1, title="p3-conn-key", status="draft",
|
||
dimensions=copy.deepcopy(DIMS), canvas_data={"connections": []})
|
||
db.add(m)
|
||
db.commit()
|
||
db.refresh(m)
|
||
for d in DIMS:
|
||
for i, o in enumerate(d["objectives"]):
|
||
db.add(MapObjective(entity_id=1, map_id=m.id, dimension_key=d["key"],
|
||
name=o["name"], sort_order=i))
|
||
db.commit()
|
||
return m
|
||
|
||
|
||
def _save_dims(client, headers, db: Session, m: StrategicMap, dims) -> None:
|
||
"""整表 PUT（legacy 通道，批处理/脚本真实会走这条）"""
|
||
db.refresh(m)
|
||
body = {"dimensions": dims,
|
||
"canvas_data": copy.deepcopy(m.canvas_data),
|
||
"base_updated_at": m.updated_at.isoformat() if m.updated_at else None}
|
||
r = client.put(f"/api/cma/maps/{m.id}", json=body, headers=headers)
|
||
assert r.status_code == 200, f"整表 PUT 失败: {r.status_code} {r.text[:300]}"
|
||
|
||
|
||
def _conn(client, headers, map_id: int) -> dict:
|
||
r = client.get(f"/api/cma/maps/{map_id}/connections", headers=headers)
|
||
assert r.status_code == 200, r.text[:300]
|
||
return r.json()
|
||
|
||
|
||
def _mk_conn(client, headers, map_id: int, frm: str, to: str) -> None:
|
||
r = client.post(f"/api/cma/maps/{map_id}/connections", headers=headers,
|
||
json={"from": frm, "to": to})
|
||
assert r.status_code == 200, f"建连线失败: {r.status_code} {r.text[:300]}"
|
||
|
||
|
||
def test_rename_keeps_render_key(client: TestClient, db: Session):
|
||
"""UI 通道（带回 `_mo_id`）改名 → 连线仍挂同一节点（key 不变、unresolved=0）
|
||
|
||
口径 B（ASK-20261023-04，任总 2026-10-23 15:13「放行 4」）：只有带回 `_mo_id` 的
|
||
通道能精确保住锚点；不带 `_mo_id` 的老通道改名会断线但可见，
|
||
见 test_rename_without_moid_breaks_visibly。
|
||
"""
|
||
create_test_user(db)
|
||
h = auth_header(get_token_for_user(client))
|
||
m = _seed(db)
|
||
_mk_conn(client, h, m.id, "learning:全员培训", "process:交付效率")
|
||
|
||
base = _conn(client, h, m.id)
|
||
assert base["total"] == 1 and base["unresolved"] == 0, base
|
||
c0 = base["connections"][0]
|
||
assert c0["from_key"] == "learning-0" and c0["to_key"] == "process-0", c0
|
||
assert c0["from_key_source"] == "objective_id"
|
||
|
||
# 真实 UI 通道：先读（读侧 m_to_dict→_merge_map_objectives 回写 `_mo_id`），改名后整表回传
|
||
r = client.get("/api/cma/maps", headers=h)
|
||
assert r.status_code == 200, r.text[:300]
|
||
got_map = next(x for x in r.json()["data"] if x["id"] == m.id)
|
||
dims = got_map["dimensions"]
|
||
assert dims[3]["objectives"][0].get("_mo_id"), "读侧未回写 _mo_id，UI 通道无法保锚点"
|
||
dims[3]["objectives"][0]["name"] = "全员培训A"
|
||
_save_dims(client, h, db, m, dims)
|
||
|
||
after = _conn(client, h, m.id)
|
||
assert after["total"] == 1, f"连线数减少: {after}"
|
||
assert after["unresolved"] == 0, after["unresolved_items"]
|
||
a0 = after["connections"][0]
|
||
# 原始串仍是旧名（禁回填），但渲染键已由锚点解析到新节点
|
||
assert a0["from"] == "learning:全员培训", a0
|
||
assert a0["from_key"] == "learning-0", a0
|
||
assert a0["from_key_source"] == "objective_id" and a0["from_key_reason"] is None, a0
|
||
|
||
|
||
def test_index_shift_follows_node(client: TestClient, db: Session):
|
||
"""删除同层前一个目标 ⇒ 节点索引位移，key 跟随节点而非冻结在旧序号"""
|
||
create_test_user(db)
|
||
h = auth_header(get_token_for_user(client))
|
||
m = _seed(db)
|
||
_mk_conn(client, h, m.id, "learning:能力认证", "process:交付效率")
|
||
|
||
base = _conn(client, h, m.id)["connections"][0]
|
||
assert base["from_key"] == "learning-1", base
|
||
|
||
dims = copy.deepcopy(m.dimensions)
|
||
dims[3]["objectives"] = dims[3]["objectives"][1:] # 删掉「全员培训」
|
||
_save_dims(client, h, db, m, dims)
|
||
|
||
after = _conn(client, h, m.id)
|
||
assert after["unresolved"] == 0, after["unresolved_items"]
|
||
a0 = after["connections"][0]
|
||
assert a0["from_key"] == "learning-0", f"key 未跟随节点位移: {a0}"
|
||
assert a0["from_key_source"] == "objective_id", a0
|
||
|
||
|
||
def test_negative_without_anchor_must_fail(client: TestClient, db: Session):
|
||
"""负测：拿掉 objective_id 锚点 + 名字已改 ⇒ 必须解析失败并计入 unresolved
|
||
|
||
证明判据有区分度：若解析键能力缺失（=改造前行为），这条必然断线且被计数，
|
||
不会静默画到错节点上。
|
||
"""
|
||
create_test_user(db)
|
||
h = auth_header(get_token_for_user(client))
|
||
m = _seed(db)
|
||
_mk_conn(client, h, m.id, "learning:全员培训", "process:交付效率")
|
||
|
||
dims = copy.deepcopy(m.dimensions)
|
||
dims[3]["objectives"][0]["name"] = "全员培训A"
|
||
_save_dims(client, h, db, m, dims)
|
||
|
||
# 模拟"无锚点"（历史脏行 / 未回填）：抹掉 from_objective_id
|
||
row = db.query(MapConnection).filter(MapConnection.map_id == m.id).first()
|
||
row.from_objective_id = None
|
||
db.commit()
|
||
|
||
broken = _conn(client, h, m.id)
|
||
b0 = broken["connections"][0]
|
||
assert b0["from_key"] is None, f"无锚点 + 旧名不应解析成功: {b0}"
|
||
assert b0["from_key_reason"] == "objective_unresolved", b0
|
||
assert broken["unresolved"] == 1, broken["unresolved_items"]
|
||
assert broken["unresolved_items"][0]["side"] == "from"
|
||
assert broken["unresolved_items"][0]["reason"] == "objective_unresolved"
|
||
# 另一端不受影响
|
||
assert b0["to_key"] == "process-0" and b0["to_key_source"] == "objective_id", b0
|
||
|
||
|
||
def test_fallback_name_source_still_works(client: TestClient, db: Session):
|
||
"""兼容期：历史行无 objective_id 但名字未改 ⇒ 仍按 `层:名字` 解析成功（不误报未解析）"""
|
||
create_test_user(db)
|
||
h = auth_header(get_token_for_user(client))
|
||
m = _seed(db)
|
||
_mk_conn(client, h, m.id, "learning:全员培训", "process:交付效率")
|
||
|
||
row = db.query(MapConnection).filter(MapConnection.map_id == m.id).first()
|
||
row.from_objective_id = None
|
||
db.commit()
|
||
|
||
got = _conn(client, h, m.id)
|
||
assert got["unresolved"] == 0, got["unresolved_items"]
|
||
c0 = got["connections"][0]
|
||
assert c0["from_key"] == "learning-0" and c0["from_key_source"] == "name", c0
|
||
|
||
|
||
# ── project 2026-10-22 23:25 裁定的 3 项准入条件（阶段B 开跑前补测）─────────────
|
||
# 条件2-a：连续两次保存（不改名，不带 _mo_id）⇒ map_objectives 行 id 集合不变（幂等）；
|
||
# 条件2-b：同一次保存内「改名 + 同层调序」且不带 _mo_id ⇒ 两个目标的 id 不得串位；
|
||
# 条件3 ：删除面 —— dims 少传整层 ⇒ 该层行才删；被删目标的连线端计入 unresolved，不得静默消失。
|
||
# 另：hold 口径断言 total = green + yellow + red + hold（hold 不得二次计入 miss）。
|
||
|
||
def _row_ids(db: Session, map_id: int) -> dict:
|
||
"""{(层, 名字): 行 id}"""
|
||
rows = db.query(MapObjective).filter(MapObjective.map_id == map_id).all()
|
||
return {(r.dimension_key, r.name): r.id for r in rows}
|
||
|
||
|
||
def test_save_twice_is_idempotent(client: TestClient, db: Session):
|
||
"""连续两次保存不改名 ⇒ 行 id 集合逐字不变（否则锚点必成野指针）"""
|
||
create_test_user(db)
|
||
h = auth_header(get_token_for_user(client))
|
||
m = _seed(db)
|
||
_mk_conn(client, h, m.id, "learning:全员培训", "process:交付效率")
|
||
|
||
first = _row_ids(db, m.id)
|
||
dims = copy.deepcopy(m.dimensions)
|
||
_save_dims(client, h, db, m, dims) # 第 1 次保存（结构未变）
|
||
second = _row_ids(db, m.id)
|
||
_save_dims(client, h, db, m, dims) # 第 2 次保存
|
||
third = _row_ids(db, m.id)
|
||
|
||
assert first == second, f"首次保存换了行 id: {first} -> {second}"
|
||
assert second == third, f"二次保存换了行 id: {second} -> {third}"
|
||
got = _conn(client, h, m.id)
|
||
assert got["unresolved"] == 0, got["unresolved_items"]
|
||
assert got["connections"][0]["from_key"] == "learning-0"
|
||
|
||
|
||
def test_rename_plus_reorder_without_moid_no_crossbind(client: TestClient, db: Session):
|
||
"""口径 B：同一次保存内「改名 + 同层调序」且不带 _mo_id ⇒ 不得串位
|
||
|
||
名字层先把未改名项钉住原行；改名项不再按位置兜底 ⇒ 当新行（旧行删除），
|
||
连线锚点指向旧行 ⇒ unresolved + stale_objective_id（断线可见，绝不静默错绑）。
|
||
"""
|
||
create_test_user(db)
|
||
h = auth_header(get_token_for_user(client))
|
||
m = _seed(db)
|
||
_mk_conn(client, h, m.id, "learning:能力认证", "process:交付效率")
|
||
before = _row_ids(db, m.id)
|
||
assert _conn(client, h, m.id)["connections"][0]["from_key"] == "learning-1"
|
||
|
||
# 调序（能力认证升到第 0 位）+ 改名（能力认证→能力认证A），整表 PUT，不带 _mo_id
|
||
dims = copy.deepcopy(m.dimensions)
|
||
dims[3]["objectives"] = [{"name": "能力认证A"}, {"name": "全员培训"}]
|
||
_save_dims(client, h, db, m, dims)
|
||
|
||
after = _row_ids(db, m.id)
|
||
assert after[("learning", "全员培训")] == before[("learning", "全员培训")], \
|
||
f"未改名的「全员培训」行 id 变了（串位）: {before} -> {after}"
|
||
assert after[("learning", "能力认证A")] != before[("learning", "能力认证")], \
|
||
f"口径 B 下老通道改名应新建行（不做位置兜底）: {before} -> {after}"
|
||
|
||
got = _conn(client, h, m.id)
|
||
assert got["total"] == 1, got
|
||
c0 = got["connections"][0]
|
||
assert got["unresolved"] == 1, got["unresolved_items"]
|
||
assert c0["from"] == "learning:能力认证", "原始串被改写（禁回填）"
|
||
assert c0["from_key"] is None and c0["from_key_reason"] == "stale_objective_id", c0
|
||
# 关键：锚点不得被改指到别的节点（旧行为会把线静默挂到「全员培训」上）
|
||
row = db.query(MapConnection).filter(MapConnection.map_id == m.id).first()
|
||
assert row.from_objective_id == before[("learning", "能力认证")], "连线锚点被静默改指（错绑）"
|
||
|
||
|
||
def test_rename_without_moid_breaks_visibly(client: TestClient, db: Session):
|
||
"""口径 B 单点验证：不带 `_mo_id` 的纯改名 ⇒ 新行 + 连线进 unresolved（断线可见）
|
||
|
||
同通道下「删A加B」与「A改名为B」数据上不可区分，位置兜底会让前者静默错绑；
|
||
放弃兜底的代价就是这里：改名会断线，但断线在接口里可见、可计数。
|
||
"""
|
||
create_test_user(db)
|
||
h = auth_header(get_token_for_user(client))
|
||
m = _seed(db)
|
||
_mk_conn(client, h, m.id, "learning:全员培训", "process:交付效率")
|
||
before = _row_ids(db, m.id)
|
||
|
||
dims = copy.deepcopy(m.dimensions)
|
||
assert not any("_mo_id" in o for o in dims[3]["objectives"]), "fixture 不应带 _mo_id（老通道）"
|
||
dims[3]["objectives"][0]["name"] = "全员培训A"
|
||
_save_dims(client, h, db, m, dims)
|
||
|
||
after = _row_ids(db, m.id)
|
||
assert after[("learning", "全员培训A")] != before[("learning", "全员培训")], \
|
||
f"老通道改名竟保住锚点（口径 B 下应为新行）: {before} -> {after}"
|
||
assert ("learning", "全员培训") not in after, f"旧行未删: {after}"
|
||
|
||
got = _conn(client, h, m.id)
|
||
assert got["total"] == 1, f"连线行被静默丢弃: {got}"
|
||
c0 = got["connections"][0]
|
||
assert c0["from"] == "learning:全员培训", "原始串被改写（禁回填）"
|
||
assert c0["from_key"] is None and c0["from_key_reason"] == "stale_objective_id", c0
|
||
assert got["unresolved"] == 1 and got["unresolved_items"][0]["reason"] == "stale_objective_id"
|
||
assert c0["to_key"] == "process-0", "另一端不应受影响"
|
||
row = db.query(MapConnection).filter(MapConnection.map_id == m.id).first()
|
||
assert row.from_objective_id == before[("learning", "全员培训")], "连线锚点被静默改指（错绑）"
|
||
|
||
|
||
def test_delete_face_unresolved_not_silent(client: TestClient, db: Session):
|
||
"""删除面：dims 少传整层 ⇒ 该层行删；被删目标的连线端计入 unresolved 且原始串保留"""
|
||
create_test_user(db)
|
||
h = auth_header(get_token_for_user(client))
|
||
m = _seed(db)
|
||
_mk_conn(client, h, m.id, "learning:全员培训", "process:交付效率")
|
||
|
||
dims = copy.deepcopy(m.dimensions)
|
||
dims = dims[:3] # 整层去掉 learning
|
||
_save_dims(client, h, db, m, dims)
|
||
|
||
left = db.query(MapObjective).filter(MapObjective.map_id == m.id,
|
||
MapObjective.dimension_key == "learning").all()
|
||
assert left == [], f"该层已不在 dims，行未删: {[r.id for r in left]}"
|
||
|
||
got = _conn(client, h, m.id)
|
||
assert got["total"] == 1, f"连线被静默丢弃: {got}" # 行仍在，不消失
|
||
c0 = got["connections"][0]
|
||
assert c0["from"] == "learning:全员培训", "原始串被改写（禁回填）"
|
||
# 锚点指向已删行 ⇒ reason 是更精确的 `stale_objective_id`（区别于「无锚点」的 objective_unresolved）
|
||
assert c0["from_key"] is None and c0["from_key_reason"] == "stale_objective_id", c0
|
||
assert got["unresolved"] == 1, got["unresolved_items"]
|
||
assert got["unresolved_items"][0]["side"] == "from"
|
||
assert c0["to_key"] == "process-0", "另一端不应受影响"
|
||
|
||
|
||
def test_review_hold_total_invariant(client: TestClient, db: Session):
|
||
"""hold 口径：total = green + yellow + red + hold；hold 不得二次计入 miss（红/黄）"""
|
||
from app.models import KPIDefinition, KPIValue
|
||
from app.api.maps import DATA_HOLD_KPIS
|
||
from tests.conftest import create_test_kpi
|
||
|
||
create_test_user(db)
|
||
h = auth_header(get_token_for_user(client))
|
||
hold_code = next(iter(DATA_HOLD_KPIS))
|
||
red_code = "P3_TOTAL_INVARIANT_RED"
|
||
db.query(KPIDefinition).filter(KPIDefinition.kpi_code == red_code).delete(synchronize_session=False)
|
||
db.commit()
|
||
red_kpi = create_test_kpi(db, kpi_code=red_code, kpi_name="P3 红项", dimension="customer",
|
||
target_value=100.0, kpi_level="strategic", unit="%")
|
||
db.add(KPIValue(kpi_id=red_kpi.id, period="2026-09", actual_value=10.0, source_type="actual"))
|
||
db.commit()
|
||
|
||
dims = [{"key": "customer", "name": "客户层", "icon": "", "color": "#409EFF", "objectives": [
|
||
{"name": "待核项", "kpis": [hold_code]},
|
||
{"name": "红项", "kpis": [red_code]},
|
||
{"name": "无指标项"},
|
||
]}]
|
||
m = StrategicMap(entity_id=1, title="p3-hold-invariant", status="active",
|
||
dimensions=dims, canvas_data={"connections": []})
|
||
db.add(m)
|
||
db.commit()
|
||
db.refresh(m)
|
||
|
||
r = client.get(f"/api/cma/maps/{m.id}/review", params={"level": "strategic"}, headers=h)
|
||
assert r.status_code == 200, r.text
|
||
s = r.json()["summary"]
|
||
assert s["total"] == 3, s
|
||
assert s["hold"] == 1, s
|
||
assert s["red"] == 1, s
|
||
assert s["green"] == 1, s
|
||
assert s["total"] == s["green"] + s["yellow"] + s["red"] + s["hold"], \
|
||
f"分母不守恒（hold 被二次计入?）: {s}"
|
||
|
||
db.query(KPIValue).filter(KPIValue.kpi_id == red_kpi.id).delete(synchronize_session=False)
|
||
db.query(KPIDefinition).filter(KPIDefinition.kpi_code == red_code).delete(synchronize_session=False)
|
||
db.commit()
|
||
|
||
|
||
|
||
|
||
def test_negative_greedy_matching_would_crossbind(client: TestClient, db: Session, monkeypatch):
|
||
"""负测（判据有效性的反证）：把定位算法换回历史「逐项贪心（名→序位→空位）」，
|
||
则 rename+reorder 场景下两个目标的 id 必然交换 —— 证明
|
||
test_rename_plus_reorder_without_moid_no_crossbind 有区分度，不是空过。"""
|
||
from app.api import maps as maps_mod
|
||
|
||
def legacy_greedy(m, db_session=None):
|
||
sess = db_session or db
|
||
dims = maps_mod._dims_list(m.dimensions)
|
||
ent = m.entity_id if m.entity_id is not None else 1
|
||
rows = sess.query(MapObjective).filter(MapObjective.map_id == m.id).order_by(
|
||
MapObjective.sort_order, MapObjective.id).all()
|
||
used, keep = set(), []
|
||
for dim in dims:
|
||
dk = dim.get("key", "")
|
||
for i, obj in enumerate(dim.get("objectives") or []):
|
||
if not isinstance(obj, dict):
|
||
continue
|
||
target = None
|
||
want = maps_mod._norm_obj_name(obj.get("name"))
|
||
for r in rows: # ① 名字
|
||
if r.id in used or r.dimension_key != dk:
|
||
continue
|
||
if maps_mod._norm_obj_name(r.name) == want:
|
||
target = r
|
||
break
|
||
if target is None:
|
||
for r in rows: # ② 同层同序位
|
||
if r.id in used or r.dimension_key != dk:
|
||
continue
|
||
if (r.sort_order or 0) == i:
|
||
target = r
|
||
break
|
||
if target is None:
|
||
for r in rows: # ③ 同层空位
|
||
if r.id in used or r.dimension_key != dk:
|
||
continue
|
||
target = r
|
||
break
|
||
if target is None:
|
||
target = MapObjective(map_id=m.id, entity_id=ent, dimension_key=dk or "finance")
|
||
sess.add(target)
|
||
else:
|
||
used.add(target.id)
|
||
target.entity_id = ent
|
||
target.dimension_key = dk
|
||
target.name = (obj.get("name") or "")[:200]
|
||
target.icon = obj.get("icon") or "target"
|
||
target.kpis = obj.get("kpis") or []
|
||
target.sort_order = i
|
||
keep.append(target)
|
||
sess.flush()
|
||
for r in rows:
|
||
if r.id not in used:
|
||
sess.delete(r)
|
||
sess.commit()
|
||
|
||
create_test_user(db)
|
||
h = auth_header(get_token_for_user(client))
|
||
m = _seed(db)
|
||
before = _row_ids(db, m.id)
|
||
|
||
monkeypatch.setattr(maps_mod, "_sync_map_objectives", legacy_greedy)
|
||
dims = copy.deepcopy(m.dimensions)
|
||
dims[3]["objectives"] = [{"name": "能力认证A"}, {"name": "全员培训"}]
|
||
_save_dims(client, h, db, m, dims)
|
||
after = _row_ids(db, m.id)
|
||
|
||
assert after[("learning", "全员培训")] != before[("learning", "全员培训")], \
|
||
f"旧算法竟未串位，则该负测无法证明新判据有效: {before} -> {after}"
|
||
assert after[("learning", "能力认证A")] == before[("learning", "全员培训")], \
|
||
f"预期旧算法把「能力认证A」绑到「全员培训」的旧行 id（串位）: {before} -> {after}"
|