from __future__ import annotations import unittest from datetime import datetime from sqlalchemy import BigInteger, create_engine, select from sqlalchemy.ext.compiler import compiles from sqlalchemy.orm import Session, sessionmaker @compiles(BigInteger, "sqlite") def _compile_bigint_for_sqlite(type_, compiler, **kw) -> str: _ = type_, compiler, kw return "INTEGER" import app.models.master_data # noqa: E402,F401 import app.models.miniapp # noqa: E402,F401 import app.models.operations # noqa: E402,F401 import app.models.org # noqa: E402,F401 import app.models.planning # noqa: E402,F401 import app.models.sales # noqa: E402,F401 from app.models.base import Base # noqa: E402 from app.models.org import Department, Employee, Permission, Role, RolePermission, User, UserRole # noqa: E402 class SystemPermissionManagementTest(unittest.TestCase): def setUp(self) -> None: engine = create_engine("sqlite+pysqlite:///:memory:", future=True) Base.metadata.create_all(engine) self.SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True) self.db: Session = self.SessionLocal() now = datetime(2026, 6, 8, 9, 0, 0) self.company = Department( dept_code="ORG_ROOT", dept_name="总公司", parent_id=None, org_node_type="COMPANY", dept_type="ADMIN", manager_name="王总", manager_employee_id=None, status="ACTIVE", sort_no=1, remark="总公司节点", created_at=now, updated_at=now, ) self.db.add(self.company) self.db.flush() def tearDown(self) -> None: self.db.close() def test_department_model_exposes_tree_and_manager_fields(self) -> None: row = self.db.scalar(select(Department).where(Department.dept_code == "ORG_ROOT")) self.assertIsNotNone(row) self.assertEqual(row.parent_id, None) self.assertEqual(row.org_node_type, "COMPANY") self.assertEqual(row.manager_name, "王总") self.assertEqual(row.manager_employee_id, None) self.assertEqual(row.remark, "总公司节点") def _seed_role_permission_user_data(self) -> None: now = datetime(2026, 6, 8, 9, 0, 0) purchase = Department( dept_code="PURCHASE", dept_name="采购部", parent_id=self.company.id, org_node_type="DEPARTMENT", dept_type="PURCHASE", manager_name="采购主管", manager_employee_id=None, status="ACTIVE", sort_no=10, remark=None, created_at=now, updated_at=now, ) self.db.add(purchase) self.db.flush() manager = Employee( employee_code="EMP-001", employee_name="张三", dept_id=purchase.id, mobile="13800000011", gender=None, hire_date=None, job_title="采购负责人", shift_code=None, manager_employee_id=None, is_operator=0, is_workshop_staff=0, status="ACTIVE", remark="来自人员主数据", created_at=now, updated_at=now, ) buyer = Employee( employee_code="EMP-002", employee_name="李四", dept_id=purchase.id, mobile="13800000012", gender=None, hire_date=None, job_title="采购专员", shift_code=None, manager_employee_id=None, is_operator=0, is_workshop_staff=0, status="ACTIVE", remark=None, created_at=now, updated_at=now, ) role = Role( role_code="PURCHASER", role_name="采购专员", role_scope="SYSTEM", status="ACTIVE", remark=None, created_at=now, updated_at=now, ) permission = Permission( permission_code="MENU_PURCHASE_ORDER", permission_name="采购订单菜单", module_code="PURCHASE_ORDER", action_code="VIEW", status="ACTIVE", created_at=now, updated_at=now, ) self.db.add_all([manager, buyer, role, permission]) self.db.flush() purchase.manager_employee_id = manager.id purchase.manager_name = manager.employee_name self.db.add(purchase) self.db.commit() self.purchase_dept_id = purchase.id self.manager_employee_id = manager.id self.buyer_employee_id = buyer.id self.purchaser_role_id = role.id self.purchase_permission_id = permission.id def test_build_org_tree_returns_managed_root_and_ignores_legacy_root_departments(self) -> None: from app.services.system_permissions import build_org_tree self._seed_role_permission_user_data() now = datetime(2026, 6, 8, 9, 0, 0) legacy = Department( dept_code="LEGACY-SALES", dept_name="旧销售部", parent_id=None, org_node_type="DEPARTMENT", dept_type="SALES", manager_name="旧主管", manager_employee_id=None, status="ACTIVE", sort_no=99, remark="旧部门不直接作为脑图根节点展示", created_at=now, updated_at=now, ) self.db.add(legacy) self.db.commit() tree = build_org_tree(self.db) self.assertEqual(len(tree), 1) self.assertEqual(tree[0].node_label, "总公司") self.assertEqual(tree[0].node_type, "COMPANY") purchase = tree[0].children[0] self.assertEqual(purchase.node_label, "采购部") self.assertEqual(purchase.manager_employee_id, self.manager_employee_id) self.assertEqual(purchase.manager_name, "张三") self.assertEqual(purchase.employees, []) def test_org_employee_leaves_are_explicit_bindings_not_department_membership(self) -> None: from app.schemas.system_permissions import OrgEmployeeBind from app.services.system_permissions import bind_org_employee, build_org_tree, unbind_org_employee self._seed_role_permission_user_data() tree_before = build_org_tree(self.db) purchase_before = tree_before[0].children[0] self.assertEqual(purchase_before.employees, []) bind_org_employee( self.db, self.purchase_dept_id, OrgEmployeeBind(employee_id=self.buyer_employee_id, sort_no=1), ) tree_after = build_org_tree(self.db) purchase_after = tree_after[0].children[0] self.assertEqual([employee.employee_name for employee in purchase_after.employees], ["李四"]) self.assertEqual(purchase_after.employees[0].mobile, "13800000012") unbind_org_employee(self.db, self.purchase_dept_id, self.buyer_employee_id) tree_unbound = build_org_tree(self.db) self.assertEqual(tree_unbound[0].children[0].employees, []) def test_update_org_node_supports_multiple_managers(self) -> None: from app.schemas.system_permissions import OrgNodeUpdate from app.services.system_permissions import build_org_tree, update_org_node self._seed_role_permission_user_data() update_org_node( self.db, self.purchase_dept_id, OrgNodeUpdate( node_label="采购部", dept_type="PURCHASE", manager_employee_ids=[self.manager_employee_id, self.buyer_employee_id], sort_no=10, remark=None, status="ACTIVE", ), ) tree = build_org_tree(self.db) purchase = tree[0].children[0] self.assertEqual(purchase.manager_employee_ids, [self.manager_employee_id, self.buyer_employee_id]) self.assertEqual(purchase.manager_names, ["张三", "李四"]) self.assertEqual(purchase.manager_employee_id, self.manager_employee_id) self.assertEqual(purchase.manager_name, "张三、李四") def test_build_org_tree_falls_back_to_legacy_single_manager(self) -> None: from app.services.system_permissions import build_org_tree self._seed_role_permission_user_data() tree = build_org_tree(self.db) purchase = tree[0].children[0] self.assertEqual(purchase.manager_employee_ids, [self.manager_employee_id]) self.assertEqual(purchase.manager_names, ["张三"]) def test_create_system_user_uses_employee_phone_as_account_and_hashes_password(self) -> None: from app.schemas.system_permissions import SystemUserCreate from app.services.auth import verify_password from app.services.system_permissions import create_system_user self._seed_role_permission_user_data() result = create_system_user( self.db, SystemUserCreate( employee_id=self.buyer_employee_id, role_ids=[self.purchaser_role_id], password="secret123", status="ACTIVE", ), ) self.assertEqual(result.username, "13800000012") self.assertEqual(result.employee_name, "李四") self.assertEqual(result.role_names, ["采购专员"]) user = self.db.scalar(select(User).where(User.username == "13800000012")) self.assertIsNotNone(user) self.assertTrue(verify_password("secret123", user.password_hash)) links = self.db.scalars(select(UserRole).where(UserRole.user_id == user.id)).all() self.assertEqual(len(links), 1) self.assertEqual(links[0].role_id, self.purchaser_role_id) def test_create_system_user_rejects_employee_without_phone(self) -> None: from fastapi import HTTPException from app.schemas.system_permissions import SystemUserCreate from app.services.system_permissions import create_system_user self._seed_role_permission_user_data() now = datetime(2026, 6, 8, 9, 0, 0) employee = Employee( employee_code="EMP-NOPHONE", employee_name="无手机号人员", dept_id=self.purchase_dept_id, mobile=None, gender=None, hire_date=None, job_title=None, shift_code=None, manager_employee_id=None, is_operator=0, is_workshop_staff=0, status="ACTIVE", remark=None, created_at=now, updated_at=now, ) self.db.add(employee) self.db.commit() with self.assertRaises(HTTPException) as ctx: create_system_user( self.db, SystemUserCreate( employee_id=employee.id, role_ids=[self.purchaser_role_id], password="secret123", status="ACTIVE", ), ) self.assertEqual(ctx.exception.status_code, 400) self.assertIn("人员没有电话号码", str(ctx.exception.detail)) def test_create_role_invalid_permission_does_not_leave_pending_role(self) -> None: from fastapi import HTTPException from app.schemas.system_permissions import RoleCreate from app.services.system_permissions import create_role with self.assertRaises(HTTPException): create_role( self.db, RoleCreate( role_code="INVALID_PERMISSION_ROLE", role_name="无效权限角色", role_scope="SYSTEM", status="ACTIVE", remark=None, permission_codes=["MISSING_PERMISSION"], ), ) self.db.commit() row = self.db.scalar(select(Role).where(Role.role_code == "INVALID_PERMISSION_ROLE")) self.assertIsNone(row) def test_update_role_invalid_permission_does_not_persist_name_change(self) -> None: from fastapi import HTTPException from app.schemas.system_permissions import RoleUpdate from app.services.system_permissions import update_role self._seed_role_permission_user_data() with self.assertRaises(HTTPException): update_role( self.db, self.purchaser_role_id, RoleUpdate( role_name="错误改名", role_scope="SYSTEM", status="ACTIVE", remark="不应持久化", permission_codes=["MISSING_PERMISSION"], ), ) self.db.commit() row = self.db.get(Role, self.purchaser_role_id) self.assertIsNotNone(row) self.assertEqual(row.role_name, "采购专员") self.assertIsNone(row.remark) def test_create_system_user_invalid_role_does_not_leave_pending_user(self) -> None: from fastapi import HTTPException from app.schemas.system_permissions import SystemUserCreate from app.services.system_permissions import create_system_user self._seed_role_permission_user_data() with self.assertRaises(HTTPException): create_system_user( self.db, SystemUserCreate( employee_id=self.buyer_employee_id, role_ids=[999999], password="secret123", status="ACTIVE", ), ) self.db.commit() employee = self.db.get(Employee, self.buyer_employee_id) self.assertIsNotNone(employee) row = self.db.scalar( select(User).where( User.employee_id == self.buyer_employee_id, User.username == employee.mobile, ) ) self.assertIsNone(row) def test_create_org_node_adds_child_under_parent(self) -> None: from app.schemas.system_permissions import OrgNodeCreate from app.services.system_permissions import create_org_node, build_org_tree payload = OrgNodeCreate( parent_id=self.company.id, node_type="BRANCH", node_label="分公司A", dept_code="BR-A", dept_type="ADMIN", manager_employee_id=None, sort_no=1, remark="分公司节点", status="ACTIVE", ) create_org_node(self.db, payload) tree = build_org_tree(self.db) self.assertEqual(tree[0].children[0].node_label, "分公司A") self.assertEqual(tree[0].children[0].parent_id, self.company.id) self.assertEqual(tree[0].children[0].node_type, "BRANCH") def test_create_org_node_allows_company_direct_department(self) -> None: from app.schemas.system_permissions import OrgNodeCreate from app.services.system_permissions import create_org_node, build_org_tree direct_department = create_org_node( self.db, OrgNodeCreate( parent_id=self.company.id, node_type="DEPARTMENT", node_label="总公司直属部门", dept_code="DIRECT-DEPT", dept_type="ADMIN", manager_employee_id=None, sort_no=1, remark=None, status="ACTIVE", ), ) tree = build_org_tree(self.db) self.assertEqual(direct_department.parent_id, self.company.id) self.assertEqual(direct_department.node_type, "DEPARTMENT") self.assertTrue( any(child.node_label == "总公司直属部门" and child.node_type == "DEPARTMENT" for child in tree[0].children) ) def test_create_org_node_rejects_invalid_child_level(self) -> None: from fastapi import HTTPException from app.schemas.system_permissions import OrgNodeCreate from app.services.system_permissions import create_org_node with self.assertRaises(HTTPException) as ctx: create_org_node( self.db, OrgNodeCreate( parent_id=self.company.id, node_type="GROUP", node_label="错误小组", dept_code="BAD-GROUP", dept_type="ADMIN", manager_employee_id=None, sort_no=1, remark=None, status="ACTIVE", ), ) self.assertEqual(ctx.exception.status_code, 400) self.assertIn("总公司下只能新增分公司或直属部门", str(ctx.exception.detail)) def test_delete_org_node_rejects_nodes_with_children_or_bound_employees(self) -> None: from fastapi import HTTPException from app.schemas.system_permissions import OrgEmployeeBind, OrgNodeCreate from app.services.system_permissions import bind_org_employee, create_org_node, delete_org_node self._seed_role_permission_user_data() branch = create_org_node( self.db, OrgNodeCreate( parent_id=self.company.id, node_type="BRANCH", node_label="分公司A", dept_code="BR-A", dept_type="ADMIN", manager_employee_id=None, sort_no=1, remark=None, status="ACTIVE", ), ) with self.assertRaises(HTTPException) as child_ctx: delete_org_node(self.db, self.company.id) self.assertEqual(child_ctx.exception.status_code, 400) self.assertIn("存在下级节点", str(child_ctx.exception.detail)) department = create_org_node( self.db, OrgNodeCreate( parent_id=branch.node_id, node_type="DEPARTMENT", node_label="业务部门", dept_code="DEPT-A", dept_type="ADMIN", manager_employee_id=None, sort_no=1, remark=None, status="ACTIVE", ), ) bind_org_employee( self.db, department.node_id, OrgEmployeeBind(employee_id=self.buyer_employee_id, sort_no=1), ) with self.assertRaises(HTTPException) as employee_ctx: delete_org_node(self.db, department.node_id) self.assertEqual(employee_ctx.exception.status_code, 400) self.assertIn("存在人员节点", str(employee_ctx.exception.detail)) def test_admin_context_gets_system_permission_code(self) -> None: from app.services.auth import get_auth_context, hash_password self._seed_role_permission_user_data() now = datetime(2026, 6, 8, 9, 0, 0) admin_role = Role( role_code="ADMIN", role_name="超级管理员", role_scope="SYSTEM", status="ACTIVE", remark=None, created_at=now, updated_at=now, ) admin_permission = Permission( permission_code="MENU_SYSTEM_PERMISSION", permission_name="系统权限管理菜单", module_code="SYSTEM_PERMISSION", action_code="VIEW", status="ACTIVE", created_at=now, updated_at=now, ) admin_employee = Employee( employee_code="EMP-ADMIN", employee_name="系统管理员", dept_id=self.purchase_dept_id, mobile="13800000099", gender=None, hire_date=None, job_title=None, shift_code=None, manager_employee_id=None, is_operator=0, is_workshop_staff=0, status="ACTIVE", remark=None, created_at=now, updated_at=now, ) self.db.add_all([admin_role, admin_permission, admin_employee]) self.db.flush() admin_user = User( username="admin", password_hash=hash_password("secret123"), employee_id=admin_employee.id, dept_id=self.purchase_dept_id, nickname="系统管理员", email=None, is_super_admin=1, last_login_at=None, status="ACTIVE", created_at=now, updated_at=now, ) self.db.add(admin_user) self.db.flush() self.db.add(UserRole(user_id=admin_user.id, role_id=admin_role.id, created_at=now)) self.db.commit() context = get_auth_context(self.db, "admin") self.assertIn("MENU_SYSTEM_PERMISSION", context.permission_codes) def _seed_permission_employee_selector_data(self) -> dict[str, int]: from app.services.auth import hash_password self._seed_role_permission_user_data() now = datetime(2026, 6, 8, 9, 0, 0) sales_permission = Permission( permission_code="MENU_SALES_PLANNING", permission_name="客户名录/销售订单菜单", module_code="SALES_PLANNING", action_code="VIEW", status="ACTIVE", created_at=now, updated_at=now, ) sales_role = Role( role_code="SALES", role_name="销售人员", role_scope="SYSTEM", status="ACTIVE", remark=None, created_at=now, updated_at=now, ) sales_employee = Employee( employee_code="EMP-SALES", employee_name="赵销售", dept_id=self.purchase_dept_id, mobile="13800000021", gender=None, hire_date=None, job_title="销售人员", shift_code=None, manager_employee_id=None, is_operator=0, is_workshop_staff=0, status="ACTIVE", remark=None, created_at=now, updated_at=now, ) inactive_employee = Employee( employee_code="EMP-INACTIVE", employee_name="停用员工", dept_id=self.purchase_dept_id, mobile="13800000022", gender=None, hire_date=None, job_title="停用", shift_code=None, manager_employee_id=None, is_operator=0, is_workshop_staff=0, status="INACTIVE", remark=None, created_at=now, updated_at=now, ) self.db.add_all([sales_permission, sales_role, sales_employee, inactive_employee]) self.db.flush() self.db.add_all( [ RolePermission(role_id=self.purchaser_role_id, permission_id=self.purchase_permission_id, created_at=now), RolePermission(role_id=sales_role.id, permission_id=sales_permission.id, created_at=now), User( username="13800000012", password_hash=hash_password("secret123"), employee_id=self.buyer_employee_id, dept_id=self.purchase_dept_id, nickname="李四", email=None, is_super_admin=0, last_login_at=None, status="ACTIVE", created_at=now, updated_at=now, ), User( username="13800000021", password_hash=hash_password("secret123"), employee_id=sales_employee.id, dept_id=self.purchase_dept_id, nickname="赵销售", email=None, is_super_admin=0, last_login_at=None, status="ACTIVE", created_at=now, updated_at=now, ), User( username="13800000022", password_hash=hash_password("secret123"), employee_id=inactive_employee.id, dept_id=self.purchase_dept_id, nickname="停用员工", email=None, is_super_admin=0, last_login_at=None, status="ACTIVE", created_at=now, updated_at=now, ), ] ) self.db.flush() buyer_user = self.db.scalar(select(User).where(User.employee_id == self.buyer_employee_id)) sales_user = self.db.scalar(select(User).where(User.employee_id == sales_employee.id)) inactive_user = self.db.scalar(select(User).where(User.employee_id == inactive_employee.id)) self.db.add_all( [ UserRole(user_id=buyer_user.id, role_id=self.purchaser_role_id, created_at=now), UserRole(user_id=sales_user.id, role_id=sales_role.id, created_at=now), UserRole(user_id=inactive_user.id, role_id=sales_role.id, created_at=now), ] ) self.db.commit() return { "sales_employee_id": sales_employee.id, "inactive_employee_id": inactive_employee.id, } def test_list_employee_options_filters_by_permission_code(self) -> None: from app.services.system_permissions import list_employee_options ids = self._seed_permission_employee_selector_data() purchase_options = list_employee_options(self.db, permission_code="MENU_PURCHASE_ORDER") sales_options = list_employee_options(self.db, permission_code="MENU_SALES_PLANNING") full_options = list_employee_options(self.db) self.assertEqual([option.employee_name for option in purchase_options], ["李四"]) self.assertEqual([option.employee_name for option in sales_options], ["赵销售"]) self.assertIn("销售人员", sales_options[0].role_names) self.assertEqual(sales_options[0].username, "13800000021") self.assertNotIn(ids["inactive_employee_id"], [option.employee_id for option in full_options]) def test_ensure_employee_has_permission_rejects_unqualified_employee(self) -> None: from fastapi import HTTPException from app.services.system_permissions import ensure_employee_has_permission ids = self._seed_permission_employee_selector_data() ensure_employee_has_permission( self.db, ids["sales_employee_id"], "MENU_SALES_PLANNING", "销售人员", ) with self.assertRaises(HTTPException) as ctx: ensure_employee_has_permission( self.db, self.buyer_employee_id, "MENU_SALES_PLANNING", "销售人员", ) self.assertEqual(ctx.exception.status_code, 400) self.assertIn("销售人员必须选择有“客户名录/销售订单菜单”权限的人员", str(ctx.exception.detail)) def test_optional_personnel_validator_allows_empty_value(self) -> None: from app.services.system_permissions import ensure_employee_has_permission self._seed_permission_employee_selector_data() ensure_employee_has_permission( self.db, None, "MENU_PURCHASE_ORDER", "采购员", required=False, ) def test_optional_personnel_validator_rejects_zero_value(self) -> None: from fastapi import HTTPException from app.services.system_permissions import ensure_employee_has_permission self._seed_permission_employee_selector_data() with self.assertRaises(HTTPException) as ctx: ensure_employee_has_permission( self.db, 0, "MENU_PURCHASE_ORDER", "采购员", required=False, ) self.assertEqual(ctx.exception.status_code, 400) self.assertIn("采购员不存在或已停用", str(ctx.exception.detail)) def test_login_session_exposes_employee_id_for_default_selectors(self) -> None: from app.services.auth import get_auth_context, hash_password self._seed_permission_employee_selector_data() context = get_auth_context(self.db, "13800000012") self.assertEqual(context.employee.id, self.buyer_employee_id) self.assertEqual(context.employee.mobile, "13800000012") if __name__ == "__main__": unittest.main()