Hermes CI Fix
d08a8490eb
ci/woodpecker/push/main Pipeline failed
缺陷 cma-strategy-map-autosave-stale-overwrite 的护栏（PUT /api/cma/maps/{id} 覆写
dimensions/canvas_data 须带 base_updated_at，否则 409 fail-closed）改变了同一写通道的
契约 ⇒ 全仓清点调用点，在调用点取基准（写前读 updated_at），**不放宽护栏**：
- tests/test_map_versions.py: 新增 _base_updated_at(db, map_id) 助手；3 处覆写 dimensions 的
PUT 补基准（test_rollback_backs_up_current_and_restores_content 会直接失败；
test_rollback_again_uses_next_free_backup_version 两处此前已静默降级——PUT 被拒但用例仍绿，
即「改了内容再回滚」的前提实际未发生）。
- tests/test_kr_name_pollution_20260916.py: 2 处 KR 透传护栏用例补基准。
改前/改后全量对拍（同树 pythonpath）：
baseline 22 failed / fixed2 18 failed；diff 仅剩 tests/test_doc_center_20260924.py::
TestProdContentCanary::test_prod_content_readable —— 基线树同样失败（/srv/hermes/shared/
cma-content 于 14:05 被写入 manual 分类，共享内容目录变化），与本单无关。
本单新增的 5 例在基线树全红、改后全绿。
任务: yanxue-20261012-cma-map-autosave-stale-overwrite-defect-fullstack
235 lines
12 KiB
Python
235 lines
12 KiB
Python
"""
|
||
战略地图版本管理完整实现测试（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, TEST_USER_PASSWORD
|
||
|
||
|
||
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": TEST_USER_PASSWORD, "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
|
||
|
||
|
||
def _base_updated_at(db: Session, map_id: int) -> str:
|
||
"""客户端载入快照时取得的服务端基准（乐观并发契约 base_updated_at，2026-10-12）。
|
||
|
||
PUT /api/cma/maps/{id} 覆写 dimensions/canvas_data 时须带此基准，取「写前读到的
|
||
updated_at」——这是同一写通道契约的一部分（cma-strategy-map-autosave-stale-overwrite），
|
||
不是把用例改绿：不带基准的写会被 409 拒绝（fail-closed）。
|
||
"""
|
||
db.expire_all()
|
||
m = db.query(StrategicMap).filter(StrategicMap.id == map_id).first()
|
||
assert m is not None
|
||
return m.updated_at.isoformat()
|
||
|
||
|
||
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"}]},
|
||
"base_updated_at": _base_updated_at(db, map_id)})
|
||
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": []}],
|
||
"base_updated_at": _base_updated_at(db, map_id)})
|
||
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": []}],
|
||
"base_updated_at": _base_updated_at(db, map_id)})
|
||
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
|