""" 职级服务单元测试（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"]