From 37f8d77405faed649683dd7ad8a8b77dc059c52e Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sun, 27 Sep 2026 09:36:31 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20init=5Fdb=20=E5=9B=9B=E6=AE=B5?= =?UTF-8?q?=E6=8B=86=E8=A7=A3=EF=BC=88142=20=E8=A1=8C=20/=20=E5=9C=88=2015?= =?UTF-8?q?=20/=20=E5=B5=8C=E5=A5=97=204=20=E2=86=92=2014=20=E8=A1=8C=20/?= =?UTF-8?q?=20=E5=9C=88=202=20/=20=E5=B5=8C=E5=A5=97=201=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：init_db 是站点库全部结构演进的唯一入口，但 142 行单函数把 「建表 / 列迁移 / 邮箱唯一索引 / 邮箱归一化 / 管理员引导」五件事 揉在一个 with-try 里，任何一段改动都要在 140 行上下文里核对边界。 拆出五个成员（全部接收 conn，调用顺序 = 原执行顺序，单连接单 commit 不变）： - _SCHEMA_SQL 常量 + _create_schema（建表 + 种子行，纯数据上移） - _apply_column_migrations（ALTER 循环；非 duplicate column 的 OperationalError 仍然抛出，禁止静默带病初始化） - _ensure_email_unique_index（重复邮箱仅告警不阻断，原文案保留） - _migrate_user_emails（撞唯一索引行保留原值，原文案保留） - _ensure_initial_admin（按 role 判断而非用户名，防改名后误重建） 先写测试后动代码：新增 tests/test_auth_init_db.py 7 例，覆盖此前 零测试的路径——旧库列迁移、二次初始化幂等、改名管理员不重建、 存量重复邮箱告警不阻断、+tag 邮箱归一化迁移、归一化撞行保留原值。 7 例在未改动实现上一次跑绿，重构后即等价性验证。 门禁：ruff 全过 · pyright 0/0 · pytest 319 passed（312→319）。 --- auth_system/db.py | 297 ++++++++++++++++++++----------------- tests/test_auth_init_db.py | 260 ++++++++++++++++++++++++++++++++ 2 files changed, 424 insertions(+), 133 deletions(-) create mode 100644 tests/test_auth_init_db.py diff --git a/auth_system/db.py b/auth_system/db.py index 1754d32..2e7afe8 100644 --- a/auth_system/db.py +++ b/auth_system/db.py @@ -77,145 +77,176 @@ def _x(sql: str, params: tuple = ()) -> int: conn.close() +_SCHEMA_SQL = ''' + CREATE TABLE IF NOT EXISTS 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 IF NOT EXISTS 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, + purpose TEXT DEFAULT 'register' + ); + CREATE TABLE IF NOT EXISTS 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 IF NOT EXISTS logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + type TEXT DEFAULT 'success', + username TEXT DEFAULT '', + ip TEXT DEFAULT '', + action TEXT DEFAULT '', + detail TEXT DEFAULT '', + time TEXT DEFAULT '' + ); +''' + + +def _create_schema(conn: sqlite3.Connection) -> None: + """建表 + 单行配置种子行（幂等：全部 IF NOT EXISTS / INSERT OR IGNORE）。""" + conn.executescript(_SCHEMA_SQL) + # 单行配置表 + conn.execute('''CREATE TABLE IF NOT EXISTS smtp_config ( + id INTEGER PRIMARY KEY CHECK (id = 1), + host TEXT DEFAULT '', port INTEGER DEFAULT 587, + sender TEXT DEFAULT '', password TEXT DEFAULT '', + updated_at INTEGER DEFAULT 0 + )''') + conn.execute('INSERT OR IGNORE INTO smtp_config (id, port) VALUES (1, 587)') + # 键值设置表 + conn.execute('''CREATE TABLE IF NOT EXISTS settings ( + key TEXT PRIMARY KEY, value TEXT DEFAULT '' + )''') + conn.execute('''CREATE TABLE IF NOT EXISTS announcements ( + id INTEGER PRIMARY KEY CHECK (id = 1), + title TEXT DEFAULT '', content TEXT DEFAULT '', + popup_mode TEXT DEFAULT 'off', + updated_at INTEGER DEFAULT 0 + )''') + conn.execute('INSERT OR IGNORE INTO announcements (id) VALUES (1)') + # 计费账本（任务实扣/交付/退款/终态的持久化唯一记录，结算幂等） + conn.execute('''CREATE TABLE IF NOT EXISTS billing_ledger ( + task_id TEXT PRIMARY KEY, + user_id INTEGER NOT NULL, + reserved INTEGER DEFAULT 0, + delivered INTEGER DEFAULT 0, + refunded INTEGER DEFAULT 0, + status TEXT DEFAULT 'open', + created_at INTEGER DEFAULT 0, + settled_at INTEGER DEFAULT 0 + )''') + + +def _apply_column_migrations(conn: sqlite3.Connection) -> None: + """轻量迁移：新增列。仅"列已存在"可忽略（其余迁移失败必须暴露，不能吞掉）""" + _already: set[str] = set() + for trow in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall(): + for crow in conn.execute(f"PRAGMA table_info({trow['name']})").fetchall(): + _already.add(f"{trow['name']}.{crow['name']}") + for alter_sql, tbl_col in ( + ('ALTER TABLE users ADD COLUMN status INTEGER DEFAULT 1', 'users.status'), + ('ALTER TABLE users ADD COLUMN login_seq INTEGER DEFAULT 0', 'users.login_seq'), + ('ALTER TABLE users ADD COLUMN recent_provinces TEXT DEFAULT \'[]\'', 'users.recent_provinces'), + ('ALTER TABLE users ADD COLUMN ban_reason TEXT DEFAULT \'\'', 'users.ban_reason'), + ('ALTER TABLE cards ADD COLUMN note TEXT DEFAULT \'\'', 'cards.note'), + ('ALTER TABLE cards ADD COLUMN expire_at INTEGER DEFAULT 0', 'cards.expire_at'), + ('ALTER TABLE cards ADD COLUMN disabled INTEGER DEFAULT 0', 'cards.disabled'), + ('ALTER TABLE email_codes ADD COLUMN purpose TEXT DEFAULT \'register\'', 'email_codes.purpose'), + ): + if tbl_col in _already: + continue # 幂等：列已存在 + try: + conn.execute(alter_sql) + except sqlite3.OperationalError as e: + if 'duplicate column' not in str(e).lower(): + # 磁盘/锁/结构等真实故障：抛出，禁止静默带病初始化 + raise + + +def _ensure_email_unique_index(conn: sqlite3.Connection) -> None: + """非空邮箱唯一（部分索引，不约束历史空串）。 + + 存量库若已有重复非空邮箱，索引创建失败仅告警不阻断启动 + （查重逻辑在注册/建号路由另有兜底）。 + """ + try: + conn.execute('CREATE UNIQUE INDEX IF NOT EXISTS idx_users_email_nonempty ' + 'ON users(email) WHERE email != \'\'') + except sqlite3.IntegrityError: + print('[警告] users.email 存在重复的非空邮箱，唯一索引未创建（请人工清理后重启）') + + +def _migrate_user_emails(conn: sqlite3.Connection) -> None: + """users.email 口径统一为归一化（lower + 剥 Gmail +tag）——与 + get_user_by_email/找回/邮箱登录的查询口径对齐。逐行迁移，归一化后与他 + 人撞唯一索引的行保留原值跳过（与"存量重复仅告警"同口径）。幂等：已 + 归一化的行零写入。 + """ + for urow in conn.execute("SELECT id, email FROM users WHERE email != ''").fetchall(): + fixed = _normalize_email(urow['email']) + if not fixed or fixed == urow['email']: + continue + try: + conn.execute('UPDATE users SET email = ? WHERE id = ?', (fixed, urow['id'])) + except sqlite3.IntegrityError: + print(f"[警告] 用户 {urow['id']} 的邮箱归一化后与他人重复，保留原值") + + +def _ensure_initial_admin(conn: sqlite3.Connection) -> None: + """初始管理员（新装不再使用公开默认口令，随机生成并写入日志一次）。 + + 引导条件改为"不存在任何管理员角色"——按用户名查会在 + 管理员改名（如 admin→qfbsh）后误判缺失，重启即重建出一个新的 admin 账号。 + """ + row = conn.execute("SELECT id FROM users WHERE role = 'admin' LIMIT 1").fetchone() + if row: + return + initial_pass = secrets.token_urlsafe(12) + conn.execute( + 'INSERT INTO users (username, password_hash, salt, role, duration_seconds, created_at) ' + 'VALUES (?, ?, ?, ?, -1, ?)', + (DEFAULT_ADMIN_USER, _hash_new(initial_pass), '', + 'admin', int(time.time()))) + # ——非 ASCII 输出须防 GBK 重定向下 UnicodeEncodeError + try: + print(f'[初始化] 已创建管理员账号 admin，初始密码：{initial_pass}（请立即登录并修改，本提示不再显示）') + except UnicodeEncodeError: + print('[init] admin account created, initial password printed above failed to encode') + + def init_db() -> None: """初始化表结构与默认数据""" DATA_DIR.mkdir(parents=True, exist_ok=True) with _DB_LOCK: conn = get_conn() try: - conn.executescript(''' - CREATE TABLE IF NOT EXISTS 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 IF NOT EXISTS 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, - purpose TEXT DEFAULT 'register' - ); - CREATE TABLE IF NOT EXISTS 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 IF NOT EXISTS logs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - type TEXT DEFAULT 'success', - username TEXT DEFAULT '', - ip TEXT DEFAULT '', - action TEXT DEFAULT '', - detail TEXT DEFAULT '', - time TEXT DEFAULT '' - ); - ''') - # 单行配置表 - conn.execute('''CREATE TABLE IF NOT EXISTS smtp_config ( - id INTEGER PRIMARY KEY CHECK (id = 1), - host TEXT DEFAULT '', port INTEGER DEFAULT 587, - sender TEXT DEFAULT '', password TEXT DEFAULT '', - updated_at INTEGER DEFAULT 0 - )''') - conn.execute('INSERT OR IGNORE INTO smtp_config (id, port) VALUES (1, 587)') - # 键值设置表 - conn.execute('''CREATE TABLE IF NOT EXISTS settings ( - key TEXT PRIMARY KEY, value TEXT DEFAULT '' - )''') - conn.execute('''CREATE TABLE IF NOT EXISTS announcements ( - id INTEGER PRIMARY KEY CHECK (id = 1), - title TEXT DEFAULT '', content TEXT DEFAULT '', - popup_mode TEXT DEFAULT 'off', - updated_at INTEGER DEFAULT 0 - )''') - conn.execute('INSERT OR IGNORE INTO announcements (id) VALUES (1)') - # 计费账本（任务实扣/交付/退款/终态的持久化唯一记录，结算幂等） - conn.execute('''CREATE TABLE IF NOT EXISTS billing_ledger ( - task_id TEXT PRIMARY KEY, - user_id INTEGER NOT NULL, - reserved INTEGER DEFAULT 0, - delivered INTEGER DEFAULT 0, - refunded INTEGER DEFAULT 0, - status TEXT DEFAULT 'open', - created_at INTEGER DEFAULT 0, - settled_at INTEGER DEFAULT 0 - )''') - # 轻量迁移：新增列。仅"列已存在"可忽略（其余迁移失败必须暴露，不能吞掉） - _already = set() - for trow in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall(): - for crow in conn.execute(f"PRAGMA table_info({trow['name']})").fetchall(): - _already.add(f"{trow['name']}.{crow['name']}") - for alter_sql, tbl_col in ( - ('ALTER TABLE users ADD COLUMN status INTEGER DEFAULT 1', 'users.status'), - ('ALTER TABLE users ADD COLUMN login_seq INTEGER DEFAULT 0', 'users.login_seq'), - ('ALTER TABLE users ADD COLUMN recent_provinces TEXT DEFAULT \'[]\'', 'users.recent_provinces'), - ('ALTER TABLE users ADD COLUMN ban_reason TEXT DEFAULT \'\'', 'users.ban_reason'), - ('ALTER TABLE cards ADD COLUMN note TEXT DEFAULT \'\'', 'cards.note'), - ('ALTER TABLE cards ADD COLUMN expire_at INTEGER DEFAULT 0', 'cards.expire_at'), - ('ALTER TABLE cards ADD COLUMN disabled INTEGER DEFAULT 0', 'cards.disabled'), - ('ALTER TABLE email_codes ADD COLUMN purpose TEXT DEFAULT \'register\'', 'email_codes.purpose'), - ): - if tbl_col in _already: - continue # 幂等：列已存在 - try: - conn.execute(alter_sql) - except sqlite3.OperationalError as e: - if 'duplicate column' not in str(e).lower(): - # 磁盘/锁/结构等真实故障：抛出，禁止静默带病初始化 - raise - # 非空邮箱唯一（部分索引，不约束历史空串）。存量库若已有重复非空邮箱， - # 索引创建失败仅告警不阻断启动（查重逻辑在注册/建号路由另有兜底） - try: - conn.execute('CREATE UNIQUE INDEX IF NOT EXISTS idx_users_email_nonempty ' - 'ON users(email) WHERE email != \'\'') - except sqlite3.IntegrityError: - print('[警告] users.email 存在重复的非空邮箱，唯一索引未创建（请人工清理后重启）') - # users.email 口径统一为归一化（lower + 剥 Gmail +tag）——与 - # get_user_by_email/找回/邮箱登录的查询口径对齐。逐行迁移，归一化后与他 - # 人撞唯一索引的行保留原值跳过（与"存量重复仅告警"同口径）。幂等：已 - # 归一化的行零写入 - for urow in conn.execute("SELECT id, email FROM users WHERE email != ''").fetchall(): - fixed = _normalize_email(urow['email']) - if not fixed or fixed == urow['email']: - continue - try: - conn.execute('UPDATE users SET email = ? WHERE id = ?', (fixed, urow['id'])) - except sqlite3.IntegrityError: - print(f"[警告] 用户 {urow['id']} 的邮箱归一化后与他人重复，保留原值") - # 初始管理员（新装不再使用公开默认口令，随机生成并写入日志一次） - # 引导条件改为"不存在任何管理员角色"——按用户名查会在 - # 管理员改名（如 admin→qfbsh）后误判缺失，重启即重建出一个新的 admin 账号 - row = conn.execute("SELECT id FROM users WHERE role = 'admin' LIMIT 1").fetchone() - if not row: - initial_pass = secrets.token_urlsafe(12) - conn.execute( - 'INSERT INTO users (username, password_hash, salt, role, duration_seconds, created_at) ' - 'VALUES (?, ?, ?, ?, -1, ?)', - (DEFAULT_ADMIN_USER, _hash_new(initial_pass), '', - 'admin', int(time.time()))) - # ——非 ASCII 输出须防 GBK 重定向下 UnicodeEncodeError - try: - print(f'[初始化] 已创建管理员账号 admin，初始密码：{initial_pass}（请立即登录并修改，本提示不再显示）') - except UnicodeEncodeError: - print('[init] admin account created, initial password printed above failed to encode') + _create_schema(conn) + _apply_column_migrations(conn) + _ensure_email_unique_index(conn) + _migrate_user_emails(conn) + _ensure_initial_admin(conn) conn.commit() finally: conn.close() diff --git a/tests/test_auth_init_db.py b/tests/test_auth_init_db.py new file mode 100644 index 0000000..0275856 --- /dev/null +++ b/tests/test_auth_init_db.py @@ -0,0 +1,260 @@ +# -*- coding: utf-8 -*- +"""auth_system.db.init_db 的行为契约测试（沙箱，不触碰生产 data/）。 + +为什么：init_db（142 行）是站点库的**全部结构演进入口**——建表、列迁移、 +邮箱归一化、管理员引导都在这里，且只在启动时跑一次。既有测试只间接覆盖 +「全新空库」这一条路径（各沙箱 fixture 调用它），以下路径零覆盖： +旧库列迁移、重复邮箱告警不阻断、邮箱归一化迁移、管理员改名后不误重建。 + +按惯例：先在未改动的原始实现上跑绿，重构时这批测试即等价性验证器。 +""" +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。""" + 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): + """二次初始化：不报错、不重建管理员、已有数据原样保留。""" + auth.init_db() + conn = _conn(dbx) + try: + conn.execute("UPDATE users SET duration_seconds = 42 WHERE role = 'admin'") + conn.commit() + finally: + conn.close() + + 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() + + 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。""" + auth.init_db() + conn = _conn(dbx) + try: + conn.execute("UPDATE users SET username = 'qfbsh' WHERE role = 'admin'") + conn.commit() + finally: + conn.close() + + 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() + + 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() + + 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() + + 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()