743 lines
27 KiB
Python
743 lines
27 KiB
Python
# -*- 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
|