NeteaseUrl/tests/test_login_flow.py

# -*- coding: utf-8 -*-
"""login 端点失败路径契约（超长密码 / 邮箱登录 / 锁定 / 封禁禁用 / 2FA 前置 / 异常映射）。

test_auth_routes.py 覆盖成功主路径；本文件钉住 login 的各拒绝分支，
作为 login 拆解的等价性验证器。
"""
import pytest
from fastapi.testclient import TestClient

import auth_system as auth
import core.db as db

SECRET = 'login-flow-test-secret'
USER_PW = 'UserPass123'
ADMIN_PW = 'AdminPass123'


@pytest.fixture()
def env(tmp_path, monkeypatch):
    db_path = tmp_path / 'site.db'
    monkeypatch.setattr(auth.db, 'DATA_DIR', tmp_path)
    monkeypatch.setattr(auth.db, 'DB_PATH', db_path)
    monkeypatch.setattr(auth.db, 'BACKUP_DIR', tmp_path / 'backups')
    monkeypatch.setattr(auth.db, 'SECRET_KEY_PATH', tmp_path / 'secret.key')
    monkeypatch.setattr(db, 'DATA_DIR', tmp_path)
    monkeypatch.setattr(db, 'DB_PATH', db_path)
    auth.init_db()

    conn = auth.get_conn()
    conn.execute('INSERT INTO users (username, password_hash, salt, role, duration_seconds, '
                 'status, login_seq, created_at, email) VALUES (?,?,?,?,?,1,0,?,?)',
                 ('mailuser', auth._hash_new(USER_PW), '', 'user', 97, 1730000000,
                  'mail@example.com'))
    conn.execute('INSERT INTO users (username, password_hash, salt, role, duration_seconds, '
                 'status, login_seq, created_at) VALUES (?,?,?,?,?,1,0,?)',
                 ('bossuser', auth._hash_new(ADMIN_PW), '', 'admin', -1, 1730000000))
    conn.commit()
    conn.close()

    monkeypatch.setattr('web.deps.secret_key', lambda: SECRET)

    import web.auth_state as auth_state
    auth_state._LOGIN_FAILS.clear()
    auth_state._LOGIN_FAIL_LOGGED.clear()
    auth_state._ADMIN_2FA.clear()

    from web.app import create_app
    return TestClient(create_app())


def _logs(conn, action: str) -> list[tuple[str, str]]:
    rows = conn.execute('SELECT type, detail FROM logs WHERE action = ?', (action,)).fetchall()
    return [(r['type'], r['detail']) for r in rows]


def test_overlong_password_rejected_without_pbkdf2(env, monkeypatch):
    """超长密码（>128）：不碰 PBKDF2，直接按密码错误拒绝。"""
    def _boom(*a, **k):
        raise AssertionError('verify_login must not be called')

    monkeypatch.setattr(auth, 'verify_login', _boom)
    r = env.post('/auth/login', json={'username': 'mailuser', 'password': 'x' * 129})
    assert r.status_code == 400
    assert r.json()['message'] == '用户名或密码错误'


def test_email_login_resolves_account(env):
    """邮箱登录：按归一化邮箱解析真实账号（大小写不敏感）。"""
    r = env.post('/auth/login', json={'username': 'MAIL@Example.com', 'password': USER_PW})
    assert r.status_code == 200
    assert r.json()['success'] is True
    assert 'session' in r.cookies


def test_unknown_email_locks_by_normalized_email(env):
    """邮箱登录失败锁定：键用归一化邮箱（不含大小写变体绕过）。"""
    for _ in range(5):
        r = env.post('/auth/login', json={'username': 'Ghost@Example.com', 'password': 'nope'})
        assert r.json()['message'] == '用户名或密码错误'
    r = env.post('/auth/login', json={'username': 'ghost@example.com', 'password': USER_PW})
    assert '尝试过于频繁' in r.json()['message']

    import web.auth_state as auth_state
    assert 'ghost@example.com' in auth_state._LOGIN_FAILS


def test_locked_log_written_once(env):
    """锁定周期内重复被封尝试只落一条 fail 日志。"""
    for _ in range(5):
        env.post('/auth/login', json={'username': 'mailuser', 'password': 'nope'})
    for _ in range(3):
        env.post('/auth/login', json={'username': 'mailuser', 'password': USER_PW})
    conn = auth.get_conn()
    locked_logs = [d for t, d in _logs(conn, '登录') if d == '失败次数过多，临时锁定']
    conn.close()
    assert len(locked_logs) == 1


def test_banned_account_message(env, monkeypatch):
    """封禁账号（status=2）：文案带原因 + 失败日志。"""
    conn = auth.get_conn()
    conn.execute("UPDATE users SET status=2, ban_reason='刷单' WHERE username='mailuser'")
    conn.commit()
    conn.close()

    r = env.post('/auth/login', json={'username': 'mailuser', 'password': USER_PW})
    assert r.json()['message'] == '账号已被封禁：刷单，请联系管理员'
    conn = auth.get_conn()
    assert ('fail', '封禁账号尝试登录: 刷单') in _logs(conn, '登录')
    conn.close()


def test_disabled_account_message(env):
    """禁用账号（status=0）：固定文案，无原因拼接。"""
    conn = auth.get_conn()
    conn.execute("UPDATE users SET status=0 WHERE username='mailuser'")
    conn.commit()
    conn.close()

    r = env.post('/auth/login', json={'username': 'mailuser', 'password': USER_PW})
    assert r.json()['message'] == '账号已被禁用，请联系管理员'


def test_admin_without_email_blocked_fail_closed(env):
    """管理员未绑定邮箱：fail-closed，不签发票据。"""
    r = env.post('/auth/login', json={'username': 'bossuser', 'password': ADMIN_PW})
    assert r.json()['message'] == '管理员账号未绑定邮箱，请联系管理员先完成绑定'
    assert 'session' not in r.cookies
    conn = auth.get_conn()
    assert ('fail', '未绑定邮箱，两步验证不可用') in _logs(conn, '管理员登录')
    conn.close()


def test_geo_ban_bumps_seq_again(env, monkeypatch):
    """异地多省封禁：login_seq 再递增一次（已签发中的会话全部顶下线）。"""
    def _banned(user, ip):
        return True, {'province': 'X'}

    monkeypatch.setattr(auth, 'record_login_location', _banned)
    r = env.post('/auth/login', json={'username': 'mailuser', 'password': USER_PW})
    assert r.json()['message'] == '检测到账号在多个省份异常登录，已自动封禁，请联系管理员'

    conn = auth.get_conn()
    seq = conn.execute("SELECT login_seq FROM users WHERE username='mailuser'").fetchone()[0]
    conn.close()
    assert seq == 2  # 会话递增 + 封禁再递增


def test_unexpected_error_is_500(env, monkeypatch):
    """非预期异常：泛化文案 + HTTP 500。"""
    def _boom(*a, **k):
        raise RuntimeError('db busy')

    monkeypatch.setattr(auth, 'verify_login', _boom)
    r = env.post('/auth/login', json={'username': 'mailuser', 'password': USER_PW})
    assert r.status_code == 500
    assert r.json()['message'] == '登录失败，请稍后重试'