# -*- 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, 'DATA_DIR', tmp_path) monkeypatch.setattr(auth, 'DB_PATH', db_path) monkeypatch.setattr(auth, 'BACKUP_DIR', tmp_path / 'backups') monkeypatch.setattr(auth, '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) monkeypatch.setattr('web.routers.auth.secret_key', lambda: SECRET) monkeypatch.setattr('web.routers.register.secret_key', lambda: SECRET) import 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('/api/v1/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('/api/v1/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('/api/v1/auth/heartbeat', json={}).status_code == 200 # 用户名/邮箱已被占用 _seed_code('new@example.com', '222222') r2 = env.post('/api/v1/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('/api/v1/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('/api/v1/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('/api/v1/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('/api/v1/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('/api/v1/auth/login', json={'username': 'buyer', 'password': 'Passw0rd12'}) r = env.post('/api/v1/auth/redeem', json={'code': code}) assert r.status_code == 200 assert r.json()['success'] is True assert r.json()['data']['credits'] == 10