test_data_scope.py 4.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128
  1. """数据权限 apply_data_scope / row_in_data_scope 单元测试。"""
  2. from __future__ import annotations
  3. from types import SimpleNamespace
  4. from unittest.mock import MagicMock, patch
  5. import pytest
  6. from fastapi import HTTPException
  7. from sqlalchemy import true
  8. from app.core.data_scope import (
  9. apply_data_scope,
  10. assert_row_in_data_scope,
  11. row_in_data_scope,
  12. )
  13. class _FakeModel:
  14. create_dept_id = MagicMock()
  15. create_by = MagicMock()
  16. owner_user_id = MagicMock()
  17. def test_super_admin_bypass_apply():
  18. query = MagicMock()
  19. result = apply_data_scope(query, _FakeModel, {"role": "super_admin"}, MagicMock(), "shanxia2.0")
  20. assert result is query
  21. query.filter.assert_not_called()
  22. def test_scope4_invokes_filter_via_scope_collection():
  23. query = MagicMock()
  24. user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"}
  25. with patch(
  26. "app.core.data_scope._resolve_user_role_scopes",
  27. return_value=[{"data_scope": "4", "details": []}],
  28. ), patch(
  29. "app.core.data_scope._collect_scope_ids",
  30. return_value=(set(), {"user-1"}),
  31. ), patch("app.core.data_scope._append_user_scope_conditions") as append_user:
  32. def _fake_append(conditions, model, user_ids, *, has_user, owner_field):
  33. conditions.append(true())
  34. append_user.side_effect = _fake_append
  35. apply_data_scope(query, _FakeModel, user_info, MagicMock(), "shanxia2.0")
  36. assert query.filter.called
  37. def test_scope4_with_owner_field_passes_owner_to_helper():
  38. query = MagicMock()
  39. user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"}
  40. with patch(
  41. "app.core.data_scope._resolve_user_role_scopes",
  42. return_value=[{"data_scope": "4", "details": []}],
  43. ), patch(
  44. "app.core.data_scope._collect_scope_ids",
  45. return_value=(set(), {"user-1"}),
  46. ), patch("app.core.data_scope._append_user_scope_conditions") as append_user:
  47. apply_data_scope(
  48. query,
  49. _FakeModel,
  50. user_info,
  51. MagicMock(),
  52. "shanxia2.0",
  53. owner_field="owner_user_id",
  54. )
  55. append_user.assert_called_once()
  56. assert append_user.call_args.kwargs["owner_field"] == "owner_user_id"
  57. def test_no_role_scopes_returns_empty_query():
  58. query = MagicMock()
  59. user_info = {"user_id": "user-1", "role": "admin"}
  60. with patch("app.core.data_scope._resolve_user_role_scopes", return_value=[]):
  61. apply_data_scope(query, _FakeModel, user_info, MagicMock(), "shanxia2.0")
  62. query.filter.assert_called()
  63. def test_row_in_data_scope_super_admin():
  64. row = SimpleNamespace(create_dept_id="x", create_by="y")
  65. assert row_in_data_scope(MagicMock(), row, {"role": "super_admin"}, "shanxia2.0") is True
  66. def test_row_in_data_scope_self_by_create_by():
  67. row = SimpleNamespace(create_dept_id="", create_by="user-1")
  68. user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"}
  69. with patch(
  70. "app.core.data_scope._resolve_user_role_scopes",
  71. return_value=[{"data_scope": "4", "details": []}],
  72. ):
  73. assert row_in_data_scope(MagicMock(), row, user_info, "shanxia2.0") is True
  74. def test_row_in_data_scope_self_by_owner_field():
  75. row = SimpleNamespace(create_dept_id="", create_by="other", owner_user_id="user-1")
  76. user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"}
  77. with patch(
  78. "app.core.data_scope._resolve_user_role_scopes",
  79. return_value=[{"data_scope": "4", "details": []}],
  80. ):
  81. assert (
  82. row_in_data_scope(
  83. MagicMock(), row, user_info, "shanxia2.0", owner_field="owner_user_id"
  84. )
  85. is True
  86. )
  87. def test_row_in_data_scope_denied():
  88. row = SimpleNamespace(create_dept_id="dept-z", create_by="other")
  89. user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"}
  90. with patch(
  91. "app.core.data_scope._resolve_user_role_scopes",
  92. return_value=[{"data_scope": "4", "details": []}],
  93. ):
  94. assert row_in_data_scope(MagicMock(), row, user_info, "shanxia2.0") is False
  95. def test_assert_row_in_data_scope_raises_403():
  96. row = SimpleNamespace(create_dept_id="dept-z", create_by="other")
  97. user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"}
  98. with patch(
  99. "app.core.data_scope._resolve_user_role_scopes",
  100. return_value=[{"data_scope": "4", "details": []}],
  101. ):
  102. with pytest.raises(HTTPException) as exc:
  103. assert_row_in_data_scope(MagicMock(), row, _FakeModel, user_info, "shanxia2.0")
  104. assert exc.value.status_code == 403