385 lines
15 KiB
Python
385 lines
15 KiB
Python
"""
|
||
用户服务单元测试（U3 用户模块接入 BaseService）
|
||
=============================================================
|
||
|
||
验证用户模块重构后的关键契约：
|
||
- 新增用户不再触发 salt AttributeError（原 form.salt.data 崩溃点）
|
||
- 密码加密 + salt 写入、行政区划拆分到蛇形列
|
||
- 更新时不传密码保留原密码；传密码则重新加密
|
||
- 分页参数统一 pageNo/pageSize（对齐 BaseService.get_page）
|
||
"""
|
||
|
||
import pytest
|
||
|
||
from src.common.utils import excel_utils
|
||
from src.core.database import db
|
||
from src.core.exceptions import BizError
|
||
from src.modules.system.user import service as user_service
|
||
from src.modules.system.user.repository import UserRepository
|
||
from src.modules.system.user.service import UserService
|
||
|
||
|
||
def _payload(username="testuser", password="123456"):
|
||
"""构造一份合法的用户新增表单数据"""
|
||
return {
|
||
"username": username,
|
||
"realname": "测试用户",
|
||
"gender": 1,
|
||
"avatar": "avatar.png",
|
||
"mobile": "13800000000",
|
||
"email": "test@test.com",
|
||
"deptId": 1,
|
||
"levelId": 1,
|
||
"positionId": 1,
|
||
"city": ["440000", "440100", "440103", "440106"],
|
||
"address": "测试路1号",
|
||
"status": 1,
|
||
"sort": 1,
|
||
"password": password,
|
||
}
|
||
|
||
|
||
def _create_user(app, username="base_user"):
|
||
"""经 UserService.add 新增一个用户并返回其ID"""
|
||
with app.app_context():
|
||
UserService.add(_payload(username=username))
|
||
db.session.commit()
|
||
user = UserRepository.first_by({"username": username})
|
||
return user.id
|
||
|
||
|
||
class TestUserAdd:
|
||
"""新增用户（salt 崩溃点回归）"""
|
||
|
||
def test_add_creates_with_encrypted_password_and_city(self, app):
|
||
with app.app_context():
|
||
UserService.add(_payload())
|
||
db.session.commit()
|
||
user = UserRepository.first_by({"username": "testuser"})
|
||
|
||
assert user is not None
|
||
# 密码已加密且 salt 已写入（原实现此处 AttributeError）
|
||
assert user.password != "123456"
|
||
assert user.salt
|
||
# 行政区划拆分到蛇形列
|
||
assert user.province_code == "440000"
|
||
assert user.city_code == "440100"
|
||
assert user.district_code == "440103"
|
||
assert user.street_code == "440106"
|
||
|
||
def test_add_invalid_city_raises(self, app):
|
||
with app.app_context():
|
||
with pytest.raises(BizError, match="请选择完整的所在城市"):
|
||
data = _payload()
|
||
data["city"] = ["440000"]
|
||
UserService.add(data)
|
||
|
||
def test_add_generates_six_digit_code_from_id(self, app):
|
||
with app.app_context():
|
||
UserService.add(_payload())
|
||
db.session.commit()
|
||
user = UserRepository.first_by({"username": "testuser"})
|
||
|
||
# 编码 = ID 补零为 6 位（如 id=10 -> 000010）
|
||
assert user.code == f"{user.id:06d}"
|
||
assert len(user.code) == 6
|
||
assert user.code.isdigit()
|
||
|
||
|
||
class TestUserUpdate:
|
||
"""更新用户（密码保留/重设）"""
|
||
|
||
def test_update_without_password_keeps_original(self, app):
|
||
user_id = _create_user(app)
|
||
with app.app_context():
|
||
data = _payload(username="base_user")
|
||
data["id"] = user_id
|
||
data.pop("password") # 不传新密码
|
||
UserService.update(data)
|
||
db.session.commit()
|
||
db.session.expire_all()
|
||
user = UserRepository.get_by_id(user_id)
|
||
|
||
assert user.password != "123456"
|
||
assert user.realname == "测试用户"
|
||
|
||
def test_update_with_new_password_reencrypts(self, app):
|
||
user_id = _create_user(app)
|
||
with app.app_context():
|
||
data = _payload(username="base_user", password="newpass888")
|
||
data["id"] = user_id
|
||
UserService.update(data)
|
||
db.session.commit()
|
||
db.session.expire_all()
|
||
user = UserRepository.get_by_id(user_id)
|
||
|
||
assert user.password != "123456"
|
||
assert user.password != "newpass888"
|
||
assert user.salt
|
||
|
||
|
||
class TestReferenceCheck:
|
||
"""删除前用户引用校验（C4 防孤儿引用）"""
|
||
|
||
def test_position_delete_blocked_when_referenced(self, app):
|
||
from src.modules.system.position.repository import PositionRepository
|
||
from src.modules.system.position.service import PositionService
|
||
|
||
with app.app_context():
|
||
pos = PositionRepository.create(name='经理岗', status=1, sort=0)
|
||
db.session.commit()
|
||
pid = pos.id
|
||
|
||
with app.app_context():
|
||
data = _payload(username='ref_user')
|
||
data['positionId'] = pid
|
||
UserService.add(data)
|
||
db.session.commit()
|
||
|
||
with app.app_context():
|
||
with pytest.raises(BizError, match="已被用户引用"):
|
||
PositionService.remove(str(pid))
|
||
|
||
def test_position_delete_unreferenced_succeeds(self, app):
|
||
from src.modules.system.position.repository import PositionRepository
|
||
from src.modules.system.position.service import PositionService
|
||
|
||
with app.app_context():
|
||
pos = PositionRepository.create(name='临时岗', status=1, sort=0)
|
||
db.session.commit()
|
||
pid = pos.id
|
||
|
||
with app.app_context():
|
||
count = PositionService.remove(str(pid))
|
||
assert count == 1
|
||
|
||
|
||
class TestUserZeroIds:
|
||
"""deptId/levelId/positionId 允许 0（无部门/无职级/无岗位）"""
|
||
|
||
def test_add_with_zero_ids(self, app):
|
||
with app.app_context():
|
||
data = _payload(username="zero_ids")
|
||
data["deptId"] = 0
|
||
data["levelId"] = 0
|
||
data["positionId"] = 0
|
||
UserService.add(data)
|
||
db.session.commit()
|
||
user = UserRepository.first_by({"username": "zero_ids"})
|
||
|
||
assert user is not None
|
||
assert user.dept_id == 0
|
||
assert user.level_id == 0
|
||
assert user.position_id == 0
|
||
|
||
|
||
class TestUserPage:
|
||
"""分页参数统一 pageNo/pageSize"""
|
||
|
||
def test_page_uses_pageNo_pageSize(self, app):
|
||
_create_user(app, "user_a")
|
||
_create_user(app, "user_b")
|
||
_create_user(app, "user_c")
|
||
|
||
with app.app_context():
|
||
data = UserService.get_page({"pageNo": 2, "pageSize": 2})
|
||
|
||
# 3 条数据、每页 2 条 → 第 2 页 1 条
|
||
assert data["total"] == 3
|
||
assert data["size"] == 2
|
||
assert data["current"] == 2
|
||
assert data["pages"] == 2
|
||
assert len(data["records"]) == 1
|
||
|
||
|
||
class TestUserImportExport:
|
||
"""用户Excel导入/导出（复用 excel_utils）"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _clear_dict_cache(self):
|
||
from src.modules.dictionary import dict_utils
|
||
|
||
dict_utils.clear_cache()
|
||
yield
|
||
dict_utils.clear_cache()
|
||
|
||
def _seed_refs(self, app):
|
||
"""构造部门/职级/岗位/角色 + 性别字典，返回各 ID（避免跨上下文传模型实例）"""
|
||
from src.modules.dictionary.dict_item.repository import DictItemRepository
|
||
from src.modules.dictionary.dict.repository import DictRepository
|
||
from src.modules.system.dept.repository import DeptRepository
|
||
from src.modules.system.level.repository import LevelRepository
|
||
from src.modules.system.position.repository import PositionRepository
|
||
from src.modules.system.role.repository import RoleRepository
|
||
|
||
with app.app_context():
|
||
dept = DeptRepository.create(name="研发部", status=1, sort=1)
|
||
level = LevelRepository.create(name="高级工程师", status=1, sort=1)
|
||
pos = PositionRepository.create(name="后端工程师", status=1, sort=1)
|
||
role = RoleRepository.create(name="普通用户", code="normal", status=1, sort=1)
|
||
dict_obj = DictRepository.create(name="性别", code="gender", sort=1, create_user="system")
|
||
for name, value, sort in [("男", "1", 1), ("女", "2", 2), ("保密", "3", 3)]:
|
||
DictItemRepository.create(
|
||
dict_id=dict_obj.id, name=name, value=value,
|
||
sort=sort, create_user="system",
|
||
)
|
||
db.session.commit()
|
||
return dept.id, level.id, pos.id, role.id
|
||
|
||
def _create_user_with_refs(self, app, dept_id, level_id, pos_id, role_id, username="exp_user", realname="导入导出"):
|
||
"""经 UserService.add 创建含部门/职级/岗位/角色的用户，用于导出测试"""
|
||
with app.app_context():
|
||
data = _payload(username=username)
|
||
data.update({
|
||
"realname": realname,
|
||
"deptId": dept_id,
|
||
"levelId": level_id,
|
||
"positionId": pos_id,
|
||
"roles": [role_id],
|
||
})
|
||
UserService.add(data)
|
||
db.session.commit()
|
||
|
||
def _patch_upload(self, monkeypatch, tmp_path, file_name="users.xlsx"):
|
||
"""将上传模拟指向 tmp_path/temp/20260817 目录，返回该文件路径"""
|
||
target_dir = tmp_path / "temp" / "20260817"
|
||
target_dir.mkdir(parents=True, exist_ok=True)
|
||
target = target_dir / file_name
|
||
monkeypatch.setattr(
|
||
excel_utils.upload, "upload_file",
|
||
lambda *a, **k: {"filePath": "/temp/20260817/" + file_name},
|
||
)
|
||
monkeypatch.setattr(excel_utils, "FLASK_UPLOAD_DIR", str(tmp_path))
|
||
return target
|
||
|
||
def _run_export(self, app, monkeypatch, params):
|
||
"""调用导出并捕获表头/行数据（跳过文件落盘）"""
|
||
captured = {}
|
||
|
||
def fake_save(headers, rows, filename_prefix="export"):
|
||
captured["headers"] = headers
|
||
captured["rows"] = rows
|
||
return "/static/temp/user_xxx.xlsx"
|
||
|
||
monkeypatch.setattr(excel_utils, "save_export_file", fake_save)
|
||
|
||
with app.app_context():
|
||
result = user_service.export_users(params)
|
||
url = result.get_json()["data"]
|
||
return captured, url
|
||
|
||
# ============================================================
|
||
# 导出
|
||
# ============================================================
|
||
|
||
def test_export_includes_enriched_names_and_url(self, app, monkeypatch):
|
||
dept_id, level_id, pos_id, role_id = self._seed_refs(app)
|
||
self._create_user_with_refs(app, dept_id, level_id, pos_id, role_id, "exp_user")
|
||
|
||
captured, url = self._run_export(app, monkeypatch, None)
|
||
|
||
assert url.startswith("/static")
|
||
# 表头为导出列定义（含富化后的展示列）
|
||
assert captured["headers"] == [
|
||
"用户编码", "登录账号", "用户姓名", "性别", "手机号", "邮箱",
|
||
"部门", "职级", "岗位", "用户角色", "状态", "详细地址", "创建时间",
|
||
]
|
||
row = captured["rows"][0]
|
||
assert len(row[0]) == 6 # 用户编码补零 6 位
|
||
assert row[1] == "exp_user"
|
||
assert row[2] == "导入导出"
|
||
assert row[3] == "男" # 性别走字典
|
||
assert row[6] == "研发部" # 部门名称
|
||
assert row[7] == "高级工程师" # 职级名称
|
||
assert row[8] == "后端工程师" # 岗位名称
|
||
assert row[9] == "普通用户" # 角色中文名
|
||
assert row[10] == "正常" # 状态展示为中文
|
||
|
||
def test_export_filters_by_realname(self, app, monkeypatch):
|
||
dept_id, level_id, pos_id, role_id = self._seed_refs(app)
|
||
self._create_user_with_refs(app, dept_id, level_id, pos_id, role_id, "exp_user", realname="张三")
|
||
self._create_user_with_refs(app, dept_id, level_id, pos_id, role_id, "other_user", realname="李四")
|
||
|
||
captured, _ = self._run_export(app, monkeypatch, {"realname": "张"})
|
||
|
||
assert [r[1] for r in captured["rows"]] == ["exp_user"]
|
||
|
||
# ============================================================
|
||
# 导入
|
||
# ============================================================
|
||
|
||
def test_import_creates_users_with_resolved_refs(self, app, monkeypatch, tmp_path):
|
||
import shutil
|
||
|
||
from openpyxl.workbook import Workbook
|
||
|
||
from src.modules.association.user_role.repository import UserRoleRepository
|
||
|
||
dept_id, level_id, pos_id, role_id = self._seed_refs(app)
|
||
|
||
# 构造真实 xlsx：部门/职级/岗位/角色按名称匹配，密码缺省用默认密码
|
||
src = tmp_path / "users.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.worksheets[0]
|
||
ws.append(["用户ID", "登录账号", "用户姓名", "性别", "手机号", "邮箱",
|
||
"部门名称", "职级名称", "岗位名称", "用户角色", "状态", "排序",
|
||
"详细地址", "个人简介", "备注", "初始密码"])
|
||
ws.append([None, "imp_user", "导入用户", "男", "13900000000", "imp@test.com",
|
||
"研发部", "高级工程师", "后端工程师", "普通用户", "正常", 1,
|
||
"测试路3号", "简介", "备注", ""])
|
||
wb.save(src)
|
||
|
||
target = self._patch_upload(monkeypatch, tmp_path)
|
||
shutil.copy(src, target)
|
||
|
||
with app.app_context():
|
||
result = user_service.import_users()
|
||
db.session.commit()
|
||
user = UserRepository.first_by({"username": "imp_user"})
|
||
|
||
assert result.get_json()["code"] == 0
|
||
assert user is not None
|
||
# 编码按 ID 补零 6 位（与 UserService.add 一致）
|
||
assert user.code == f"{user.id:06d}"
|
||
# 部门/职级/岗位按名称解析为 ID
|
||
assert user.dept_id == dept_id
|
||
assert user.level_id == level_id
|
||
assert user.position_id == pos_id
|
||
# 角色按名称关联
|
||
rels = UserRoleRepository.find_list({"user_id": user.id})
|
||
assert [r.role_id for r in rels] == [role_id]
|
||
# 密码缺省为默认密码并已加密
|
||
assert user.password != "123456"
|
||
assert user.salt
|
||
assert not target.exists() # 临时导入文件已清理
|
||
|
||
def test_import_duplicate_username_fails(self, app, monkeypatch, tmp_path):
|
||
import shutil
|
||
|
||
from openpyxl.workbook import Workbook
|
||
|
||
with app.app_context():
|
||
UserService.add(_payload(username="dup_user"))
|
||
db.session.commit()
|
||
|
||
src = tmp_path / "users.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.worksheets[0]
|
||
ws.append(["用户ID", "登录账号", "用户姓名", "性别", "手机号", "邮箱",
|
||
"部门名称", "职级名称", "岗位名称", "用户角色", "状态", "排序",
|
||
"详细地址", "个人简介", "备注", "初始密码"])
|
||
ws.append([None, "dup_user", "导入用户", "男", "13900000000", "imp@test.com",
|
||
"", "", "", "", "正常", 1, "测试路3号", "", "", ""])
|
||
wb.save(src)
|
||
|
||
target = self._patch_upload(monkeypatch, tmp_path)
|
||
shutil.copy(src, target)
|
||
|
||
with app.app_context():
|
||
result = user_service.import_users()
|
||
|
||
body = result.get_json()
|
||
assert body["code"] == 1
|
||
assert "dup_user" in body["msg"]
|
||
assert "已存在" in body["msg"]
|