diff --git a/backend/app/api/objectives.py b/backend/app/api/objectives.py index f69d0324..d59b9b64 100644 --- a/backend/app/api/objectives.py +++ b/backend/app/api/objectives.py @@ -13,7 +13,10 @@ router = APIRouter(prefix="/api/cma/maps", tags=["战略地图目标"], @router.get("/{map_id}/objectives") def list_objectives(map_id: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): - """获取某地图下的所有目标""" + """获取某地图下的所有目标（账套隔离: 跨企业返回空, OpenCode执行层发现）""" + m = db.query(StrategicMap).filter(StrategicMap.id == map_id, StrategicMap.entity_id == entity_id).first() + if not m: + return {"data": []} objs = db.query(MapObjective).filter( MapObjective.map_id == map_id ).order_by(MapObjective.sort_order).all() diff --git a/backend/tests/test_maps.py b/backend/tests/test_maps.py index 4886f9ec..6b6ec78d 100644 --- a/backend/tests/test_maps.py +++ b/backend/tests/test_maps.py @@ -4,9 +4,30 @@ import pytest from fastapi.testclient import TestClient from sqlalchemy.orm import Session +from app.models import Entity, User, UserEntity from tests.conftest import create_test_user, get_token_for_user, auth_header +def _create_second_entity(db: Session, user: User) -> int: + """创建第二个企业账套并授权测试用户，返回 entity_id""" + 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: + """以指定账套登录获取token""" + 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") + + class TestMaps: """战略地图CRUD测试""" @@ -124,3 +145,142 @@ class TestMaps: assert resp.status_code == 200 # 同维度连线现在被允许了，不再是旧的拒绝逻辑 # assert "不能" in resp.json()["detail"] + + def test_create_objective(self, client: TestClient, db: Session): + """在地图上新增目标并返回目标对象""" + create_test_user(db) + token = get_token_for_user(client) + + create_resp = client.post( + "/api/cma/maps/create-with-template", + headers=auth_header(token), + json={"title": "新增目标测试"}, + ) + map_id = create_resp.json()["id"] + + resp = client.post( + f"/api/cma/maps/{map_id}/objectives", + headers=auth_header(token), + json={"dimension_key": "finance", "name": "新增营收目标", "icon": "target"}, + ) + assert resp.status_code == 200 + data = resp.json() + assert data["name"] == "新增营收目标" + assert data["dimension_key"] == "finance" + assert data["map_id"] == map_id + assert data["entity_id"] == create_resp.json()["entity_id"] + assert "id" in data + + def test_create_objective_wrong_entity_404(self, client: TestClient, db: Session): + """跨账套新增目标返回404（地图不属于当前企业）""" + create_test_user(db) + token1 = get_token_for_user(client) + + create_resp = client.post( + "/api/cma/maps/create-with-template", + headers=auth_header(token1), + json={"title": "账户A地图"}, + ) + map_id = create_resp.json()["id"] + + # 第二个企业账套 + user = db.query(User).filter(User.username == "testadmin").first() + _create_second_entity(db, user) + token2 = _token_for_entity(client, 2) + + resp = client.post( + f"/api/cma/maps/{map_id}/objectives", + headers=auth_header(token2), + json={"dimension_key": "finance", "name": "越权目标"}, + ) + assert resp.status_code == 404 + + def test_create_snapshot_inherits_entity(self, client: TestClient, db: Session): + """手动快照创建版本并继承地图企业""" + create_test_user(db) + token1 = get_token_for_user(client) + + create_resp = client.post( + "/api/cma/maps/create-with-template", + headers=auth_header(token1), + json={"title": "快照继承测试"}, + ) + map_id = create_resp.json()["id"] + map_entity = create_resp.json()["entity_id"] + + snap_resp = client.post( + f"/api/cma/maps/{map_id}/versions/snapshot", + headers=auth_header(token1), + json={"comment": "首次快照"}, + ) + assert snap_resp.status_code == 200 + snap = snap_resp.json() + assert snap["map_id"] == map_id + assert snap["version"] == "v1.0" + assert snap["entity_id"] == map_entity + + def test_update_objective_rejects_entity_id_injection(self, client: TestClient, db: Session): + """更新目标时注入entity_id被忽略，企业归属保持不变""" + create_test_user(db) + token = get_token_for_user(client) + + create_resp = client.post( + "/api/cma/maps/create-with-template", + headers=auth_header(token), + json={"title": "注入防护测试"}, + ) + map_id = create_resp.json()["id"] + orig_entity = create_resp.json()["entity_id"] + + obj_resp = client.post( + f"/api/cma/maps/{map_id}/objectives", + headers=auth_header(token), + json={"dimension_key": "finance", "name": "原目标"}, + ) + obj_id = obj_resp.json()["id"] + + update_resp = client.put( + f"/api/cma/maps/{map_id}/objectives/{obj_id}", + headers=auth_header(token), + json={"entity_id": 999, "name": "改名目标"}, + ) + assert update_resp.status_code == 200 + data = update_resp.json() + # entity_id 不在白名单字段中，注入被忽略 → 归属不变 + assert data["entity_id"] == orig_entity + assert data["entity_id"] != 999 + assert data["name"] == "改名目标" + + def test_cross_entity_list_objectives_empty(self, client: TestClient, db: Session): + """跨账套查询目标列表返回空（目标不属于当前企业）""" + create_test_user(db) + token1 = get_token_for_user(client) + + # 企业1创建空白地图（无任何目标） + create_resp = client.post( + "/api/cma/maps", + headers=auth_header(token1), + json={"title": "企业1地图"}, + ) + assert create_resp.status_code == 200 + map_id = create_resp.json()["id"] + + # 企业1自己能看到空列表 + own_resp = client.get( + f"/api/cma/maps/{map_id}/objectives", + headers=auth_header(token1), + ) + assert own_resp.status_code == 200 + assert own_resp.json()["data"] == [] + + # 第二个企业账套 → 跨账套查询同样返回空 + user = db.query(User).filter(User.username == "testadmin").first() + _create_second_entity(db, user) + token2 = _token_for_entity(client, 2) + + cross_resp = client.get( + f"/api/cma/maps/{map_id}/objectives", + headers=auth_header(token2), + ) + assert cross_resp.status_code == 200 + assert cross_resp.json()["data"] == []