Archived
feat(routes): 迁移账号/全局歌单端点 + 鉴权依赖测试
迁移路由(/api/v1/*,与 Flask 同端点逐字段对齐): - account/vip:admin/user 双路径 + 5 分钟缓存(降级结果 30s), VIP 探测单飞锁与 Flask 同源跨栈共享;未登录可访问(同 Flask) - global/playlists、global/playlist:管理员闸(require_admin,403 对齐) - web/deps.py 新增 require_admin;_resolve_user_cookies 支持可选用户(匿名审计保底) 测试:46/46(新增 10 项鉴权依赖:有效/管理员/封禁/禁用/顶号/伪造 Cookie、 401/403 闸);ruff All checks passed;pyright 0 errors。 契约奇偶校验:16 项全 OK(新增 account:vip、global:playlists/playlist、 非管理员 403);uvicorn 实跑双栈 account/vip 均 200、未授权歌单 403。
This commit is contained in:
1 parent
a2758aa0da
commit
6c3181341b
5 files changed
+287
-2
No files matched your search
@@ -0,0 +1,90 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""FastAPI 鉴权依赖测试:会话解析、封禁/顶号拒绝、登录与管理员闸。"""
|
||||
import sqlite3
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
import core.db as db
|
||||
from core import session as cs
|
||||
from web.deps import SiteUser, current_user, require_admin, require_login
|
||||
from web.errors import AppError
|
||||
|
||||
SECRET = 'deps-test-secret'
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def site(tmp_path, monkeypatch):
|
||||
path = tmp_path / 'site.db'
|
||||
conn = sqlite3.connect(str(path))
|
||||
conn.executescript("""
|
||||
CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT, role TEXT DEFAULT 'user',
|
||||
status INTEGER DEFAULT 1, login_seq INTEGER DEFAULT 0,
|
||||
duration_seconds INTEGER DEFAULT 0, ban_reason TEXT DEFAULT '');
|
||||
INSERT INTO users (id, username, role, status, login_seq, duration_seconds)
|
||||
VALUES (1,'qfbsh','admin',1,376,-1),
|
||||
(26,'normal','user',1,5,97),
|
||||
(27,'banned','user',2,1,0),
|
||||
(28,'disabled','user',0,1,0);
|
||||
""")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
monkeypatch.setattr(db, 'DATA_DIR', tmp_path)
|
||||
monkeypatch.setattr(db, 'DB_PATH', path)
|
||||
monkeypatch.setattr('web.deps.secret_key', lambda: SECRET)
|
||||
return path
|
||||
|
||||
|
||||
def _cookie(uid, seq, ts=None):
|
||||
return cs.dumps(SECRET, {'site_uid': uid, 'site_uid_seq': seq,
|
||||
'site_uid_ts': ts or int(time.time())})
|
||||
|
||||
|
||||
async def test_valid_user(site):
|
||||
u = await current_user(_cookie(26, 5))
|
||||
assert u is not None and u.id == 26 and u.username == 'normal'
|
||||
assert not u.is_admin and not u.unlimited
|
||||
|
||||
|
||||
async def test_admin_unlimited(site):
|
||||
u = await current_user(_cookie(1, 376))
|
||||
assert u is not None and u.is_admin and u.unlimited
|
||||
|
||||
|
||||
async def test_missing_cookie(site):
|
||||
assert await current_user(None) is None
|
||||
|
||||
|
||||
async def test_banned_rejected(site):
|
||||
assert await current_user(_cookie(27, 1)) is None
|
||||
|
||||
|
||||
async def test_disabled_rejected(site):
|
||||
assert await current_user(_cookie(28, 1)) is None
|
||||
|
||||
|
||||
async def test_stale_login_seq_rejected(site):
|
||||
"""顶号:会话版本与库不一致 → 视为未登录。"""
|
||||
assert await current_user(_cookie(26, 4)) is None
|
||||
|
||||
|
||||
async def test_forged_cookie_rejected(site):
|
||||
assert await current_user('not-a-valid-cookie') is None
|
||||
|
||||
|
||||
async def test_require_login_raises(site):
|
||||
with pytest.raises(AppError) as ei:
|
||||
await require_login(None)
|
||||
assert ei.value.status_code == 401
|
||||
|
||||
|
||||
async def test_require_admin_forbidden_for_user(site):
|
||||
u = SiteUser(id=26, username='normal', role='user', credits=97)
|
||||
with pytest.raises(AppError) as ei:
|
||||
await require_admin(u)
|
||||
assert ei.value.status_code == 403
|
||||
|
||||
|
||||
async def test_require_admin_ok(site):
|
||||
u = SiteUser(id=1, username='qfbsh', role='admin', credits=-1)
|
||||
assert (await require_admin(u)).is_admin
|
||||
Reference in new issue
Block a user