""" 战略地图版本管理完整实现测试（project-20260908-map-versions-page） 覆盖: 列表摘要/详情全量/回滚自动备份/备注编辑/跨entity 404/不存在404 """ import pytest from fastapi.testclient import TestClient from sqlalchemy.orm import Session from app.models import Entity, User, UserEntity, StrategicMap, StrategicMapVersion from tests.conftest import create_test_user, get_token_for_user, auth_header def _create_second_entity(db: Session, user: User) -> int: db.add(Entity(id=2, name="第二企业", short_name="企业2", status="active")) db.add(UserEntity(user_id=user.id, entity_id=2, granted_by=1)) db.commit() return 2 def _token_for_entity(client: TestClient, entity_id: int) -> str: resp = client.post("/api/cma/auth/login", json={ "username": "testadmin", "password": "admin123", "entity_id": entity_id, }) assert resp.status_code == 200 data = resp.json() return data.get("token") or data.get("access_token") def _create_published_map(client: TestClient, token: str, title: str) -> int: """建模板地图 → 发布 → 返回 map_id（自动产生 v1.0 快照）""" create_resp = client.post( "/api/cma/maps/create-with-template", headers=auth_header(token), json={"title": title}, ) assert create_resp.status_code == 200 map_id = create_resp.json()["id"] pub_resp = client.put( f"/api/cma/maps/{map_id}", headers=auth_header(token), json={"status": "published"}, ) assert pub_resp.status_code == 200 return map_id class TestMapVersions: """战略地图版本 endpoints""" def test_list_versions_empty_then_publish_has_summary(self, client, db): create_test_user(db) token = get_token_for_user(client) # 未发布空列表 m = client.post("/api/cma/maps/create-with-template", headers=auth_header(token), json={"title": "版本测试A"}) map_id = m.json()["id"] empty = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)) assert empty.status_code == 200 assert empty.json()["data"] == [] # 发布后 v1.0 快照（列表带摘要字段） client.put(f"/api/cma/maps/{map_id}", headers=auth_header(token), json={"status": "published"}) ver = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)) assert ver.status_code == 200 rows = ver.json()["data"] assert len(rows) == 1 r = rows[0] assert r["version"] == "v1.0" assert r["is_current"] is True assert r["dimension_count"] == 4 assert r["objective_count"] > 0 assert isinstance(r["kpi_count"], int) assert r["comment"] == "发布 v1.0" def test_get_version_detail_has_full_content(self, client, db): create_test_user(db) token = get_token_for_user(client) map_id = _create_published_map(client, token, "版本详情B") versions = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"] vid = versions[0]["id"] detail = client.get(f"/api/cma/maps/{map_id}/versions/{vid}", headers=auth_header(token)) assert detail.status_code == 200 d = detail.json() assert isinstance(d["dimensions"], list) and len(d["dimensions"]) == 4 assert isinstance(d["canvas_data"], dict) assert d["version"] == "v1.0" def test_rollback_backs_up_current_and_restores_content(self, client, db): """核心安全: 回滚前自动备份当前内容为新版本; 回滚后内容=目标版本""" create_test_user(db) token = get_token_for_user(client) map_id = _create_published_map(client, token, "回滚安全C") # 发布后快照 v1.0（内容=模板初始） versions = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"] v1_id = versions[0]["id"] v1_content = client.get(f"/api/cma/maps/{map_id}/versions/{v1_id}", headers=auth_header(token)).json() # 修改地图内容为全新结构（published 状态下保存，不触发自动快照） new_dims = [{"key": "finance", "name": "财务层", "icon": "💰", "color": "#F56C6C", "objectives": [{"name": "回滚前最新目标", "icon": "target", "kpis": ["F_REVENUE"], "krs": [ {"name": "KR新", "kpi_code": "F_REVENUE", "weight": 100, "target_value": "100"} ]}]}] upd = client.put(f"/api/cma/maps/{map_id}", headers=auth_header(token), json={"dimensions": new_dims, "canvas_data": {"connections": [{"from": "a", "to": "b"}]}}) assert upd.status_code == 200 # 回滚到 v1.0 → 应自动备份当前(v1.0→v1.1) 且内容还原为 v1.0 rb = client.post(f"/api/cma/maps/{map_id}/versions/{v1_id}/rollback", headers=auth_header(token)) assert rb.status_code == 200 data = rb.json() assert data["version"] == "v1.0" assert data["backup_version"] == "v1.1" # 列表: 备份(v1.1 最新) + v1.0 after = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"] assert len(after) == 2 assert {x["version"] for x in after} == {"v1.0", "v1.1"} backup = next(x for x in after if x["version"] == "v1.1") assert backup["comment"] == "回滚前自动备份" # 当前地图内容 == v1.0 快照内容（语义还原；列表读路径会经 map_objectives 合并补 description/icon 默认键，按业务字段归一化比较） maps = client.get("/api/cma/maps", headers=auth_header(token)).json()["data"] cur = next(x for x in maps if x["id"] == map_id) assert cur["version"] == "v1.0" def norm(dims): return [ { "key": d.get("key"), "objectives": [ {k: o.get(k) for k in ("name", "kpis", "krs")} for o in d.get("objectives", []) ], } for d in dims ] assert norm(cur["dimensions"]) == norm(v1_content["dimensions"]) assert cur["canvas_data"] == v1_content["canvas_data"] def test_rollback_again_uses_next_free_backup_version(self, client, db): """连续两次回滚同一目标: 备份版本号不撞号（v1.1 → v1.2）""" create_test_user(db) token = get_token_for_user(client) map_id = _create_published_map(client, token, "回滚两次D") versions = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"] v1_id = versions[0]["id"] client.put(f"/api/cma/maps/{map_id}", headers=auth_header(token), json={"dimensions": [{"key": "finance", "name": "财务层", "icon": "💰", "color": "#F56C6C", "objectives": []}]}) rb1 = client.post(f"/api/cma/maps/{map_id}/versions/{v1_id}/rollback", headers=auth_header(token)).json() assert rb1["backup_version"] == "v1.1" # 再次改内容再回滚 → 备份 v1.2（不撞 v1.1） client.put(f"/api/cma/maps/{map_id}", headers=auth_header(token), json={"dimensions": [{"key": "customer", "name": "客户层", "icon": "👥", "color": "#409EFF", "objectives": []}]}) rb2 = client.post(f"/api/cma/maps/{map_id}/versions/{v1_id}/rollback", headers=auth_header(token)).json() assert rb2["backup_version"] == "v1.2" def test_update_comment(self, client, db): create_test_user(db) token = get_token_for_user(client) map_id = _create_published_map(client, token, "备注编辑E") versions = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"] vid = versions[0]["id"] resp = client.put(f"/api/cma/maps/{map_id}/versions/{vid}", headers=auth_header(token), json={"comment": "2026H2 定稿版"}) assert resp.status_code == 200 assert resp.json()["comment"] == "2026H2 定稿版" rows = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"] assert rows[0]["comment"] == "2026H2 定稿版" def test_update_comment_requires_field(self, client, db): create_test_user(db) token = get_token_for_user(client) map_id = _create_published_map(client, token, "备注必填F") versions = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token)).json()["data"] vid = versions[0]["id"] resp = client.put(f"/api/cma/maps/{map_id}/versions/{vid}", headers=auth_header(token), json={}) assert resp.status_code == 400 def test_cross_entity_all_versions_endpoints_404(self, client, db): """跨账套: 列表/详情/备注/回滚 全部 404（地图归属+版本 entity 双重隔离）""" create_test_user(db) token1 = get_token_for_user(client) map_id = _create_published_map(client, token1, "隔离G") user = db.query(User).filter(User.username == "testadmin").first() _create_second_entity(db, user) token2 = _token_for_entity(client, 2) versions = client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token1)).json()["data"] vid = versions[0]["id"] assert client.get(f"/api/cma/maps/{map_id}/versions", headers=auth_header(token2)).status_code == 404 assert client.get(f"/api/cma/maps/{map_id}/versions/{vid}", headers=auth_header(token2)).status_code == 404 assert client.put(f"/api/cma/maps/{map_id}/versions/{vid}", headers=auth_header(token2), json={"comment": "越权"}).status_code == 404 assert client.post(f"/api/cma/maps/{map_id}/versions/{vid}/rollback", headers=auth_header(token2)).status_code == 404 def test_rollback_unknown_version_404(self, client, db): create_test_user(db) token = get_token_for_user(client) map_id = _create_published_map(client, token, "不存在H") resp = client.post(f"/api/cma/maps/{map_id}/versions/99999/rollback", headers=auth_header(token)) assert resp.status_code == 404 def test_rollback_other_maps_version_404(self, client, db): """vid 属于另一张地图 → 404（map_id + vid 双条件匹配）""" create_test_user(db) token = get_token_for_user(client) map_a = _create_published_map(client, token, "地图A-I") map_b = _create_published_map(client, token, "地图B-I") va = client.get(f"/api/cma/maps/{map_a}/versions", headers=auth_header(token)).json()["data"][0] resp = client.post(f"/api/cma/maps/{map_b}/versions/{va['id']}/rollback", headers=auth_header(token)) assert resp.status_code == 404