cma-management/backend/tests/test_map_connection_persistence_20261020.py

"""D2（2026-10-20 fullstack-20261020-strategy-map-3defects-fix）

缺陷：`DELETE /api/cma/maps/{id}/connections` 返回 200 但库里没变（POST 同样不可靠）。
根因：`StrategicMap.canvas_data` 是裸 `Column(JSON)`，无变更跟踪；接口用
「`m.canvas_data["connections"] = new_conns`」就地改，flush 时该列未被判脏 ⇒ 不写库。

判据铁律（任务 §1.2）：**不能**用「响应体里 connections 变短了」当落库证据 ——
必须新开 session 用原生 SQL 复读 `canvas_data`。
"""
import json

from sqlalchemy import text

from app.models import StrategicMap


def _raw_canvas_data(map_id: int) -> dict:
    """**新 session + 原生 SQL** 复读（不复用请求内 session）。"""
    from tests.conftest import TEST_SESSION_LOCAL

    session = TEST_SESSION_LOCAL()
    try:
        row = session.execute(
            text("SELECT canvas_data FROM strategic_maps WHERE id = :mid"), {"mid": map_id}
        ).first()
        assert row is not None, f"strategic_maps.id={map_id} 不存在"
        raw = row[0]
    finally:
        session.close()
    if isinstance(raw, (str, bytes)):
        return json.loads(raw)
    return raw or {}


def _pairs(canvas: dict):
    return [(c.get("from"), c.get("to")) for c in (canvas.get("connections") or [])]


def test_add_connection_persists_across_sessions(client, db):
    from tests.conftest import auth_header, create_test_map, create_test_user, get_token_for_user

    create_test_user(db)
    m = create_test_map(db, entity_id=1)  # canvas_data={"connections": []}
    db.commit()
    token = get_token_for_user(client)
    h = auth_header(token)

    resp = client.post(
        f"/api/cma/maps/{m.id}/connections",
        json={"from": "learning-0", "to": "process-0"},
        headers=h,
    )
    assert resp.status_code == 200, resp.text
    assert ("learning-0", "process-0") in _pairs(_raw_canvas_data(m.id)), (
        "POST /connections 返回 200 但新 session 原生 SQL 读不到该连线（未落库）"
    )


def test_delete_connection_persists_across_sessions(client, db):
    from tests.conftest import auth_header, create_test_map, create_test_user, get_token_for_user

    create_test_user(db)
    m = create_test_map(
        db,
        entity_id=1,
        canvas_data={"connections": [
            {"from": "learning-0", "to": "process-0", "style": "solid", "effect": "positive", "label": ""},
            {"from": "process-0", "to": "customer-0", "style": "solid", "effect": "positive", "label": ""},
        ]},
    )
    db.commit()
    token = get_token_for_user(client)
    h = auth_header(token)

    resp = client.request(
        "DELETE",
        f"/api/cma/maps/{m.id}/connections",
        json={"from": "learning-0", "to": "process-0"},
        headers=h,
    )
    assert resp.status_code == 200, resp.text

    persisted = _pairs(_raw_canvas_data(m.id))
    assert ("learning-0", "process-0") not in persisted, (
        f"DELETE 返回 200 但新 session 原生 SQL 仍读到已删连线：{persisted}"
    )
    assert ("process-0", "customer-0") in persisted, "无关连线被误删"


def test_delete_missing_connection_still_404(client, db):
    """负测：删除不存在的 from/to 仍 404（不得因标脏改动变成 200）。"""
    from tests.conftest import auth_header, create_test_map, create_test_user, get_token_for_user

    create_test_user(db)
    m = create_test_map(db, entity_id=1)
    db.commit()
    token = get_token_for_user(client)
    h = auth_header(token)

    resp = client.request(
        "DELETE",
        f"/api/cma/maps/{m.id}/connections",
        json={"from": "learning-9", "to": "process-9"},
        headers=h,
    )
    assert resp.status_code == 404, resp.text


def test_canvas_data_mutation_tracked_by_mutable_dict(db):
    """模型层双保险：MutableDict 包装后，**不经 flag_modified** 的 dict 赋值也应落库。

    说明：`MutableDict.as_mutable(JSON)` 返回的仍是 JSON 类型实例（跟踪靠 mapper 级
    attribute 事件 + coerce），因此这里用**行为断言**而不是断言列类型类名。
    """
    m = StrategicMap(title="MutableDict 行为验证", entity_id=1, canvas_data={"connections": []})
    db.add(m)
    db.commit()
    mid = m.id
    db.expire_all()

    loaded = db.query(StrategicMap).filter(StrategicMap.id == mid).first()
    # 就地改 dict（等价于接口里的 `m.canvas_data["connections"] = new_conns`）
    loaded.canvas_data["connections"] = [{"from": "learning-0", "to": "process-0"}]
    db.commit()  # 不调用 flag_modified

    assert ("learning-0", "process-0") in _pairs(_raw_canvas_data(mid)), (
        "canvas_data 未被 MutableDict 跟踪：dict 赋值后 commit 未落库"
    )