NeteaseUrl/tests/test_deps.py

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


async def _fake_secret_key() -> str:
    """web.deps.secret_key 已异步化：替身须为协程函数（否则 await 处 TypeError）。"""
    return 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', _fake_secret_key)
    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_expired_cookie_rejected(site, monkeypatch):
    """超龄会话 Cookie 由**服务端**拒绝（不只是浏览器层面的 max_age）。

    Cookie 的 max_age 只是浏览器行为：被保存/复制的旧 Cookie 仍带着有效签名，
    服务端不校验签发时间就等于登录态永不过期。

    构造方式：用 monkeypatch.context() 把时钟拨回 SESSION_MAX_AGE 之前签发
    （itsdangerous 的签名时间戳取自 dumps 时刻），再以当前时间解析。
    必须用 context 而非 monkeypatch.undo()——后者会连带撤销 site fixture 的设置。
    """
    past = time.time() - cs.SESSION_MAX_AGE - 60
    with monkeypatch.context() as m:
        m.setattr(time, 'time', lambda: past)
        stale = _cookie(26, 5)
    assert await current_user(stale) is None, '超龄会话必须被服务端拒绝'

    fresh = await current_user(_cookie(26, 5))
    assert fresh is not None and fresh.id == 26, '有效期内不受影响'


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