cma-management/backend/tests/test_map_objective_merge_20261012.py

"""按 objective 粒度合并（P2 cma-strategy-map-objective-merge, 2026-10-12）

残留面（A 上线后）：dimensions 仍是整表提交与覆写 ⇒ 两个会话编辑同一地图的**不同** objective 时，
后到者带过期基准被 409 拒绝（安全，但体验有税；409 语义被稀释）。

本文件断言 P2 落地后的语义：
  - 不同 objective 并发改 ⇒ 两个会话都 200，双方改动都在（他方新增/删除的节点被保留，不被整表回置）；
  - 同一 objective 并发改 ⇒ 后者 409（X-CMA-Concurrency-Code: stale_objective），且一行都不改；
  - 无脏项 ⇒ 不写（即使基准已过期也不 409、也不覆写）；
  - canvas_data：与库内现值一致 ⇒ 不参与覆写；有实质差异 ⇒ 退回 A 的 map 级校验（过期 409 stale_map）；
  - granular 通道仍要求基准（缺/不可解析 → 409）；A 的整表兜底不得移除（对照用例逐条验证）。

时间戳说明：库内 updated_at 由 SQLite 的 func.now()（秒级）产生，用例内的连续写无法自然拉开时刻，
故用 `_advance()` 显式推进 updated_at 若干秒，模拟「他方写已落库、本地基准已过期」的生产时刻差。
它只改时间戳，不改任何业务列，也不改断言目标。

用内存 SQLite（conftest），零生产库/零生产 Redis。
"""
import copy
from datetime import timedelta

from app.models import MapObjective, StrategicMap
from tests.conftest import auth_header, create_test_user, get_token_for_user

BASE = "2026-10-12T09:36:02"


def _dims(tag, names=("目标A", "目标B")):
    """两层维度结构：finance 带 objectives（名称可控），customer 空"""
    return [
        {"key": "finance", "name": "财务层", "icon": "💰", "color": "#F56C6C",
         "objectives": [{"name": n, "description": f"{tag}-{n}"} for n in names]},
        {"key": "customer", "name": "客户层", "icon": "👥", "color": "#409EFF", "objectives": []},
    ]


def _mk_user_and_token(client, db):
    create_test_user(db, username="mapadmin", role="ceo")
    return get_token_for_user(client, username="mapadmin")


def _create_map(client, headers, title="MERGE-P2", names=("目标A", "目标B")):
    resp = client.post("/api/cma/maps", headers=headers, json={
        "title": title, "dimensions": _dims("初始", names), "canvas_data": {"connections": []},
    })
    assert resp.status_code == 200, resp.text
    return resp.json()


def _snapshot(client, headers, map_id):
    """会话载入：GET /maps 取该地图（含 updated_at 基准 + dimensions 原文）"""
    return next(x for x in client.get("/api/cma/maps", headers=headers).json()["data"] if x["id"] == map_id)


def _advance(db, map_id, seconds=30):
    """显式推进服务端 updated_at：模拟「他方写落库后本地基准已过期」的时刻差（只动时间戳）"""
    db.expire_all()
    m = db.query(StrategicMap).filter(StrategicMap.id == map_id).first()
    m.updated_at = (m.updated_at or m.created_at) + timedelta(seconds=seconds)
    db.commit()
    db.expire_all()


def _map_dims(db, map_id):
    db.expire_all()
    m = db.query(StrategicMap).filter(StrategicMap.id == map_id).first()
    return copy.deepcopy(m.dimensions)


def _fin_objs(db, map_id):
    """finance 层 objectives（存库顺序）"""
    for d in _map_dims(db, map_id):
        if d.get("key") == "finance":
            return d.get("objectives") or []
    return []


def _obj_rows(db, map_id):
    """map_objectives 行（id, name, sort_order）：用于证明 409 时一行都不改"""
    db.expire_all()
    return [(o.id, o.name, o.sort_order) for o in
            db.query(MapObjective).filter(MapObjective.map_id == map_id)
            .order_by(MapObjective.sort_order, MapObjective.id).all()]


def _dirty(dim_key, name, base, value):
    return {"dim_key": dim_key, "objective_id": None, "name": name, "base": base, "value": value}


def _put(client, headers, map_id, dims, dirty, base, canvas=None):
    payload = {"dimensions": dims, "dirty_objectives": dirty, "base_updated_at": base}
    if canvas is not None:
        payload["canvas_data"] = canvas
    return client.put(f"/api/cma/maps/{map_id}", headers=headers, json=payload)


# ── ① 不同 objective 并发 ⇒ 均成功且双方改动都在 ────────────────────────

def test_two_sessions_edit_different_objectives_both_land(client, db):
    h = auth_header(_mk_user_and_token(client, db))
    m = _create_map(client, h)
    snap_a = _snapshot(client, h, m["id"])
    snap_b = _snapshot(client, h, m["id"])
    base = snap_a["updated_at"]

    # 会话A：改 目标A
    dims_a = copy.deepcopy(snap_a["dimensions"])
    dims_a[0]["objectives"][0]["description"] = "A改"
    r1 = _put(client, h, m["id"], dims_a,
              [_dirty("finance", "目标A", snap_a["dimensions"][0]["objectives"][0],
                      dims_a[0]["objectives"][0])],
              base, canvas=snap_a["canvas_data"])
    assert r1.status_code == 200, r1.text

    _advance(db, m["id"], 30)          # 时钟推进 ⇒ 会话B 的基准确定过期

    # 会话B：改 目标B（改的是**不同** objective ⇒ 不应被 409 拒）
    dims_b = copy.deepcopy(snap_b["dimensions"])
    dims_b[0]["objectives"][1]["description"] = "B改"
    r2 = _put(client, h, m["id"], dims_b,
              [_dirty("finance", "目标B", snap_b["dimensions"][0]["objectives"][1],
                      dims_b[0]["objectives"][1])],
              base, canvas=snap_b["canvas_data"])
    assert r2.status_code == 200, r2.text

    fin = _fin_objs(db, m["id"])
    assert [o["name"] for o in fin] == ["目标A", "目标B"]
    assert fin[0]["description"] == "A改"      # 会话A 的改动保留
    assert fin[1]["description"] == "B改"      # 会话B 的改动落库
    assert len(_obj_rows(db, m["id"])) == 2

    # 对照（A 的整表兜底不得移除）：同一份过期快照走旧通道（去掉 dirty_objectives）⇒ 409
    # 注：服务端每次写都会把 updated_at 刷成真实 now（SQLite func.now()，秒级），
    #    故对照前需再推进一次时刻，才能真正落到「基准过期」的生产形态。
    _advance(db, m["id"], 90)
    legacy = client.put(f"/api/cma/maps/{m['id']}", headers=h, json={
        "dimensions": dims_b, "canvas_data": snap_b["canvas_data"], "base_updated_at": base,
    })
    assert legacy.status_code == 409, legacy.text
    assert legacy.headers["X-CMA-Concurrency-Code"] == "stale_map"


# ── ② 同一 objective 并发 ⇒ 后者 409，且一行都不改 ───────────────────────

def test_same_objective_concurrent_second_is_409_and_nothing_changes(client, db):
    h = auth_header(_mk_user_and_token(client, db))
    m = _create_map(client, h)
    snap_a = _snapshot(client, h, m["id"])
    snap_b = _snapshot(client, h, m["id"])
    base = snap_a["updated_at"]

    dims_a = copy.deepcopy(snap_a["dimensions"])
    dims_a[0]["objectives"][0]["description"] = "A改"
    assert _put(client, h, m["id"], dims_a,
                [_dirty("finance", "目标A", snap_a["dimensions"][0]["objectives"][0],
                        dims_a[0]["objectives"][0])], base).status_code == 200

    _advance(db, m["id"], 30)
    rows_before = _obj_rows(db, m["id"])

    dims_b = copy.deepcopy(snap_b["dimensions"])
    dims_b[0]["objectives"][0]["description"] = "B改"
    r = _put(client, h, m["id"], dims_b,
             [_dirty("finance", "目标A", snap_b["dimensions"][0]["objectives"][0],
                     dims_b[0]["objectives"][0])], base)
    assert r.status_code == 409, r.text
    assert r.headers["X-CMA-Concurrency-Code"] == "stale_objective"
    assert "目标A" in r.json()["detail"]
    assert r.headers["X-CMA-Map-Updated-At"]

    assert _obj_rows(db, m["id"]) == rows_before                  # 行一行不动（含 id）
    assert _fin_objs(db, m["id"])[0]["description"] == "A改"        # 内容不被回置


# ── ③ 他方新增的节点被保留（残留缺陷面：不再被整表覆写删掉）────────────

def test_granular_merge_preserves_objective_added_by_other_session(client, db):
    h = auth_header(_mk_user_and_token(client, db))
    m = _create_map(client, h)
    snap_a = _snapshot(client, h, m["id"])
    snap_b = _snapshot(client, h, m["id"])

    # 会话A：新增 目标C（base=None ⇒ 本地新增）
    dims_a = copy.deepcopy(snap_a["dimensions"])
    dims_a[0]["objectives"].append({"name": "目标C", "description": "A新增"})
    r1 = _put(client, h, m["id"], dims_a,
              [_dirty("finance", "目标C", None, dims_a[0]["objectives"][2])],
              snap_a["updated_at"])
    assert r1.status_code == 200, r1.text
    assert [o["name"] for o in _fin_objs(db, m["id"])] == ["目标A", "目标B", "目标C"]

    _advance(db, m["id"], 30)

    # 会话B（快照里没有 目标C）：改 目标B ⇒ 200，且 目标C 必须还在
    dims_b = copy.deepcopy(snap_b["dimensions"])
    dims_b[0]["objectives"][1]["description"] = "B改"
    r2 = _put(client, h, m["id"], dims_b,
              [_dirty("finance", "目标B", snap_b["dimensions"][0]["objectives"][1],
                      dims_b[0]["objectives"][1])],
              snap_b["updated_at"])
    assert r2.status_code == 200, r2.text

    fin = _fin_objs(db, m["id"])
    assert [o["name"] for o in fin] == ["目标A", "目标B", "目标C"]
    assert fin[1]["description"] == "B改"
    assert fin[2]["description"] == "A新增"          # 他方新增节点未被本地过期快照删掉
    assert len(_obj_rows(db, m["id"])) == 3           # map_objectives 行同样未被删


# ── ④ 本地删除只删本地目标；未标脏的本地快照不回置服务端 ────────────────

def test_granular_delete_removes_only_local_target(client, db):
    h = auth_header(_mk_user_and_token(client, db))
    m = _create_map(client, h)
    snap_a = _snapshot(client, h, m["id"])
    snap_b = _snapshot(client, h, m["id"])

    # 会话A：删除 目标A（value=None）
    dims_a = copy.deepcopy(snap_a["dimensions"])
    dims_a[0]["objectives"] = [o for o in dims_a[0]["objectives"] if o["name"] != "目标A"]
    r1 = _put(client, h, m["id"], dims_a,
              [_dirty("finance", "目标A", snap_a["dimensions"][0]["objectives"][0], None)],
              snap_a["updated_at"])
    assert r1.status_code == 200, r1.text
    assert [o["name"] for o in _fin_objs(db, m["id"])] == ["目标B"]

    _advance(db, m["id"], 30)

    # 会话B（快照里仍有 目标A，但没有标脏它）：改 目标B ⇒ 200，且不得把 目标A 复活
    dims_b = copy.deepcopy(snap_b["dimensions"])
    dims_b[0]["objectives"][1]["description"] = "B改"
    r2 = _put(client, h, m["id"], dims_b,
              [_dirty("finance", "目标B", snap_b["dimensions"][0]["objectives"][1],
                      dims_b[0]["objectives"][1])],
              snap_b["updated_at"])
    assert r2.status_code == 200, r2.text

    fin = _fin_objs(db, m["id"])
    assert [o["name"] for o in fin] == ["目标B"]
    assert fin[0]["description"] == "B改"


# ── ⑤ 无脏项 ⇒ 不写（过期基准也不 409、也不覆写）──────────────────────

def test_granular_noop_payload_does_not_write_nor_409(client, db):
    h = auth_header(_mk_user_and_token(client, db))
    m = _create_map(client, h)
    snap = _snapshot(client, h, m["id"])

    dims = copy.deepcopy(snap["dimensions"])
    dims[0]["objectives"][0]["description"] = "A改"
    assert _put(client, h, m["id"], dims,
                [_dirty("finance", "目标A", snap["dimensions"][0]["objectives"][0],
                        dims[0]["objectives"][0])], snap["updated_at"]).status_code == 200
    after = _snapshot(client, h, m["id"])
    rows_after = _obj_rows(db, m["id"])

    _advance(db, m["id"], 30)

    # 会话B：全量快照仍是载入时的旧内容、无脏项、连线一致 ⇒ 未改动 ⇒ 不写、不 409
    r = _put(client, h, m["id"], snap["dimensions"], [], snap["updated_at"],
             canvas=snap["canvas_data"])
    assert r.status_code == 200, r.text

    now = _snapshot(client, h, m["id"])
    assert now["updated_at"] == after["updated_at"]          # 无写 ⇒ 基准未被推进
    assert _fin_objs(db, m["id"])[0]["description"] == "A改"    # 且未被本地过期快照回置
    assert _obj_rows(db, m["id"]) == rows_after


# ── ⑥ granular 通道仍 fail-closed：基准必需 ────────────────────────────

def test_granular_still_requires_base(client, db):
    h = auth_header(_mk_user_and_token(client, db))
    m = _create_map(client, h)
    snap = _snapshot(client, h, m["id"])
    dims = copy.deepcopy(snap["dimensions"])
    dirty = [_dirty("finance", "目标A", snap["dimensions"][0]["objectives"][0],
                    dims[0]["objectives"][0])]

    missing = client.put(f"/api/cma/maps/{m['id']}", headers=h,
                         json={"dimensions": dims, "dirty_objectives": dirty})
    assert missing.status_code == 409, missing.text
    assert missing.headers["X-CMA-Concurrency-Code"] == "concurrency_base_required"

    bad = client.put(f"/api/cma/maps/{m['id']}", headers=h,
                     json={"dimensions": dims, "dirty_objectives": dirty,
                           "base_updated_at": "not-a-timestamp"})
    assert bad.status_code == 409, bad.text
    assert bad.headers["X-CMA-Concurrency-Code"] == "concurrency_base_invalid"
    assert len(_obj_rows(db, m["id"])) == 2


# ── ⑦ canvas_data：一致 ⇒ 不覆写；有差异 + 过期基准 ⇒ 409 stale_map ─────

def test_granular_canvas_data_rules(client, db):
    h = auth_header(_mk_user_and_token(client, db))
    m = _create_map(client, h)
    snap = _snapshot(client, h, m["id"])
    dims = copy.deepcopy(snap["dimensions"])
    dims[0]["objectives"][0]["description"] = "A改"
    assert _put(client, h, m["id"], dims,
                [_dirty("finance", "目标A", snap["dimensions"][0]["objectives"][0],
                        dims[0]["objectives"][0])], snap["updated_at"]).status_code == 200

    _advance(db, m["id"], 30)

    dirty_b = [_dirty("finance", "目标B", snap["dimensions"][0]["objectives"][1],
                      snap["dimensions"][0]["objectives"][1])]

    # 连线有实质差异（新增一条）+ 过期基准 ⇒ 退回 A 的 map 级校验 → 409
    stale_canvas = _put(client, h, m["id"], snap["dimensions"], dirty_b, snap["updated_at"],
                        canvas={"connections": [{"from": "finance-0", "to": "customer-0"}]})
    assert stale_canvas.status_code == 409, stale_canvas.text
    assert stale_canvas.headers["X-CMA-Concurrency-Code"] == "stale_map"

    # 连线与库内现值一致 ⇒ 不参与覆写 ⇒ 只有脏 objective 走合并 → 200
    ok = _put(client, h, m["id"], snap["dimensions"], dirty_b, snap["updated_at"],
              canvas=snap["canvas_data"])
    assert ok.status_code == 200, ok.text
    db.expire_all()
    cur = db.query(StrategicMap).filter(StrategicMap.id == m["id"]).first()
    assert cur.canvas_data.get("connections") in ([], None)      # 连线未被本地旧快照覆写
    assert [o["name"] for o in _fin_objs(db, m["id"])] == ["目标A", "目标B"]


# ── ⑧ 本地重排（全部标脏、内容不变）⇒ 顺序被采纳，他方节点稳定排末尾 ────

def test_granular_reorder_adopted_and_foreign_node_kept_last(client, db):
    h = auth_header(_mk_user_and_token(client, db))
    m = _create_map(client, h)
    snap_a = _snapshot(client, h, m["id"])
    snap_b = _snapshot(client, h, m["id"])

    dims_a = copy.deepcopy(snap_a["dimensions"])
    dims_a[0]["objectives"].append({"name": "目标C", "description": "A新增"})
    assert _put(client, h, m["id"], dims_a,
                [_dirty("finance", "目标C", None, dims_a[0]["objectives"][2])],
                snap_a["updated_at"]).status_code == 200

    _advance(db, m["id"], 30)

    # 会话B：本地重排 [目标B, 目标A]（内容不变 ⇒ base 与服务端一致）
    dims_b = copy.deepcopy(snap_b["dimensions"])
    dims_b[0]["objectives"] = [dims_b[0]["objectives"][1], dims_b[0]["objectives"][0]]
    dirty_b = [
        _dirty("finance", "目标B", snap_b["dimensions"][0]["objectives"][1], dims_b[0]["objectives"][0]),
        _dirty("finance", "目标A", snap_b["dimensions"][0]["objectives"][0], dims_b[0]["objectives"][1]),
    ]
    r = _put(client, h, m["id"], dims_b, dirty_b, snap_b["updated_at"])
    assert r.status_code == 200, r.text

    fin = _fin_objs(db, m["id"])
    assert [o["name"] for o in fin] == ["目标B", "目标A", "目标C"]   # 本地顺序采纳；他方节点排末尾
    assert [row[2] for row in _obj_rows(db, m["id"])] == [0, 1, 2]