737 lines
27 KiB
Python
737 lines
27 KiB
Python
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_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="DEPARTMENT",
|
|
node_label="错误部门",
|
|
dept_code="BAD-DEPT",
|
|
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()
|