116 lines
4.0 KiB
Python
116 lines
4.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
T08 / F11 权限底座自测脚本（离线 SQL 语义自测 + 叠加规则验证）
|
||
|
||
运行方式：
|
||
cd repos/hr-system
|
||
python3 scripts/test_permission.py
|
||
|
||
说明：不依赖真实 MariaDB 连接也可跑通纯逻辑部分（叠加规则、子树展开）；
|
||
若配置了 hrs 库则可执行 init 幂等性验证（需 appbase/sql 环境）。
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
|
||
def _json_loads(raw, default):
|
||
if not raw:
|
||
return default
|
||
try:
|
||
return json.loads(raw)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def expand_org_tree(parent_map, org_ids):
|
||
"""纯逻辑子树展开（模拟 org_unit.parent_id 自引用）。"""
|
||
if not org_ids:
|
||
return []
|
||
result = set(org_ids)
|
||
frontier = list(org_ids)
|
||
seen = set(org_ids)
|
||
while frontier:
|
||
children = []
|
||
for oid in frontier:
|
||
for child in parent_map.get(oid, []):
|
||
if child not in seen:
|
||
seen.add(child)
|
||
result.add(child)
|
||
children.append(child)
|
||
frontier = children
|
||
return sorted(result)
|
||
|
||
|
||
def test_org_expand():
|
||
# 组织树：A -> B -> C
|
||
parent_map = {"A": ["B"], "B": ["C"]}
|
||
out = expand_org_tree(parent_map, ["A"])
|
||
assert out == ["A", "B", "C"], out
|
||
return "PASS: 组织维度子树展开"
|
||
|
||
|
||
def test_scope_overlay():
|
||
"""模拟 get_data_scope 叠加规则（双维度 + 多角色 + admin 全量）。"""
|
||
# 两个 scope：org 并集；field 并集；admin 全量
|
||
scopes = [
|
||
{"scope_type": "org", "org_ids": ["A", "B"], "field_ids": []},
|
||
{"scope_type": "field", "org_ids": [], "field_ids": ["f1", "f2"]},
|
||
]
|
||
org_union, field_union, field_restricted = set(), set(), False
|
||
for s in scopes:
|
||
if s["scope_type"] in ("org", "all") and s["org_ids"]:
|
||
org_union.update(s["org_ids"])
|
||
if s["scope_type"] in ("field", "all"):
|
||
field_restricted = True
|
||
field_union.update(s["field_ids"])
|
||
assert org_union == {"A", "B"}
|
||
assert field_union == {"f1", "f2"}
|
||
assert field_restricted is True
|
||
return "PASS: 双维度叠加并集"
|
||
|
||
|
||
def test_field_all_when_unrestricted():
|
||
scopes = [{"scope_type": "org", "org_ids": ["A"], "field_ids": []}]
|
||
field_restricted = False
|
||
field_union = set()
|
||
for s in scopes:
|
||
if s["scope_type"] in ("field", "all"):
|
||
field_restricted = True
|
||
field_union.update(s["field_ids"])
|
||
assert field_restricted is False # -> field_all = True
|
||
return "PASS: 未配置字段维度 -> 字段不受限"
|
||
|
||
|
||
def test_admin_full():
|
||
role_codes = ["manager", "sys_admin"]
|
||
assert ("sys_admin" in role_codes or "admin" in role_codes) is True
|
||
return "PASS: sys_admin/admin 全量数据范围"
|
||
|
||
|
||
def main():
|
||
tests = [test_org_expand, test_scope_overlay, test_field_all_when_unrestricted, test_admin_full]
|
||
passed = 0
|
||
for t in tests:
|
||
try:
|
||
print(t())
|
||
passed += 1
|
||
except AssertionError as e:
|
||
print("FAIL:", t.__name__, e)
|
||
print("\n权限矩阵自测结果：{}/{} 通过".format(passed, len(tests)))
|
||
print("""
|
||
权限矩阵（自测基线）：
|
||
| 角色 | 路径级功能权限(rbac) | 组织数据范围 | 花名册字段范围 |
|
||
|--------------|---------------------------|--------------------------|------------------------|
|
||
| 员工自助 | 个人档案/自助申请 | 仅本人(org_ids=空) | 本人可见字段白名单 |
|
||
| 经理 | 本部门入转调离管理+审批 | 本部门+子树(并集展开) | 经理视角字段白名单 |
|
||
| HR管理员 | 全公司人事业务 | 按 sys_data_scope 并集 | 按 sys_data_scope 并集 |
|
||
| 系统管理员 | 系统配置+权限+审计 | org_all=True | field_all=True |
|
||
| admin | 全量(等价 sys_admin) | org_all=True | field_all=True |
|
||
""")
|
||
return 0 if passed == len(tests) else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|