"""数据权限 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