# -*- coding: utf-8 -*-
"""BUG-5 ~ BUG-9 及 H-1 ~ H-4 修复验证测试。"""
import asyncio
import json
import sys

import pytest

sys.path.insert(0, 'src')


# ============================================================
# BUG-5: 权限装饰器走缓存（异步 + 同步端点）
# ============================================================

def _make_perm_app(monkeypatch, user_id, perm_calls, denied=False):
    from fastapi import FastAPI
    from starlette.middleware.base import BaseHTTPMiddleware
    from starlette.requests import Request

    from core.access_decorators import permission_required

    async def fake_get_permissions_list(uid):
        perm_calls.append(uid)
        return [] if denied else ['sys:x:list']

    import modules.system.menu.service as menu_service
    monkeypatch.setattr(menu_service, 'get_permissions_list', fake_get_permissions_list)
    sync_db_calls = []
    monkeypatch.setattr(
        menu_service, 'get_permissions_list_sync',
        lambda uid: sync_db_calls.append(uid) or ([] if denied else ['sys:x:list']))

    app = FastAPI()

    @app.middleware("http")
    async def inject_user(request: Request, call_next):
        request.state.user_id = user_id
        return await call_next(request)

    @app.get("/sync")
    @permission_required("sys:x:list")
    def sync_ep(request: Request):
        return {"code": 0, "ok": True}

    @app.get("/async")
    @permission_required("sys:x:list")
    async def async_ep(request: Request):
        return {"code": 0, "ok": True}

    return app, sync_db_calls


def test_perm_decorator_sync_endpoint_uses_sync_check(monkeypatch):
    from fastapi.testclient import TestClient

    calls = []
    app, sync_db_calls = _make_perm_app(monkeypatch, user_id=2, perm_calls=calls)
    with TestClient(app) as client:
        resp = client.get("/sync")
    assert resp.json()["ok"] is True
    # 同步端点直接走同步 DB 校验（不再每请求新建事件循环 / 依赖 Redis 缓存）
    assert calls == []
    assert sync_db_calls == [2]


def test_perm_decorator_async_endpoint_uses_cache(monkeypatch):
    from fastapi.testclient import TestClient

    calls = []
    app, sync_db_calls = _make_perm_app(monkeypatch, user_id=2, perm_calls=calls)
    with TestClient(app) as client:
        resp = client.get("/async")
    assert resp.json()["ok"] is True
    assert calls == [2]
    assert sync_db_calls == []


def test_perm_decorator_denied(monkeypatch):
    from fastapi.testclient import TestClient

    calls = []
    app, sync_db_calls = _make_perm_app(
        monkeypatch, user_id=2, perm_calls=calls, denied=True)
    with TestClient(app) as client:
        resp = client.get("/async")
    assert resp.json()["code"] == 1
    assert resp.json()["ok"] is False


def test_perm_decorator_admin_bypass(monkeypatch):
    from fastapi.testclient import TestClient

    calls = []
    app, sync_db_calls = _make_perm_app(monkeypatch, user_id=1, perm_calls=calls)
    with TestClient(app) as client:
        resp = client.get("/async")
    assert resp.json()["ok"] is True
    assert calls == []  # 管理员不查权限


# ============================================================
# BUG-6: CORS 为最外层
# ============================================================

def test_cors_is_outermost():
    from core.app import create_app
    from starlette.middleware.cors import CORSMiddleware

    app = create_app()
    classes = [m.cls for m in app.user_middleware]
    assert classes[0] is CORSMiddleware


# ============================================================
# BUG-7: 时间字段序列化统一（naive/aware UTC 均转本地）
# ============================================================

def test_to_dict_naive_utc_matches_aware_utc():
    from datetime import datetime, timezone
    from modules.system.menu.models import Menu

    naive_obj = Menu()
    naive_obj.create_time = datetime(2026, 1, 1, 8, 0, 0)  # naive，表示 UTC 08:00
    aware_obj = Menu()
    aware_obj.create_time = datetime(2026, 1, 1, 8, 0, 0, tzinfo=timezone.utc)

    d1 = naive_obj.to_dict()
    d2 = aware_obj.to_dict()
    # 两者应一致（本地时间），证明读回 naive 不再显示 UTC 原始值
    assert d1['create_time'] == d2['create_time']
    expected = datetime(2026, 1, 1, 8, 0, 0, tzinfo=timezone.utc) \
        .astimezone().strftime('%Y-%m-%d %H:%M:%S')
    assert d1['create_time'] == expected


# ============================================================
# BUG-8: 导出 URL 反斜杠转 / 分隔
# ============================================================

def test_export_level_url_uses_forward_slashes(monkeypatch, tmp_path):
    from fastapi import FastAPI
    from starlette.middleware.base import BaseHTTPMiddleware
    from starlette.requests import Request
    from starlette.testclient import TestClient

    import modules.system.level.service as level_svc
    import time

    up = tmp_path / 'uploads'
    temp = up / 'temp'
    monkeypatch.setattr(level_svc, 'FASTAPI_UPLOAD_DIR', str(up))
    monkeypatch.setattr(level_svc, 'FASTAPI_TEMP_PATH', str(temp))
    monkeypatch.setattr(level_svc, 'FASTAPI_FILE_URL', 'http://img.test')
    monkeypatch.setattr(level_svc, 'get_db_session', lambda: None)

    class FakeQuery:
        def filter(self, *a, **k):
            return self
        def order_by(self, *a, **k):
            return self
        def limit(self, *a, **k):
            return self
        def yield_per(self, *a, **k):
            return iter(())  # 无数据，跳过导出与 expunge

    class FakeRepo:
        def filter_by(self):
            return FakeQuery()

    monkeypatch.setattr(level_svc.level_service, 'repo', FakeRepo())

    app = FastAPI()

    @app.middleware("http")
    async def inject(request: Request, call_next):
        request.state.user_id = 1
        return await call_next(request)

    @app.get('/export')
    def export(request: Request):
        return level_svc.level_service.export_level(request)

    with TestClient(app) as client:
        resp = client.get('/export')
    url = resp.json()['data']
    assert url.startswith('http://img.test/')
    assert '\\' not in url
    assert '/temp/' in url
    assert url.endswith('.xlsx')


# ============================================================
# BUG-9: UserForm.city/roles 由 Any 收紧为带类型校验
# ============================================================

def _valid_user_form():
    from datetime import date
    return dict(
        realname='张三', nickname='张三', gender=1, avatar='a.png',
        mobile='13800000000', email='a@b.com', birthday=date(2000, 1, 1),
        deptId=1, levelId=1, positionId=1,
        city=['110000', '110100', '110101'], address='北京市',
        username='zhangsan', password='123456', status=1, sort=1, roles=[1, 2],
    )


def test_user_form_city_rejects_string_and_oversize():
    from pydantic import ValidationError
    from modules.system.user.schemas import UserForm

    base = _valid_user_form()
    # 合法数据应通过
    UserForm(**base)
    # city 传字符串应拒绝
    with pytest.raises(ValidationError):
        UserForm(**{**base, 'city': '北京市'})
    # city 元素非字符串（Pydantic v2 不隐式转 str）应拒绝
    with pytest.raises(ValidationError):
        UserForm(**{**base, 'city': [1, 2, 3]})
    # city 超过 3 个元素应拒绝
    with pytest.raises(ValidationError):
        UserForm(**{**base, 'city': ['1', '2', '3', '4']})


def test_user_form_roles_rejects_string():
    from pydantic import ValidationError
    from modules.system.user.schemas import UserForm

    base = _valid_user_form()
    # roles 传字符串应拒绝
    with pytest.raises(ValidationError):
        UserForm(**{**base, 'roles': '1,2'})
    # roles 合法整型列表应通过
    form = UserForm(**{**base, 'roles': [1, 2]})
    assert form.roles == [1, 2]


# ============================================================
# H-1: add_user 排除 id（防批量赋值伪造）
# ============================================================

def _make_request(user_id=1):
    import types
    req = types.SimpleNamespace()
    req.state = types.SimpleNamespace(user_id=user_id)
    return req


def test_add_user_rejects_forged_id(sqlite_db, monkeypatch):
    import asyncio
    from modules.system.user import service as user_svc
    from modules.system.user.schemas import UserForm
    from modules.system.user.repository import user_repo

    # 跳过头像文件落盘，直接返回原值
    monkeypatch.setattr(user_svc, 'save_file', lambda url, _dir: url)

    # 客户端伪造 id=999，应被排除，不写入数据库
    form = UserForm(**{**_valid_user_form(), 'id': 999})
    asyncio.run(user_svc.add_user(_make_request(), form))

    user = user_repo.get_one(username='zhangsan')
    assert user is not None
    assert user.id != 999
    assert user_repo.get_by_id(999) is None


# ============================================================
# H-2: login_required 直接查库校验用户状态
# ============================================================

def _make_login_app():
    from fastapi import Depends, FastAPI
    from starlette.requests import Request
    from starlette.responses import JSONResponse

    import api.deps as deps
    from core.exceptions import AuthorizationException

    app = FastAPI()

    @app.exception_handler(AuthorizationException)
    async def _auth_exc(request: Request, exc: AuthorizationException):
        return JSONResponse(
            status_code=200,
            content={"code": exc.code, "data": None, "msg": exc.msg, "ok": False},
        )

    @app.get('/protected')
    async def protected(request: Request, _=Depends(deps.login_required)):
        return {"code": 0, "ok": True, "user_id": request.state.user_id}

    return app


def test_login_required_rejects_disabled_and_deleted_user(monkeypatch):
    import types
    from fastapi.testclient import TestClient

    import api.deps as deps
    import modules.system.user.repository as user_repo_mod

    async def fake_blacklist(_token):
        return False
    monkeypatch.setattr(deps, 'is_token_blacklisted', fake_blacklist)

    def _payload(user_id):
        return {'code': 0, 'data': {'userId': user_id, 'username': 'u', 'realname': 'r'}}

    class FakeRepo:
        def __init__(self, users):
            self.users = users

        def get_by_id(self, uid):
            return self.users.get(uid)

    # 正常用户（status=1）放行
    monkeypatch.setattr(deps, 'parse_payload', lambda _t: _payload(2))
    monkeypatch.setattr(user_repo_mod, 'user_repo', FakeRepo({2: types.SimpleNamespace(status=1)}))
    with TestClient(_make_login_app()) as client:
        resp = client.get('/protected', headers={'Authorization': 'Bearer x'})
    assert resp.json()['ok'] is True
    assert resp.json()['user_id'] == 2

    # 禁用用户（status=2）拒绝
    monkeypatch.setattr(deps, 'parse_payload', lambda _t: _payload(3))
    monkeypatch.setattr(user_repo_mod, 'user_repo', FakeRepo({3: types.SimpleNamespace(status=2)}))
    with TestClient(_make_login_app()) as client:
        resp = client.get('/protected', headers={'Authorization': 'Bearer x'})
    assert resp.json()['ok'] is False
    assert resp.json()['code'] == 401

    # 已删除/不存在用户（查无此人）拒绝
    monkeypatch.setattr(deps, 'parse_payload', lambda _t: _payload(4))
    monkeypatch.setattr(user_repo_mod, 'user_repo', FakeRepo({}))
    with TestClient(_make_login_app()) as client:
        resp = client.get('/protected', headers={'Authorization': 'Bearer x'})
    assert resp.json()['ok'] is False
    assert resp.json()['code'] == 401


# ============================================================
# H-2b: login_required 用户有效状态缓存（命中免查库，失效即时踢出）
# ============================================================

def _login_request():
    from starlette.requests import Request
    scope = {
        'type': 'http', 'method': 'GET', 'path': '/protected',
        'headers': [(b'authorization', b'Bearer x')],
        'query_string': b'', 'route': None,
    }
    return Request(scope)


@pytest.mark.asyncio
async def test_login_required_cache_hit_skips_db(monkeypatch):
    import fakeredis.aioredis
    import api.deps as deps
    import modules.system.user.repository as user_repo_mod
    from utils import perm_cache

    async def _blacklist(_t):
        return False
    monkeypatch.setattr(deps, 'is_token_blacklisted', _blacklist)
    monkeypatch.setattr(deps, 'parse_payload',
                        lambda _t: {'code': 0, 'data': {'userId': 2, 'username': 'u', 'realname': 'r'}})

    fake_redis = fakeredis.aioredis.FakeRedis(decode_responses=True)
    monkeypatch.setattr(perm_cache, 'get_redis', lambda: fake_redis)
    await fake_redis.set('user:active:2', '1')

    # 缓存命中：不应查询数据库
    class _NoDb:
        def get_by_id(self, uid):
            raise AssertionError('缓存命中时不应查询数据库')
    monkeypatch.setattr(user_repo_mod, 'user_repo', _NoDb())

    request = _login_request()
    await deps.login_required(request)
    assert request.state.user_id == 2


@pytest.mark.asyncio
async def test_login_required_cache_inactive_rejects_without_db(monkeypatch):
    import fakeredis.aioredis
    import api.deps as deps
    import modules.system.user.repository as user_repo_mod
    from utils import perm_cache
    from core.exceptions import AuthorizationException

    async def _blacklist(_t):
        return False
    monkeypatch.setattr(deps, 'is_token_blacklisted', _blacklist)
    monkeypatch.setattr(deps, 'parse_payload',
                        lambda _t: {'code': 0, 'data': {'userId': 3, 'username': 'u', 'realname': 'r'}})

    fake_redis = fakeredis.aioredis.FakeRedis(decode_responses=True)
    monkeypatch.setattr(perm_cache, 'get_redis', lambda: fake_redis)
    await fake_redis.set('user:active:3', '0')  # 缓存已标记禁用

    class _NoDb:
        def get_by_id(self, uid):
            raise AssertionError('缓存命中（禁用）时不应查询数据库')
    monkeypatch.setattr(user_repo_mod, 'user_repo', _NoDb())

    with pytest.raises(AuthorizationException):
        await deps.login_required(_login_request())


@pytest.mark.asyncio
async def test_login_required_cache_miss_queries_db_and_caches(monkeypatch):
    import types
    import fakeredis.aioredis
    import api.deps as deps
    import modules.system.user.repository as user_repo_mod
    from utils import perm_cache

    async def _blacklist(_t):
        return False
    monkeypatch.setattr(deps, 'is_token_blacklisted', _blacklist)
    monkeypatch.setattr(deps, 'parse_payload',
                        lambda _t: {'code': 0, 'data': {'userId': 5, 'username': 'u', 'realname': 'r'}})

    fake_redis = fakeredis.aioredis.FakeRedis(decode_responses=True)
    monkeypatch.setattr(perm_cache, 'get_redis', lambda: fake_redis)

    class _Repo:
        def get_by_id(self, uid):
            return types.SimpleNamespace(status=1)
    monkeypatch.setattr(user_repo_mod, 'user_repo', _Repo())

    request = _login_request()
    await deps.login_required(request)
    assert request.state.user_id == 5
    # 回源 DB 后回填缓存，供后续请求命中
    assert await fake_redis.get('user:active:5') == '1'


# ============================================================
# H-3: 上传流式读取，超限立即中断（内存 DoS 防护）
# ============================================================

class _ChunkedFile:
    """模拟异步分块读取的 UploadFile（每次 read 返回一块）。"""

    def __init__(self, chunks):
        self.chunks = list(chunks)

    async def read(self, size=65536):
        if not self.chunks:
            return b''
        return self.chunks.pop(0)


@pytest.mark.asyncio
async def test_read_upload_limited_aborts_before_reading_all():
    from modules.upload.service import _read_upload_limited, _max_upload_bytes

    limit = _max_upload_bytes()  # 例如 10MB
    chunk = b'x' * 65536          # 64KB
    n_chunks = limit // 65536 + 5  # 总量必然超限

    file = _ChunkedFile([chunk] * n_chunks)
    content, err = await _read_upload_limited(file, limit)

    assert content is None
    body = json.loads(err.body)
    assert body['code'] == 1
    assert '超过限制' in body['msg']
    # 未读完全部块即中断，证明不是先整文件读入内存
    assert len(file.chunks) > 0


@pytest.mark.asyncio
async def test_read_upload_limited_ok_under_limit():
    from modules.upload.service import _read_upload_limited, _max_upload_bytes

    payload = b'hello' * 100
    content, err = await _read_upload_limited(_ChunkedFile([payload]), _max_upload_bytes())
    assert err is None
    assert content == payload


# ============================================================
# H-4: set_permission 生成的权限节点默认启用（status=0）
# ============================================================

def test_set_permission_creates_enabled_nodes(sqlite_db):
    from modules.system.menu.service import set_permission
    from modules.system.menu.repository import menu_repo

    set_permission(0, [1, 5], '用户管理', '/system/user', 1, 1)

    nodes = menu_repo.get_all(parent_id=1, type=1)
    assert len(nodes) == 2
    perms = {node.permission for node in nodes}
    assert perms == {'sys:user:page', 'sys:user:add'}
    for node in nodes:
        assert node.status == 0


# ============================================================
# H-5: set_permission 编辑菜单时复用旧按钮节点 id，role_menu 授权不失效
# ============================================================

def test_set_permission_preserves_node_ids_on_reedit(sqlite_db):
    from modules.system.menu.service import set_permission
    from modules.system.menu.repository import menu_repo
    from modules.system.role_menu.models import RoleMenu
    from modules.system.role_menu.repository import rolemenu_repo
    from core.database import get_db_session

    # 首次保存：生成 查询(1)/添加(5) 两个按钮节点
    set_permission(0, [1, 5], '用户管理', '/system/user', 1, 1)
    first = {node.sort: node.id for node in menu_repo.get_all(parent_id=1, type=1)}
    assert set(first) == {1, 5}

    # 模拟角色已把"添加"按钮授权给角色 1
    get_db_session().add(RoleMenu(role_id=1, menu_id=first[5]))
    sqlite_db.commit()

    # 再次编辑菜单（勾选增加 修改/10）→ 已勾选旧节点必须复用原 id
    set_permission(0, [1, 5, 10], '用户管理', '/system/user', 1, 1)
    second = {node.sort: node.id for node in menu_repo.get_all(parent_id=1, type=1)}
    assert set(second) == {1, 5, 10}
    assert second[1] == first[1]              # 查询节点 id 未变
    assert second[5] == first[5]              # 添加节点 id 未变 → role_menu 引用仍有效
    assert second[10] not in first.values()   # 修改为新增节点

    # role_menu 授权仍指向有效（is_delete=0）的按钮节点
    refs = rolemenu_repo.filter(RoleMenu.menu_id == first[5]).all()
    assert len(refs) == 1
    assert menu_repo.get_by_id(first[5]) is not None


def test_set_permission_soft_deletes_unchecked_nodes(sqlite_db):
    from modules.system.menu.service import set_permission
    from modules.system.menu.repository import menu_repo

    set_permission(0, [1, 5], '用户管理', '/system/user', 1, 1)
    first = {node.sort: node.id for node in menu_repo.get_all(parent_id=1, type=1)}
    assert set(first) == {1, 5}

    # 编辑时去掉 添加(5) → 旧节点软删，不再参与鉴权
    set_permission(0, [1], '用户管理', '/system/user', 1, 1)
    remaining = menu_repo.get_all(parent_id=1, type=1)
    assert [node.sort for node in remaining] == [1]
    assert menu_repo.get_by_id(first[5]) is None  # 已软删，等值查询不可见


# ============================================================
# M-1: update_user_status 变更状态后失效权限缓存
# ============================================================

def test_update_user_status_invalidates_perm_cache(sqlite_db, monkeypatch):
    import asyncio
    from modules.system.user import service as user_svc
    from modules.system.user.schemas import UserStatusForm
    from modules.system.user.repository import user_repo

    # 预置超管占位记录（id=1），使被测用户获得非超管 ID，避开超管写保护
    user_repo.create(
        username='superadmin', realname='r', nickname='n', gender=1, avatar='a',
        mobile='1', email='a@b.com', province_code='11', city_code='1101',
        district_code='110101', street_code='', address='北京市', status=1, sort=1,
    )
    user = user_repo.create(
        username='u1', realname='r', nickname='n', gender=1, avatar='a',
        mobile='1', email='a@b.com', province_code='11', city_code='1101',
        district_code='110101', street_code='', address='北京市', status=1, sort=1,
    )

    invalidated = []

    async def _recorder(uid):
        invalidated.append(uid)

    monkeypatch.setattr(user_svc, 'invalidate_user_perms', _recorder)

    resp = asyncio.run(user_svc.update_user_status(
        _make_request(), UserStatusForm(id=user.id, status=2)))
    assert json.loads(resp.body)['code'] == 0
    assert invalidated == [user.id]


# ============================================================
# M-2: invalidate_role_perms 清理角色下所有用户缓存（DB 查询走线程池）
# ============================================================

@pytest.mark.asyncio
async def test_invalidate_role_perms_clears_all_user_caches(sqlite_db, monkeypatch):
    import fakeredis.aioredis
    from modules.system.user_role.models import UserRole
    from core.database import get_db_session
    from utils import perm_cache

    # 构造：角色2 → 用户5/6
    get_db_session().add_all([
        UserRole(user_id=5, role_id=2),
        UserRole(user_id=6, role_id=2),
    ])
    sqlite_db.commit()

    fake_redis = fakeredis.aioredis.FakeRedis(decode_responses=True)
    await fake_redis.set('perm:version', '0')
    await fake_redis.set('perm:list:5:0', '["sys:user:add"]')
    await fake_redis.set('perm:menu:6:0', '[]')
    await fake_redis.set('perm:list:99:0', '["sys:user:page"]')  # 无关用户，不应被清
    monkeypatch.setattr(perm_cache, 'get_redis', lambda: fake_redis)

    await perm_cache.invalidate_role_perms(2)

    # 该角色下所有用户的权限/菜单缓存被清（DB 查询经 asyncio.to_thread 执行）
    assert await fake_redis.get('perm:list:5:0') is None
    assert await fake_redis.get('perm:menu:6:0') is None
    # 无关用户缓存不受影响
    assert await fake_redis.get('perm:list:99:0') == '["sys:user:page"]'


# ============================================================
# M-3: 失效缓存失败时 fail-closed 降级递增全局版本号，避免旧权限残留 2 小时
# ============================================================

@pytest.mark.asyncio
async def test_invalidate_user_perms_success_no_fallback(monkeypatch):
    import fakeredis.aioredis
    from utils import perm_cache

    fake_redis = fakeredis.aioredis.FakeRedis(decode_responses=True)
    await fake_redis.set('perm:version', '0')
    await fake_redis.set('perm:list:5:0', '["sys:user:add"]')
    await fake_redis.set('perm:menu:5:0', '[]')
    monkeypatch.setattr(perm_cache, 'get_redis', lambda: fake_redis)

    calls = []

    async def _incr_recorder():
        calls.append('incr')
        return 1

    monkeypatch.setattr(perm_cache, '_incr_version', _incr_recorder)

    await perm_cache.invalidate_user_perms(5)
    # 删除成功：不触发降级版本号递增，缓存键已删除
    assert calls == []
    assert await fake_redis.get('perm:list:5:0') is None
    assert await fake_redis.get('perm:menu:5:0') is None


@pytest.mark.asyncio
async def test_invalidate_user_perms_failure_falls_back_to_version_bump(monkeypatch):
    from utils import perm_cache

    def _raise(*args, **kwargs):
        raise RuntimeError('redis down')

    calls = []

    async def _incr_recorder():
        calls.append('incr')
        return 1

    monkeypatch.setattr(perm_cache, 'get_redis', _raise)
    monkeypatch.setattr(perm_cache, '_incr_version', _incr_recorder)

    await perm_cache.invalidate_user_perms(5)
    # 删除失败 → fail-closed 降级递增全局版本号，孤立旧缓存，杜绝残留
    assert calls == ['incr']


# ============================================================
# M-4: save_content 白名单清洗防存储型 XSS
# ============================================================

def test_save_content_sanitizes_xss():
    from utils.rich_text import save_content

    out = save_content('<p>x<script>alert(1)</script></p>', '标题', 'notice')
    assert '<script>' not in out  # 脚本标签被剥离（文本内容残留为无害纯文本）

    out = save_content('<a href="javascript:alert(1)">x</a>', '标题', 'notice')
    assert 'javascript:' not in out
    assert '<a>' in out

    out = save_content('<img src="x.png" onerror="alert(1)">', '标题', 'notice')
    assert 'onerror' not in out

    out = save_content('<div style="color:red;width:expression(alert(1))">t</div>', '标题', 'notice')
    assert 'expression' not in out

    # 安全样式应保留
    out = save_content('<div style="color:red;text-align:center">ok</div>', '标题', 'notice')
    assert out == '<div style="color:red;text-align:center;">ok</div>'


# ============================================================
# M-5: 职级导入逐行校验，非法行跳过并返回行号原因
# ============================================================

def test_import_level_skips_invalid_rows(sqlite_db, monkeypatch, tmp_path):
    import asyncio
    import json
    from openpyxl import Workbook

    import modules.system.level.service as level_svc
    from modules.system.level.repository import level_repo

    # 构造 Excel：表头 + 2 行合法 + 3 行非法（空名称/重复名/非法排序）
    up = tmp_path / 'uploads'
    up.mkdir(parents=True)
    wb = Workbook()
    ws = wb.worksheets[0]
    ws.append(['职级ID', '职级名称', '职级排序', '职级状态', '创建时间', '更新时间'])
    ws.append([None, '高级', 1, '正常'])
    ws.append([None, '', 2, '正常'])        # 空名称
    ws.append([None, '高级', 3, '正常'])     # 名称重复
    ws.append([None, '特级', 'abc', '正常'])  # 排序非数字
    ws.append([None, '中级', 5, '禁用'])
    xlsx = up / 'import.xlsx'
    wb.save(xlsx)

    monkeypatch.setattr(level_svc, 'FASTAPI_UPLOAD_DIR', str(up))

    async def fake_upload(file, exts):
        from starlette.responses import JSONResponse
        return JSONResponse({"code": 0, "data": {"filePath": "/import.xlsx"}})

    monkeypatch.setattr(level_svc.upload, 'upload_file', fake_upload)

    resp = asyncio.run(level_svc.level_service.import_level(_make_request(), None))
    body = json.loads(resp.body)
    assert body['code'] == 0
    assert body['data']['success'] == 2
    assert body['data']['fail'] == 3
    assert any('第3行' in e for e in body['data']['errors'])
    assert any('第4行' in e for e in body['data']['errors'])
    assert any('第5行' in e for e in body['data']['errors'])
    assert level_repo.get_one(name='高级') is not None
    assert level_repo.get_one(name='中级') is not None
    assert level_repo.count() == 2