Archived
refactor: init_db 四段拆解（142 行 / 圈 15 / 嵌套 4 → 14 行 / 圈 2 / 嵌套 1）
为什么：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）。
This commit is contained in:
snowy-yang committed
1 parent
1c6b17f718
commit
37f8d77405
2 files changed
+424
-133
No files matched your search
@@ -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()
|
||||
Reference in new issue
Block a user