Files
zlwuandCursor f5ab023ab5 feat(auth): 扩展数据权限核心能力并接入系统管理辅助接口
为 apply_data_scope 增加 owner_field 与行级校验,补充 DbUserContext.to_user_info() 及单元测试,并在用户/角色/字典下拉与用户详情接口生效。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-28 20:30:36 +08:00

129 lines
4.6 KiB
Python

"""数据权限 apply_data_scope / row_in_data_scope 单元测试。"""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
from fastapi import HTTPException
from sqlalchemy import true
from app.core.data_scope import (
apply_data_scope,
assert_row_in_data_scope,
row_in_data_scope,
)
class _FakeModel:
create_dept_id = MagicMock()
create_by = MagicMock()
owner_user_id = MagicMock()
def test_super_admin_bypass_apply():
query = MagicMock()
result = apply_data_scope(query, _FakeModel, {"role": "super_admin"}, MagicMock(), "shanxia2.0")
assert result is query
query.filter.assert_not_called()
def test_scope4_invokes_filter_via_scope_collection():
query = MagicMock()
user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"}
with patch(
"app.core.data_scope._resolve_user_role_scopes",
return_value=[{"data_scope": "4", "details": []}],
), patch(
"app.core.data_scope._collect_scope_ids",
return_value=(set(), {"user-1"}),
), patch("app.core.data_scope._append_user_scope_conditions") as append_user:
def _fake_append(conditions, model, user_ids, *, has_user, owner_field):
conditions.append(true())
append_user.side_effect = _fake_append
apply_data_scope(query, _FakeModel, user_info, MagicMock(), "shanxia2.0")
assert query.filter.called
def test_scope4_with_owner_field_passes_owner_to_helper():
query = MagicMock()
user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"}
with patch(
"app.core.data_scope._resolve_user_role_scopes",
return_value=[{"data_scope": "4", "details": []}],
), patch(
"app.core.data_scope._collect_scope_ids",
return_value=(set(), {"user-1"}),
), patch("app.core.data_scope._append_user_scope_conditions") as append_user:
apply_data_scope(
query,
_FakeModel,
user_info,
MagicMock(),
"shanxia2.0",
owner_field="owner_user_id",
)
append_user.assert_called_once()
assert append_user.call_args.kwargs["owner_field"] == "owner_user_id"
def test_no_role_scopes_returns_empty_query():
query = MagicMock()
user_info = {"user_id": "user-1", "role": "admin"}
with patch("app.core.data_scope._resolve_user_role_scopes", return_value=[]):
apply_data_scope(query, _FakeModel, user_info, MagicMock(), "shanxia2.0")
query.filter.assert_called()
def test_row_in_data_scope_super_admin():
row = SimpleNamespace(create_dept_id="x", create_by="y")
assert row_in_data_scope(MagicMock(), row, {"role": "super_admin"}, "shanxia2.0") is True
def test_row_in_data_scope_self_by_create_by():
row = SimpleNamespace(create_dept_id="", create_by="user-1")
user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"}
with patch(
"app.core.data_scope._resolve_user_role_scopes",
return_value=[{"data_scope": "4", "details": []}],
):
assert row_in_data_scope(MagicMock(), row, user_info, "shanxia2.0") is True
def test_row_in_data_scope_self_by_owner_field():
row = SimpleNamespace(create_dept_id="", create_by="other", owner_user_id="user-1")
user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"}
with patch(
"app.core.data_scope._resolve_user_role_scopes",
return_value=[{"data_scope": "4", "details": []}],
):
assert (
row_in_data_scope(
MagicMock(), row, user_info, "shanxia2.0", owner_field="owner_user_id"
)
is True
)
def test_row_in_data_scope_denied():
row = SimpleNamespace(create_dept_id="dept-z", create_by="other")
user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"}
with patch(
"app.core.data_scope._resolve_user_role_scopes",
return_value=[{"data_scope": "4", "details": []}],
):
assert row_in_data_scope(MagicMock(), row, user_info, "shanxia2.0") is False
def test_assert_row_in_data_scope_raises_403():
row = SimpleNamespace(create_dept_id="dept-z", create_by="other")
user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"}
with patch(
"app.core.data_scope._resolve_user_role_scopes",
return_value=[{"data_scope": "4", "details": []}],
):
with pytest.raises(HTTPException) as exc:
assert_row_in_data_scope(MagicMock(), row, _FakeModel, user_info, "shanxia2.0")
assert exc.value.status_code == 403