签入版本

2026-09-12 19:09:28 +08:00
commit 4eae62bc17
803 changed files with 105034 additions and 0 deletions
@@ -0,0 +1,213 @@
+"""
+职级服务单元测试（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"]