214 lines
8.6 KiB
Python
214 lines
8.6 KiB
Python
"""
|
||
职级服务单元测试（Excel 导入/导出复用 excel_utils）
|
||
=============================================================
|
||
|
||
验证：
|
||
- export_levels 支持与分页一致的筛选参数（name 模糊、status 精确）
|
||
- 无筛选时导出全部记录；表头与数据列对齐
|
||
- import_levels 上传/解析/入库/清理（临时文件生命周期）
|
||
- 上传失败与文件缺失时透传错误
|
||
|
||
文件落盘通过 monkeypatch excel_utils.save_export_file 跳过，
|
||
仅验证查询、列构建与写盘调用契约。
|
||
"""
|
||
|
||
import pytest
|
||
|
||
from src.common.utils import excel_utils
|
||
from src.core.exceptions import BizError
|
||
from src.modules.system.level import service as level_service
|
||
from src.modules.system.level.repository import LevelRepository
|
||
|
||
|
||
@pytest.fixture
|
||
def seed_levels(app):
|
||
"""插入职级测试数据（sqlite 内存库）"""
|
||
from src.core.database import db
|
||
|
||
with app.app_context():
|
||
LevelRepository.create(name="初级工程师", status=1, sort=1)
|
||
LevelRepository.create(name="中级工程师", status=1, sort=2)
|
||
LevelRepository.create(name="高级工程师", status=2, sort=3)
|
||
LevelRepository.create(name="产品经理", status=1, sort=4)
|
||
db.session.commit()
|
||
|
||
|
||
def _run_export(app, monkeypatch, params):
|
||
"""调用导出并捕获表头/行数据（跳过文件落盘，仅验证查询与列构建）"""
|
||
captured = {}
|
||
|
||
def fake_save(headers, rows, filename_prefix="export"):
|
||
captured["headers"] = headers
|
||
captured["rows"] = rows
|
||
return "/static/temp/level_xxx.xlsx"
|
||
|
||
monkeypatch.setattr(excel_utils, "save_export_file", fake_save)
|
||
|
||
with app.app_context():
|
||
result = level_service.export_levels(params)
|
||
url = result.get_json()["data"]
|
||
|
||
return captured, url
|
||
|
||
|
||
class TestExportLevels:
|
||
"""Excel 导出：筛选参数 + 列对齐"""
|
||
|
||
HEADER = ["职级ID", "职级名称", "职级排序", "职级状态", "创建时间", "更新时间"]
|
||
|
||
def test_export_no_filter_exports_all(self, app, monkeypatch, seed_levels):
|
||
captured, url = _run_export(app, monkeypatch, None)
|
||
assert captured["headers"] == self.HEADER
|
||
assert [r[1] for r in captured["rows"]] == ["初级工程师", "中级工程师", "高级工程师", "产品经理"]
|
||
assert url.startswith("/static")
|
||
|
||
def test_export_filters_by_name(self, app, monkeypatch, seed_levels):
|
||
captured, _ = _run_export(app, monkeypatch, {"name": "工程师"})
|
||
assert captured["headers"] == self.HEADER
|
||
assert [r[1] for r in captured["rows"]] == ["初级工程师", "中级工程师", "高级工程师"]
|
||
|
||
def test_export_filters_by_status(self, app, monkeypatch, seed_levels):
|
||
captured, _ = _run_export(app, monkeypatch, {"status": 2})
|
||
assert [r[1] for r in captured["rows"]] == ["高级工程师"]
|
||
|
||
def test_export_filters_by_name_and_status(self, app, monkeypatch, seed_levels):
|
||
captured, _ = _run_export(app, monkeypatch, {"name": "工程师", "status": 1})
|
||
assert [r[1] for r in captured["rows"]] == ["初级工程师", "中级工程师"]
|
||
|
||
def test_export_columns_aligned(self, app, monkeypatch, seed_levels):
|
||
captured, _ = _run_export(app, monkeypatch, {"name": "产品经理"})
|
||
assert captured["headers"] == self.HEADER
|
||
row = captured["rows"][0]
|
||
assert row[0] is not None # 职级ID
|
||
assert row[1] == "产品经理" # 职级名称
|
||
assert row[2] == 4 # 职级排序
|
||
assert row[3] == "正常" # 职级状态（formatter 展示为中文）
|
||
assert row[4] # 创建时间
|
||
assert row[5] # 更新时间
|
||
|
||
def test_export_empty_result_keeps_header(self, app, monkeypatch, seed_levels):
|
||
captured, _ = _run_export(app, monkeypatch, {"name": "不存在的职级"})
|
||
assert captured["headers"] == self.HEADER
|
||
assert captured["rows"] == []
|
||
|
||
|
||
class TestLevelList:
|
||
"""职级列表（无分页）：参数条件筛选 + status=1 硬性过滤"""
|
||
|
||
def test_list_no_filter_returns_enabled_only(self, app, seed_levels):
|
||
with app.app_context():
|
||
data = level_service.get_level_list(None)
|
||
assert [d["name"] for d in data] == ["初级工程师", "中级工程师", "产品经理"]
|
||
|
||
def test_list_filters_by_name(self, app, seed_levels):
|
||
with app.app_context():
|
||
data = level_service.get_level_list({"name": "工程师"})
|
||
assert [d["name"] for d in data] == ["初级工程师", "中级工程师"]
|
||
|
||
def test_list_status_param_is_ignored(self, app, seed_levels):
|
||
# status 为硬性条件，即使传入 status=2 也只返回启用（status=1）记录
|
||
with app.app_context():
|
||
data = level_service.get_level_list({"name": "工程师", "status": 2})
|
||
assert [d["name"] for d in data] == ["初级工程师", "中级工程师"]
|
||
|
||
|
||
class TestImportLevels:
|
||
"""职级Excel导入（复用 excel_utils 上传/解析/清理机制）"""
|
||
|
||
def _patch_upload(self, monkeypatch, tmp_path, file_name="levels.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 test_import_success(self, app, monkeypatch, tmp_path):
|
||
import shutil
|
||
|
||
from openpyxl.workbook import Workbook
|
||
|
||
from src.modules.system.level.models import Level
|
||
|
||
# 构造真实 xlsx 并放入模拟的上传目录（temp/日期/文件）
|
||
src = tmp_path / "levels.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.worksheets[0]
|
||
ws.append(["职级ID", "职级名称", "职级排序", "职级状态"])
|
||
ws.append([None, "初级", 1, "正常"])
|
||
ws.append([None, "停用级", 2, "禁用"])
|
||
ws.append([None, None, None, None]) # 整行为空 → 跳过
|
||
wb.save(src)
|
||
|
||
target = self._patch_upload(monkeypatch, tmp_path)
|
||
shutil.copy(src, target)
|
||
|
||
with app.app_context():
|
||
result = level_service.import_levels()
|
||
rows = Level.query.order_by(Level.sort).all()
|
||
|
||
assert result.get_json()["code"] == 0
|
||
assert [r.name for r in rows] == ["初级", "停用级"]
|
||
assert rows[0].status == 1 # "正常" → 1
|
||
assert rows[1].status == 2 # "禁用" → 2
|
||
assert not target.exists() # 临时导入文件已清理
|
||
|
||
def test_import_required_missing_fails_whole_import(self, app, monkeypatch, tmp_path):
|
||
import shutil
|
||
|
||
from openpyxl.workbook import Workbook
|
||
|
||
from src.modules.system.level.models import Level
|
||
|
||
# 名称必填缺失 → 行级错误 → 整体失败，无任何数据入库
|
||
src = tmp_path / "levels.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.worksheets[0]
|
||
ws.append(["职级ID", "职级名称", "职级排序", "职级状态"])
|
||
ws.append([None, "", 1, "正常"])
|
||
wb.save(src)
|
||
|
||
target = self._patch_upload(monkeypatch, tmp_path)
|
||
shutil.copy(src, target)
|
||
|
||
with app.app_context():
|
||
result = level_service.import_levels()
|
||
count = Level.query.count()
|
||
|
||
body = result.get_json()
|
||
assert body["code"] == 1
|
||
assert "职级名称不能为空" in body["msg"]
|
||
assert count == 0
|
||
|
||
def test_import_upload_failure_propagates(self, app, monkeypatch):
|
||
# 上传失败由 upload_file 抛 BizError，import_levels 捕获后透传具体错误信息
|
||
def _fail_upload(*args, **kwargs):
|
||
raise BizError("文件格式不支持，仅支持 .xls, .xlsx 格式")
|
||
|
||
monkeypatch.setattr(excel_utils.upload, "upload_file", _fail_upload)
|
||
|
||
with app.app_context():
|
||
result = level_service.import_levels()
|
||
|
||
body = result.get_json()
|
||
assert body["code"] == 1
|
||
assert "文件格式不支持" in body["msg"]
|
||
|
||
def test_import_file_missing_returns_failed(self, app, monkeypatch):
|
||
monkeypatch.setattr(
|
||
excel_utils.upload, "upload_file",
|
||
lambda *a, **k: {"filePath": "/temp/20260817/nonexist.xlsx"},
|
||
)
|
||
monkeypatch.setattr(excel_utils, "FLASK_UPLOAD_DIR", "nonexistent_dir")
|
||
|
||
with app.app_context():
|
||
result = level_service.import_levels()
|
||
|
||
body = result.get_json()
|
||
assert body["code"] == 1
|
||
assert "导入文件不存在" in body["msg"]
|