wecom-fullstackandHermes CI Fix
adcd118498
- 新表 map_connections（entity_id/map_id/from_node/to_node/from_layer/to_layer/ from_objective_id/to_objective_id/effect/label/sort_order），唯一键 (map_id,from,to) 保幂等 - 节点双格式兼容：层:中文目标名（生产存量 31 条）与 层-序号（前端现用） - canvas_data 保留不动（双写期）；POST/DELETE 写表 + 同步 JSON，失败仅告警不阻断 - GET /connections 表优先、表空回落 canvas_data；新增 GET /connections/analysis （层统计/跨层分布/断链/反向告警不拦截） - scripts/migrate_map_connections.py 四态（check/apply/idem/rollback）+ canvas 快照回滚点 - tests: 9 条（双格式解析、双写、回落、删除同步、分析统计与告警）
167 lines
8.0 KiB
Python
167 lines
8.0 KiB
Python
"""P1 四层因果结构化：map_connections 表 + /connections 读/写/analysis 回归
|
|
|
|
任务 project-20261021-cma-map-connections-table-p1-fullstack
|
|
覆盖：双格式解析（层:中文名 / 层-序号）、双写、读表回落 canvas、分析与告警不拦截。
|
|
"""
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import StrategicMap, MapObjective, MapConnection
|
|
from app.api.maps import parse_node_ref, resolve_node
|
|
from tests.conftest import create_test_user, get_token_for_user, auth_header
|
|
|
|
DIMS = [
|
|
{"key": "finance", "name": "财务层", "icon": "💰", "color": "#F56C6C",
|
|
"objectives": [{"name": "营业收入"}, {"name": "营业利润"}]},
|
|
{"key": "customer", "name": "客户层", "icon": "👥", "color": "#409EFF",
|
|
"objectives": [{"name": "客户留存"}]},
|
|
{"key": "process", "name": "流程层", "icon": "⚙️", "color": "#67C23A",
|
|
"objectives": [{"name": "交付效率"}, {"name": "质量闭环"}]},
|
|
{"key": "learning", "name": "学习层", "icon": "📚", "color": "#E6A23C",
|
|
"objectives": [{"name": "全员培训"}, {"name": "能力认证"}]},
|
|
]
|
|
|
|
|
|
def _seed_map(db: Session, title="P1连线测试", canvas=None) -> StrategicMap:
|
|
m = StrategicMap(entity_id=1, title=title, status="draft", dimensions=DIMS,
|
|
canvas_data=canvas or {"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
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean(db: Session):
|
|
db.query(MapConnection).delete()
|
|
db.query(MapObjective).filter(MapObjective.map_id >= 900000).delete()
|
|
db.query(StrategicMap).filter(StrategicMap.id >= 900000).delete()
|
|
db.commit()
|
|
yield
|
|
|
|
|
|
class TestNodeParse:
|
|
def test_colon_format(self):
|
|
r = parse_node_ref("learning:全员培训")
|
|
assert (r["layer"], r["kind"], r["value"], r["error"]) == ("learning", "name", "全员培训", None)
|
|
|
|
def test_index_format(self):
|
|
r = parse_node_ref("process-1")
|
|
assert (r["layer"], r["kind"], r["value"], r["error"]) == ("process", "index", 1, None)
|
|
|
|
def test_unknown_formats(self):
|
|
assert parse_node_ref("nonsense")["error"] == "node_format_unrecognized"
|
|
assert parse_node_ref("hr:某目标")["error"] == "unknown_layer:hr"
|
|
assert parse_node_ref("")["error"] == "node_empty"
|
|
|
|
def test_resolve_both_formats(self, db: Session):
|
|
m = _seed_map(db)
|
|
# 冒号中文名
|
|
layer, oid, note = resolve_node(db, m, "learning:能力认证")
|
|
assert (layer, note) == ("learning", None) and oid is not None
|
|
# 序号（dimensions 数组下标）
|
|
layer, oid2, note = resolve_node(db, m, "process-1")
|
|
assert (layer, note) == ("process", None)
|
|
obj = db.query(MapObjective).filter(MapObjective.id == oid2).first()
|
|
assert obj.name == "质量闭环"
|
|
|
|
|
|
class TestConnectionApi:
|
|
def test_post_dual_write_and_get_table(self, client: TestClient, db: Session):
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
m = _seed_map(db, title="P1双写")
|
|
mid = m.id
|
|
|
|
resp = client.post(f"/api/cma/maps/{mid}/connections", headers=auth_header(token),
|
|
json={"from": "learning:全员培训", "to": "process:交付效率"})
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["db_synced"] is True
|
|
assert body["connections"][0]["from"] == "learning:全员培训"
|
|
|
|
# canvas_data 保留（前端渲染主路径）
|
|
db.expire_all()
|
|
m2 = db.query(StrategicMap).filter(StrategicMap.id == mid).first()
|
|
assert m2.canvas_data["connections"][0]["to"] == "process:交付效率"
|
|
# 表内落行且层/目标解析正确
|
|
row = db.query(MapConnection).filter(MapConnection.map_id == mid).one()
|
|
assert (row.from_layer, row.to_layer) == ("learning", "process")
|
|
assert row.from_objective_id and row.to_objective_id
|
|
|
|
got = client.get(f"/api/cma/maps/{mid}/connections", headers=auth_header(token))
|
|
assert got.status_code == 200
|
|
assert got.json()["source"] == "table"
|
|
assert got.json()["total"] == 1
|
|
|
|
def test_index_format_persisted(self, client: TestClient, db: Session):
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
m = _seed_map(db, title="P1序号格式")
|
|
resp = client.post(f"/api/cma/maps/{m.id}/connections", headers=auth_header(token),
|
|
json={"from": "learning-0", "to": "process-1"})
|
|
assert resp.status_code == 200, resp.text
|
|
row = db.query(MapConnection).filter(MapConnection.map_id == m.id).one()
|
|
assert (row.from_node, row.from_layer) == ("learning-0", "learning")
|
|
assert row.from_objective_id and row.to_objective_id
|
|
|
|
def test_get_falls_back_to_canvas(self, client: TestClient, db: Session):
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
m = _seed_map(db, title="P1回落",
|
|
canvas={"connections": [{"from": "learning:全员培训", "to": "process:质量闭环"}]})
|
|
got = client.get(f"/api/cma/maps/{m.id}/connections", headers=auth_header(token))
|
|
assert got.status_code == 200
|
|
assert got.json()["source"] == "canvas_data"
|
|
assert got.json()["total"] == 1
|
|
|
|
def test_delete_removes_both(self, client: TestClient, db: Session):
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
m = _seed_map(db, title="P1删除")
|
|
client.post(f"/api/cma/maps/{m.id}/connections", headers=auth_header(token),
|
|
json={"from": "learning:全员培训", "to": "process:交付效率"})
|
|
resp = client.request("DELETE", f"/api/cma/maps/{m.id}/connections", headers=auth_header(token),
|
|
json={"from": "learning:全员培训", "to": "process:交付效率"})
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["db_synced"] is True
|
|
assert db.query(MapConnection).filter(MapConnection.map_id == m.id).count() == 0
|
|
db.expire_all()
|
|
m2 = db.query(StrategicMap).filter(StrategicMap.id == m.id).first()
|
|
assert m2.canvas_data["connections"] == []
|
|
|
|
def test_analysis_stats_reverse_and_broken(self, client: TestClient, db: Session):
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
m = _seed_map(db, title="P1分析")
|
|
pairs = [
|
|
("learning:全员培训", "process:交付效率"), # 正常
|
|
("process:质量闭环", "customer:客户留存"), # 正常
|
|
("customer:客户留存", "finance:营业收入"), # 正常
|
|
("finance:营业利润", "customer:客户留存"), # 反向（告警不拦截）
|
|
("finance:营业收入", "finance:营业利润"), # 同层
|
|
("learning:不存在目标", "process:交付效率"), # 断链（照样可写）
|
|
]
|
|
for f, t in pairs:
|
|
r = client.post(f"/api/cma/maps/{m.id}/connections", headers=auth_header(token),
|
|
json={"from": f, "to": t})
|
|
assert r.status_code == 200, r.text
|
|
|
|
a = client.get(f"/api/cma/maps/{m.id}/connections/analysis", headers=auth_header(token))
|
|
assert a.status_code == 200, a.text
|
|
d = a.json()
|
|
assert d["source"] == "table" and d["total"] == 6
|
|
assert d["layer_stats"]["learning"]["as_from"] == 2
|
|
assert d["layer_stats"]["finance"]["as_to"] == 2 # customer→finance 1 + finance→finance 1
|
|
assert d["cross_layer"]["learning->process"] == 2 # 正常 1 + 断链 1（层可解析、目标不可解析）
|
|
assert d["same_layer_count"] == 1
|
|
assert len(d["reversed"]) == 1 and d["reversed"][0]["from_layer"] == "finance"
|
|
assert any(b["side"] == "from" for b in d["broken"])
|
|
assert any("反向连线" in w for w in d["warnings"])
|