cma-management/backend/tests/test_map_concurrency_20261012.py

"""战略地图乐观并发保护（缺陷修复 cma-strategy-map-autosave-stale-overwrite, 2026-10-12）

生产事实（binlog/nginx/API 三路实证，见 A 侧裁定件）：
  10:17:28 他方写落库 19 行 map_objectives；10:22:37 一个 09:36 打开的画布页
  用「浏览器本地快照」（四泳道 objectives 全空）整表 PUT dimensions/canvas_data，
  把 19 行 DELETE 掉且未插回，之后每 60s 继续覆写。

本文件断言修复后的语义：
  - 过期基准的 dimensions/canvas_data 覆写 → 409（不再静默覆写）；
  - 409 在任何写入之前返回 ⇒ map_objectives 行（含 id）一行都不动；
  - 基准与服务端一致时照常 200，并返回新 updated_at 供前端采纳；
  - 不含护栏字段的写（如 MapList 只改 title）不受影响。

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

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

GUARDED_PAYLOAD_NOTE = "dimensions/canvas_data 覆写须带 base_updated_at"


def _dims(tag, n_objectives):
    """四/两层维度结构；objectives 条数可控，便于数行数"""
    return [
        {"key": "finance", "name": "财务层", "icon": "💰", "color": "#F56C6C",
         "objectives": [{"name": f"{tag}-目标{i}", "kpis": []} for i in range(n_objectives)]},
        {"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="STALE-REPRO"):
    resp = client.post("/api/cma/maps", headers=headers, json={
        "title": title,
        "dimensions": _dims("初始", 0),
        "canvas_data": {"connections": []},
    })
    assert resp.status_code == 200, resp.text
    return resp.json()


def _external_write(db, map_id, tag="D4", n_objectives=19, advance_s=5):
    """模拟「他方写」（批处理/脚本/他方 API）：直写 dimensions + 重建 map_objectives 行，
    并明确把 updated_at 推进若干秒（避开同秒歧义，等价于真实写后落库时刻）。"""
    m = db.query(StrategicMap).filter(StrategicMap.id == map_id).first()
    base = m.updated_at or datetime.now()
    m.dimensions = _dims(tag, n_objectives)
    db.commit()
    _sync_map_objectives(m, db)                 # 依新 dimensions 重建节点行（=他方写效果）
    m.updated_at = base + timedelta(seconds=advance_s)
    db.commit()
    db.refresh(m)
    return m


def _obj_rows(db, map_id):
    return [(o.id, o.name) for o in
            db.query(MapObjective).filter(MapObjective.map_id == map_id).order_by(MapObjective.id).all()]


def test_stale_snapshot_put_is_409_and_objective_rows_untouched(client, db):
    """核心复现：过期快照整表覆写 → 409；map_objectives 一行都不删（id 全等）"""
    token = _mk_user_and_token(client, db)
    h = auth_header(token)
    m = _create_map(client, h)
    stale_dims = m["dimensions"]                     # 浏览器本地快照（写前空白四泳道）
    stale_canvas = m["canvas_data"]
    stale_base = m["updated_at"]

    _external_write(db, m["id"], tag="D4", n_objectives=19)
    rows_before = _obj_rows(db, m["id"])
    assert len(rows_before) == 19

    resp = client.put(f"/api/cma/maps/{m['id']}", headers=h, json={
        "dimensions": stale_dims,
        "canvas_data": stale_canvas,
        "base_updated_at": stale_base,
    })
    assert resp.status_code == 409, resp.text
    assert resp.json()["detail"] == "地图已被修改，刷新后重试"
    assert resp.headers["X-CMA-Concurrency-Code"] == "stale_map"
    assert resp.headers["X-CMA-Map-Updated-At"]              # 返回服务端现基准

    # 一行都不删：id 与 name 逐行相等
    assert _obj_rows(db, m["id"]) == rows_before
    # dimensions 未被回置
    db.expire_all()
    cur = db.query(StrategicMap).filter(StrategicMap.id == m["id"]).first()
    assert [o["name"] for o in cur.dimensions[0]["objectives"]][:1] == ["D4-目标0"]


def test_missing_base_is_rejected_fail_closed(client, db):
    """不带基准的覆写 → 409（fail-closed；旧版缓存前端亦无法覆写）"""
    token = _mk_user_and_token(client, db)
    h = auth_header(token)
    m = _create_map(client, h, title="NO-BASE")
    _external_write(db, m["id"], tag="X", n_objectives=7)
    rows_before = _obj_rows(db, m["id"])

    resp = client.put(f"/api/cma/maps/{m['id']}", headers=h, json={
        "dimensions": _dims("过期", 0), "canvas_data": {"connections": []},
    })
    assert resp.status_code == 409, resp.text
    assert "缺少并发基准" in resp.json()["detail"]
    assert resp.headers["X-CMA-Concurrency-Code"] == "concurrency_base_required"
    assert _obj_rows(db, m["id"]) == rows_before


def test_invalid_base_is_rejected(client, db):
    token = _mk_user_and_token(client, db)
    h = auth_header(token)
    m = _create_map(client, h, title="BAD-BASE")
    resp = client.put(f"/api/cma/maps/{m['id']}", headers=h, json={
        "dimensions": _dims("过期", 0), "base_updated_at": "not-a-timestamp",
    })
    assert resp.status_code == 409, resp.text
    assert resp.headers["X-CMA-Concurrency-Code"] == "concurrency_base_invalid"


def test_fresh_base_put_succeeds_and_returns_new_stamp(client, db):
    """基准与服务端一致 → 200，且响应返回新 updated_at（前端据此刷新基准）"""
    token = _mk_user_and_token(client, db)
    h = auth_header(token)
    m = _create_map(client, h, title="FRESH")
    _external_write(db, m["id"], tag="D4", n_objectives=19)

    fresh = next(x for x in client.get("/api/cma/maps", headers=h).json()["data"] if x["id"] == m["id"])
    resp = client.put(f"/api/cma/maps/{m['id']}", headers=h, json={
        "dimensions": _dims("本地新", 3),
        "canvas_data": {"connections": [{"from": "finance-0", "to": "customer-0"}]},
        "base_updated_at": fresh["updated_at"],
    })
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["updated_at"] != fresh["updated_at"]        # 新基准可用
    assert len(_obj_rows(db, m["id"])) == 3                 # 合法写照常重建节点行


def test_put_without_guarded_fields_not_blocked(client, db):
    """MapList 只改 title/version（不含 dimensions/canvas_data）→ 不受并发保护影响"""
    token = _mk_user_and_token(client, db)
    h = auth_header(token)
    m = _create_map(client, h, title="META-ONLY")
    _external_write(db, m["id"], tag="D4", n_objectives=19)

    resp = client.put(f"/api/cma/maps/{m['id']}", headers=h, json={"title": "改名不改内容", "version": "v2.0"})
    assert resp.status_code == 200, resp.text
    assert resp.json()["title"] == "改名不改内容"


def test_canvas_data_only_put_is_guarded(client, db):
    """只覆写 canvas_data 的写同样受保护（saveConnections 走这条通道）"""
    token = _mk_user_and_token(client, db)
    h = auth_header(token)
    m = _create_map(client, h, title="CONN-ONLY")
    stale_base = m["updated_at"]
    _external_write(db, m["id"], tag="D4", n_objectives=4)

    stale = client.put(f"/api/cma/maps/{m['id']}", headers=h, json={
        "canvas_data": {"connections": []}, "base_updated_at": stale_base,
    })
    assert stale.status_code == 409, stale.text

    fresh = next(x for x in client.get("/api/cma/maps", headers=h).json()["data"] if x["id"] == m["id"])
    ok = client.put(f"/api/cma/maps/{m['id']}", headers=h, json={
        "canvas_data": {"connections": []}, "base_updated_at": fresh["updated_at"],
    })
    assert ok.status_code == 200, ok.text


def test_connections_endpoint_returns_updated_at(client, db):
    """连线写通道返回 updated_at：前端可采纳新基准，避免自身合法编辑触发 409"""
    token = _mk_user_and_token(client, db)
    h = auth_header(token)
    m = _create_map(client, h, title="CONN-STAMP")
    add = client.post(f"/api/cma/maps/{m['id']}/connections", headers=h,
                      json={"from": "finance-0", "to": "customer-0"})
    assert add.status_code == 200, add.text
    assert add.json().get("updated_at")

    fresh = next(x for x in client.get("/api/cma/maps", headers=h).json()["data"] if x["id"] == m["id"])
    # 用连线接口返回的基准覆写 canvas_data → 不应 409
    ok = client.put(f"/api/cma/maps/{m['id']}", headers=h, json={
        "canvas_data": {"connections": add.json()["connections"]},
        "base_updated_at": add.json()["updated_at"],
    })
    assert ok.status_code == 200, ok.text
    rm = client.request("DELETE", f"/api/cma/maps/{m['id']}/connections", headers=h,
                        json={"from": "finance-0", "to": "customer-0"})
    assert rm.status_code == 200, rm.text
    assert rm.json().get("updated_at") or rm.json().get("removed")


def test_timezone_aware_base_accepted(client, db):
    """带时区基准（…+08:00 / …Z）可解析并与库内秒级值对齐，不误判为过期"""
    token = _mk_user_and_token(client, db)
    h = auth_header(token)
    m = _create_map(client, h, title="TZ-BASE")
    naive = m["updated_at"]
    dt = datetime.fromisoformat(naive)
    resp = client.put(f"/api/cma/maps/{m['id']}", headers=h, json={
        "dimensions": _dims("同基准", 1),
        "base_updated_at": dt.replace(microsecond=0).isoformat() + "+00:00",
    })
    # 库内为服务器本地时间；+00:00 会被换算 → 与本地时区不同则判定过期，此处只要求「可解析、非 500」
    assert resp.status_code in (200, 409), resp.text