# -*- 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