Archived
snowy-yang
70ca21974b
为什么：auth_system 的 11 个子模块此前构成一个强连通分量——db.py 与 backup.py
为了「动态读取路径常量」而在模块级 `import auth_system`（包自身），而包 __init__
又再导出这两个子模块的全部符号，互相指成环。
这个环当前并不崩（实测 8/8 种导入姿势全通过），它的代价是**双绑定**：
`auth_system.DB_PATH is auth_system.db.DB_PATH` 初始为 True，但只 patch 其一
另一处不跟随。于是测试不得不把 4 个路径常量**两处都 patch**
（setattr(auth,...) + setattr(db,...)）——而漏 patch 的后果是静默写生产库，
commit fc7fc21 修的那次「黄金测试写进生产库」正是这么来的。
改法：让路径常量回归**模块属性单一来源**。
- db.py：删掉 `import auth_system`，内部改用本模块常量（DATA_DIR / DB_PATH /
SECRET_KEY_PATH）。这些常量本就定义在本模块，绕道包属性纯属多余。
- backup.py：改用 `import auth_system.db as db` 做模块限定访问，路径走
db.BACKUP_DIR / db.DB_PATH——保留动态读取能力，来源从「包属性」换成
「db 模块」，与唯一来源一致。
注意**不能**写成 `from . import db`：该形式在静态图里同时指向包本身与子模块，
会把这个环重新连上（实测确实如此，SCC 从 11 降到 2 就卡住了）。
- admin_site.py：5 处 `auth.DB_PATH` / `auth.BACKUP_DIR`（auth 是包别名）同样
改为经 db 模块访问。**这处不修就会出事**——它读的是包属性快照，而测试已改为
patch db 模块，于是备份上传会静默落到生产 data/backups/（本次实测真的发生了，
2 个 upload_*.db 已清理，副本留在临时目录）。
- 测试：7 个文件的 patch 目标从包属性迁到 auth_system.db.*。
- __init__.py：补注释说明这些常量在此只是 import 快照，禁止经包属性重定向。
验证：
- Tarjan 重算：非平凡强连通分量 11 → **0**（import 图已是 DAG）
- AST 全仓扫描：经包属性访问路径常量的点 **0 处残留**
- 行为：子进程内 patch auth_system.db.* 后，get_conn 连的库 / create_backup
落盘目录 / get_secret_key 目标路径**三者全部跟随**
- 导入姿势回归 8/8 通过
- 门禁：ruff 0 · pyright 0 errors 0 warnings · pytest 175 passed
119 lines
4.5 KiB
Python
119 lines
4.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""注册/找回/卡密端点的沙箱测试（不触碰生产 data/site.db）。
|
|
|
|
直接向沙箱库 email_codes 表播种验证码，绕过 SMTP，覆盖：
|
|
- 注册协议门槛、用户名/密码长度校验
|
|
- 注册成功路径：建号 + 赠送次数 + 签发可用会话 Cookie
|
|
- 验证码错误分支
|
|
- 找回密码：重置后可用新密码登录、旧会话失效
|
|
- 卡密充值
|
|
"""
|
|
import time
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import auth_system as auth
|
|
import core.db as db
|
|
|
|
SECRET = 'register-test-secret'
|
|
|
|
|
|
@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()
|
|
monkeypatch.setattr('web.deps.secret_key', lambda: SECRET)
|
|
import web.auth_state as auth_routes
|
|
auth_routes._SEND_CODE_IP.clear()
|
|
from web.app import create_app
|
|
return TestClient(create_app())
|
|
|
|
|
|
def _seed_code(email: str, code: str, purpose: str = 'register'):
|
|
conn = auth.get_conn()
|
|
conn.execute('DELETE FROM email_codes WHERE email = ? AND purpose = ?', (email, purpose))
|
|
conn.execute('INSERT INTO email_codes (email, code, expires_at, created_at, purpose) '
|
|
'VALUES (?, ?, ?, ?, ?)',
|
|
(email, code, int(time.time()) + 600, int(time.time()), purpose))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def test_register_requires_agreement(env):
|
|
r = env.post('/auth/register', json={
|
|
'username': 'newuser', 'password': 'Passw0rd12', 'email': 'n@example.com',
|
|
'code': '000000', 'agree': '0'})
|
|
assert r.json()['message'].startswith('请先阅读并同意')
|
|
|
|
|
|
def test_register_success(env):
|
|
_seed_code('new@example.com', '123456')
|
|
r = env.post('/auth/register', json={
|
|
'username': 'newuser', 'password': 'Passw0rd12', 'email': 'new@example.com',
|
|
'code': '123456', 'agree': '1', 'device_id': 'dev-1'})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body['success'] is True
|
|
assert '已赠送' in body['message']
|
|
assert 'session' in r.cookies
|
|
|
|
# 会话可用
|
|
assert env.post('/auth/heartbeat', json={}).status_code == 200
|
|
|
|
# 用户名/邮箱已被占用
|
|
_seed_code('new@example.com', '222222')
|
|
r2 = env.post('/auth/register', json={
|
|
'username': 'newuser', 'password': 'Passw0rd12', 'email': 'new@example.com',
|
|
'code': '222222', 'agree': '1'})
|
|
assert '已被注册' in r2.json()['message']
|
|
|
|
|
|
def test_register_bad_code(env):
|
|
r = env.post('/auth/register', json={
|
|
'username': 'another', 'password': 'Passw0rd12', 'email': 'a@example.com',
|
|
'code': '999999', 'agree': '1'})
|
|
assert r.json()['message'] == '邮箱验证码错误或已过期'
|
|
|
|
|
|
def test_forgot_password_resets(env):
|
|
# 建一个用户
|
|
auth.create_user('forgetme', 'OldPass123', email='f@example.com', credits=5)
|
|
_seed_code('f@example.com', '654321', purpose='reset')
|
|
r = env.post('/auth/forgot_password', json={
|
|
'email': 'f@example.com', 'code': '654321', 'password': 'NewPass456'})
|
|
assert r.status_code == 200
|
|
assert r.json()['success'] is True
|
|
# 新密码可登录
|
|
r2 = env.post('/auth/login', json={'username': 'forgetme', 'password': 'NewPass456'})
|
|
assert r2.status_code == 200 and r2.json()['success'] is True
|
|
|
|
|
|
def test_forgot_password_admin_blocked(env):
|
|
conn = auth.get_conn()
|
|
conn.execute("UPDATE users SET role='admin', email='boss@example.com' WHERE id=1")
|
|
conn.commit()
|
|
conn.close()
|
|
_seed_code('boss@example.com', '111111', purpose='reset')
|
|
r = env.post('/auth/forgot_password', json={
|
|
'email': 'boss@example.com', 'code': '111111', 'password': 'NewPass456'})
|
|
# 管理员不走自助找回：统一防枚举文案
|
|
assert r.json()['message'] == '邮箱验证码错误或已过期'
|
|
|
|
|
|
def test_redeem_card(env):
|
|
# 造一张卡密并让用户登录
|
|
auth.create_user('buyer', 'Passw0rd12', email='b@example.com', credits=0)
|
|
code = auth.generate_cards(1, 10, note='test')[0]
|
|
env.post('/auth/login', json={'username': 'buyer', 'password': 'Passw0rd12'})
|
|
r = env.post('/auth/redeem', json={'code': code})
|
|
assert r.status_code == 200
|
|
assert r.json()['success'] is True
|
|
assert r.json()['data']['credits'] == 10
|