为 apply_data_scope 增加 owner_field 与行级校验,补充 DbUserContext.to_user_info() 及单元测试,并在用户/角色/字典下拉与用户详情接口生效。 Co-authored-by: Cursor <cursoragent@cursor.com>
129 lines
4.6 KiB
Python
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
|