NeteaseUrl/tests/test_auth_init_db.py

# -*- coding: utf-8 -*-
"""auth_system.db.init_db 的行为契约测试（沙箱，不触碰生产 data/）。

为什么：init_db（142 行）是站点库的**全部结构演进入口**——建表、列迁移、
邮箱归一化、管理员引导都在这里，且只在启动时跑一次。既有测试只间接覆盖
「全新空库」这一条路径（各沙箱 fixture 调用它），以下路径零覆盖：
旧库列迁移、重复邮箱告警不阻断、邮箱归一化迁移、管理员改名后不误重建。

按惯例：先在未改动的原始实现上跑绿，重构时这批测试即等价性验证器。
"""
import asyncio
import sqlite3
from typing import Any

import pytest

import auth_system as auth

# 旧版 users 表：不带 status/login_seq/recent_provinces/ban_reason 四列
_LEGACY_SCHEMA = '''
    CREATE TABLE users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        username TEXT UNIQUE NOT NULL,
        password_hash TEXT NOT NULL,
        salt TEXT NOT NULL,
        email TEXT DEFAULT '',
        role TEXT DEFAULT 'user',
        duration_seconds INTEGER DEFAULT 0,
        trial_device TEXT DEFAULT '',
        register_ip TEXT DEFAULT '',
        register_ua TEXT DEFAULT '',
        netease_cookie TEXT DEFAULT '',
        ann_seen TEXT DEFAULT '',
        last_active INTEGER DEFAULT 0,
        created_at INTEGER DEFAULT 0,
        last_login INTEGER DEFAULT 0
    );
    CREATE TABLE cards (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        code TEXT UNIQUE NOT NULL,
        minutes INTEGER NOT NULL,
        created_by TEXT DEFAULT '',
        created_at INTEGER DEFAULT 0,
        used_by TEXT DEFAULT '',
        used_at INTEGER DEFAULT 0
    );
    CREATE TABLE email_codes (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        email TEXT NOT NULL,
        code TEXT NOT NULL,
        expires_at INTEGER NOT NULL,
        created_at INTEGER DEFAULT 0
    );
'''


@pytest.fixture()
def dbx(tmp_path, monkeypatch):
    """auth.db 路径指向临时目录。"""
    monkeypatch.setattr(auth.db, 'DATA_DIR', tmp_path)
    monkeypatch.setattr(auth.db, 'DB_PATH', tmp_path / 'site.db')
    monkeypatch.setattr(auth.db, 'BACKUP_DIR', tmp_path / 'backups')
    monkeypatch.setattr(auth.db, 'SECRET_KEY_PATH', tmp_path / 'secret.key')
    return {'root': tmp_path, 'db': tmp_path / 'site.db'}


def _conn(dbx: dict[str, Any]) -> sqlite3.Connection:
    conn = sqlite3.connect(str(dbx['db']))
    conn.row_factory = sqlite3.Row
    return conn


def _cols(dbx: dict[str, Any], table: str) -> set[str]:
    conn = _conn(dbx)
    try:
        return {r['name'] for r in conn.execute(f'PRAGMA table_info({table})')}
    finally:
        conn.close()


def _tables(dbx: dict[str, Any]) -> set[str]:
    conn = _conn(dbx)
    try:
        return {r['name'] for r in
                conn.execute("SELECT name FROM sqlite_master WHERE type='table'")}
    finally:
        conn.close()


# ============ 全新建库 ============

def test_fresh_db_creates_tables_and_admin(dbx, capsys):
    """全新库：全部表就位、单行配置有种子行、初始管理员 role=admin 且余额 -1。"""
    asyncio.run(auth.init_db())

    tables = _tables(dbx)
    for t in ('users', 'email_codes', 'cards', 'logs',
              'smtp_config', 'settings', 'announcements', 'billing_ledger'):
        assert t in tables, f'缺表 {t}'

    conn = _conn(dbx)
    try:
        smtp = conn.execute('SELECT port FROM smtp_config WHERE id = 1').fetchone()
        assert smtp is not None and smtp['port'] == 587
        ann = conn.execute('SELECT id FROM announcements WHERE id = 1').fetchone()
        assert ann is not None
        admin = conn.execute("SELECT username, role, duration_seconds FROM users "
                             "WHERE role = 'admin'").fetchone()
        assert admin is not None
        assert admin['username'] == 'admin'
        assert admin['duration_seconds'] == -1, '初始管理员余额为 -1（占位，不计费）'
    finally:
        conn.close()
    assert '[初始化]' in capsys.readouterr().out, '初始密码提示必须打到 stdout 一次'


def test_second_run_is_idempotent(dbx):
    """二次初始化：不报错、不重建管理员、已有数据原样保留。"""
    asyncio.run(auth.init_db())
    conn = _conn(dbx)
    try:
        conn.execute("UPDATE users SET duration_seconds = 42 WHERE role = 'admin'")
        conn.commit()
    finally:
        conn.close()

    asyncio.run(auth.init_db())   # 不应抛

    conn = _conn(dbx)
    try:
        admins = conn.execute("SELECT id, duration_seconds FROM users "
                              "WHERE role = 'admin'").fetchall()
        assert len(admins) == 1, f'管理员不得重建，实际 {len(admins)} 个'
        assert admins[0]['duration_seconds'] == 42, '已有数据不得被重置'
    finally:
        conn.close()


# ============ 旧库列迁移 ============

def test_legacy_db_columns_get_added(dbx):
    """旧版库（缺 status/login_seq/... 列）：init_db 补齐列，既有行原样保留。"""
    conn = sqlite3.connect(str(dbx['db']))
    conn.executescript(_LEGACY_SCHEMA)
    conn.execute("INSERT INTO users (username, password_hash, salt, email, role, "
                 "duration_seconds) VALUES ('olduser', 'h', 's', 'o@x.com', 'user', 7)")
    conn.commit()
    conn.close()

    asyncio.run(auth.init_db())   # 不应抛

    assert {'status', 'login_seq', 'recent_provinces', 'ban_reason'} <= _cols(dbx, 'users')
    assert {'note', 'expire_at', 'disabled'} <= _cols(dbx, 'cards')
    assert 'purpose' in _cols(dbx, 'email_codes')

    conn = _conn(dbx)
    try:
        row = conn.execute("SELECT email, duration_seconds, status FROM users "
                           "WHERE username = 'olduser'").fetchone()
        assert row is not None, '旧用户行不得丢失'
        assert row['email'] == 'o@x.com'
        assert row['duration_seconds'] == 7
        assert row['status'] == 1, '新列按 DEFAULT 填充'
    finally:
        conn.close()


def test_renamed_admin_not_recreated(dbx):
    """管理员改名（admin→qfbsh）后重启：按「无管理员角色」判断，不得再建新 admin。"""
    asyncio.run(auth.init_db())
    conn = _conn(dbx)
    try:
        conn.execute("UPDATE users SET username = 'qfbsh' WHERE role = 'admin'")
        conn.commit()
    finally:
        conn.close()

    asyncio.run(auth.init_db())

    conn = _conn(dbx)
    try:
        admins = conn.execute("SELECT username FROM users WHERE role = 'admin'").fetchall()
        assert len(admins) == 1, f'管理员应为 1 个，实际 {len(admins)}'
        assert admins[0]['username'] == 'qfbsh', '不得重建出新的 admin 账号'
        dup = conn.execute("SELECT id FROM users WHERE username = 'admin'").fetchone()
        assert dup is None
    finally:
        conn.close()


# ============ 邮箱唯一索引与归一化 ============

def test_duplicate_nonempty_email_warns_and_continues(dbx, capsys):
    """存量重复非空邮箱：告警不阻断启动，两行都保留，唯一索引未创建。"""
    conn = sqlite3.connect(str(dbx['db']))
    conn.executescript(_LEGACY_SCHEMA)
    conn.execute("INSERT INTO users (username, password_hash, salt, email) "
                 "VALUES ('a1', 'h', 's', 'same@x.com')")
    conn.execute("INSERT INTO users (username, password_hash, salt, email) "
                 "VALUES ('a2', 'h', 's', 'same@x.com')")
    conn.commit()
    conn.close()

    asyncio.run(auth.init_db())   # 不应抛（查重兜底在注册/建号路由，此处只告警）

    out = capsys.readouterr().out
    assert '重复的非空邮箱' in out, f'应打印人工清理提示，实际输出: {out!r}'
    conn = _conn(dbx)
    try:
        n = conn.execute("SELECT COUNT(*) AS n FROM users WHERE email = 'same@x.com'"
                         ).fetchone()['n']
        assert n == 2, '存量行不得被动删改'
        idx = conn.execute("SELECT name FROM sqlite_master WHERE type='index' "
                           "AND name = 'idx_users_email_nonempty'").fetchone()
        assert idx is None, '唯一索引创建失败时不应存在半成品'
    finally:
        conn.close()


def test_email_normalization_migrates_plus_tag(dbx):
    """存量 +tag 邮箱：init_db 逐行归一化落库（lower + 剥 +tag）。"""
    conn = sqlite3.connect(str(dbx['db']))
    conn.executescript(_LEGACY_SCHEMA)
    conn.execute("INSERT INTO users (username, password_hash, salt, email) "
                 "VALUES ('u1', 'h', 's', 'Foo+Tag@Gmail.com')")
    conn.commit()
    conn.close()

    asyncio.run(auth.init_db())

    conn = _conn(dbx)
    try:
        row = conn.execute("SELECT email FROM users WHERE username = 'u1'").fetchone()
        assert row is not None and row['email'] == 'foo@gmail.com', row
    finally:
        conn.close()


def test_email_normalization_collision_keeps_original(dbx, capsys):
    """归一化后与他人撞唯一索引：保留原值跳过，告警但不抛。"""
    conn = sqlite3.connect(str(dbx['db']))
    conn.executescript(_LEGACY_SCHEMA)
    conn.execute("INSERT INTO users (username, password_hash, salt, email) "
                 "VALUES ('u1', 'h', 's', 'a+x@gmail.com')")
    conn.execute("INSERT INTO users (username, password_hash, salt, email) "
                 "VALUES ('u2', 'h', 's', 'a@gmail.com')")
    conn.commit()
    conn.close()

    asyncio.run(auth.init_db())   # 不应抛

    out = capsys.readouterr().out
    assert '重复' in out, f'应打印撞行告警，实际输出: {out!r}'
    conn = _conn(dbx)
    try:
        e1 = conn.execute("SELECT email FROM users WHERE username = 'u1'").fetchone()
        e2 = conn.execute("SELECT email FROM users WHERE username = 'u2'").fetchone()
        assert e1['email'] == 'a+x@gmail.com', '撞行的行保留原值'
        assert e2['email'] == 'a@gmail.com'
    finally:
        conn.close()