"""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}"