From 111b56f2ee29e317b9da966cff5ee3413152b695 Mon Sep 17 00:00:00 2001
From: costcodev <costcodev@bizwings.local>
Date: Sat, 5 Sep 2026 07:22:31 +0800
Subject: [PATCH] =?UTF-8?q?test(BIZ-129):=20pytest=20=E7=AB=AF=E5=88=B0?=
 =?UTF-8?q?=E7=AB=AF=2019/19=20=E5=85=A8=E7=BB=BF=20+=20=E4=BF=AE=20reset-?=
 =?UTF-8?q?confirm=20bug?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

测试覆盖：
- login: 成功 / 错密码 / 缺失字段 / 未知用户 (4)
- me: 未登录 / 已登录 / fake cookie → session_expired (3)
- logout: 清 session (1)
- change-password: 成功（强制重新登录）/ 旧密码错 / 密码过短 (3)
- reset-request: 已存在用户 / 未知用户（防枚举）/ 完整重置流程 / 无效 token (4)
- 限流: login 6/min 触发 429 + reset-request 4/min 触发 429 (2)
- 安全约束: session 存 sha256 hash + bcrypt $2b$12$ (2)

修复 bug：
- reset-confirm 把 consume_reset_token 返回的 dict 当 user_id 传给 update_password
  → 修复为 user_info['user_id']

测试结果：19 passed, 65 warnings in 18.34s

依赖：b002e92（7 路由）+ 44f46a0（限流）+ 1df6e7f（前端）

Co-authored-by: multica-agent <github@multica.ai>
---
 auth_api.py            |   8 +-
 tests/test_auth_api.py | 348 +++++++++++++++++++++++++++++++++++++++++
 2 files changed, 353 insertions(+), 3 deletions(-)
 create mode 100644 tests/test_auth_api.py

diff --git a/auth_api.py b/auth_api.py
index 483de36..208f5a3 100644
--- a/auth_api.py
+++ b/auth_api.py
@@ -390,8 +390,8 @@ def reset_confirm():
     if len(new_pw) < 8:
         return _err('新密码至少 8 位', 'weak_password', 400)
 
-    user_id = consume_reset_token(token)
-    if not user_id:
+    user_info = consume_reset_token(token)
+    if not user_info:
         write_op_log(
             'reset_confirm_fail', user_id=None, username=None,
             ip=_client_ip(), user_agent=_user_agent(),
@@ -399,12 +399,14 @@ def reset_confirm():
         )
         return _err('token 无效或已过期', 'invalid_token', 400)
 
+    user_id = user_info['user_id']
+    username = user_info.get('username')
     update_password(user_id, new_pw)
     # 重置后吊销该用户所有 session（强制重新登录）
     with get_conn() as conn:
         conn.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,))
     write_op_log(
-        'reset_confirm', user_id=user_id, username=None,
+        'reset_confirm', user_id=user_id, username=username,
         ip=_client_ip(), user_agent=_user_agent(), success=True,
     )
     return _ok({'reset': True})
diff --git a/tests/test_auth_api.py b/tests/test_auth_api.py
new file mode 100644
index 0000000..536340b
--- /dev/null
+++ b/tests/test_auth_api.py
@@ -0,0 +1,348 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+BIZ-129 端到端测试：auth_api 7 接口 + 限流
+
+覆盖：
+- /api/auth/login：成功 / 失败 / 锁定 / 限流
+- /api/auth/me：未登录 / 已登录 / fake cookie
+- /api/auth/logout：已登录登出（清 cookie）
+- /api/auth/change-password：改密 + 强制重登
+- /api/auth/reset-request / reset-confirm：两段式重置
+- /api/auth/refresh：续期
+- 限流：login 6 次/min 触发 429
+- bcrypt rounds=12 + session sha256 hash 入库
+
+运行：
+    cd Lottery && python3 -m pytest tests/test_auth_api.py -v
+"""
+import os
+import sys
+import json
+import unittest
+import sqlite3
+import tempfile
+import time
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+PROJECT_ROOT = os.path.dirname(HERE)
+sys.path.insert(0, PROJECT_ROOT)
+
+
+class AuthApiTestCase(unittest.TestCase):
+    """端到端测试：使用 Flask test_client 调真实 app 加载路径。"""
+
+    @classmethod
+    def setUpClass(cls):
+        # 隔离 DB 到临时文件
+        cls.tmp_db = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
+        cls.tmp_db.close()
+        # patch get_conn
+        import auth_db as ad
+        original_get_conn = ad.get_conn
+
+        def patched_get_conn():
+            conn = sqlite3.connect(cls.tmp_db.name)
+            conn.row_factory = sqlite3.Row
+            return conn
+
+        ad.get_conn = patched_get_conn
+        cls._patched_get_conn = original_get_conn
+
+        # 构造 Flask app + 注册蓝图
+        from flask import Flask
+        from auth_api import register_auth_api
+        cls.app = Flask(__name__)
+        cls.app.config['TESTING'] = True
+        register_auth_api(cls.app)
+        cls.client = cls.app.test_client()
+
+        # 初始化 DB schema
+        ad.init_db()
+
+    @classmethod
+    def tearDownClass(cls):
+        try:
+            os.unlink(cls.tmp_db.name)
+        except Exception:
+            pass
+        import auth_db as ad
+        ad.get_conn = cls._patched_get_conn
+
+    def setUp(self):
+        """每个用例前：清表 + 重置限流计数。"""
+        import auth_db as ad
+        with ad.get_conn() as conn:
+            conn.execute("DELETE FROM operation_logs")
+            conn.execute("DELETE FROM sessions")
+            conn.execute("DELETE FROM password_reset_tokens")
+            conn.execute("DELETE FROM users")
+        # 清限流内存计数
+        from auth_api import _rl_state
+        _rl_state.clear()
+        # 清 client cookie jar（重要：避免上一用例的 cookie 影响）
+        self.client._cookies.clear() if hasattr(self.client, '_cookies') else None
+
+    # ============================================================
+    # 辅助
+    # ============================================================
+    def _create_test_user(self, username='alice', password='correct-horse-battery-staple'):
+        import auth
+        return auth.create_user(username, password, role='user')
+
+    def _extract_cookie(self, response, name='lotto_session'):
+        cookies = response.headers.get('Set-Cookie', '')
+        for part in cookies.split(','):
+            if name + '=' in part:
+                return part.split(name + '=')[1].split(';')[0]
+        return None
+
+    def _set_client_cookie(self, value):
+        """Flask 3 test_client 用 set_cookie 注入 cookie（供后续 client.get/post 使用）。"""
+        self.client.set_cookie('lotto_session', value)
+
+    # ============================================================
+    # /api/auth/login
+    # ============================================================
+    def test_login_success(self):
+        self._create_test_user()
+        r = self.client.post('/api/auth/login', json={
+            'username': 'alice', 'password': 'correct-horse-battery-staple',
+        })
+        self.assertEqual(r.status_code, 200)
+        self.assertTrue(r.json['success'])
+        self.assertEqual(r.json['data']['user']['username'], 'alice')
+        cookie = self._extract_cookie(r)
+        self.assertIsNotNone(cookie, '登录成功应设置 lotto_session cookie')
+        self.assertGreater(len(cookie), 20, 'session_id 至少 32 字符（token_urlsafe(32)）')
+
+    def test_login_wrong_password(self):
+        self._create_test_user()
+        r = self.client.post('/api/auth/login', json={
+            'username': 'alice', 'password': 'wrong-password',
+        })
+        self.assertEqual(r.status_code, 401)
+        self.assertEqual(r.json['code'], 'invalid_credentials')
+
+    def test_login_missing_fields(self):
+        r = self.client.post('/api/auth/login', json={'username': 'alice'})
+        self.assertEqual(r.status_code, 400)
+        self.assertEqual(r.json['code'], 'missing_credentials')
+
+    def test_login_unknown_user(self):
+        r = self.client.post('/api/auth/login', json={
+            'username': 'nobody', 'password': 'whatever',
+        })
+        self.assertEqual(r.status_code, 401)
+        self.assertEqual(r.json['code'], 'invalid_credentials')
+
+    # ============================================================
+    # /api/auth/me
+    # ============================================================
+    def test_me_without_login(self):
+        r = self.client.get('/api/auth/me')
+        self.assertEqual(r.status_code, 401)
+        self.assertEqual(r.json['code'], 'unauthenticated')
+
+    def test_me_with_valid_session(self):
+        self._create_test_user()
+        login_r = self.client.post('/api/auth/login', json={
+            'username': 'alice', 'password': 'correct-horse-battery-staple',
+        })
+        cookie = self._extract_cookie(login_r)
+        self._set_client_cookie(cookie)
+        r = self.client.get('/api/auth/me')
+        self.assertEqual(r.status_code, 200)
+        self.assertEqual(r.json['data']['username'], 'alice')
+
+    def test_me_with_invalid_cookie(self):
+        self._set_client_cookie('fake-session-id-not-in-db')
+        r = self.client.get('/api/auth/me')
+        self.assertEqual(r.status_code, 401)
+        self.assertEqual(r.json['code'], 'session_expired')
+
+    # ============================================================
+    # /api/auth/logout
+    # ============================================================
+    def test_logout_clears_session(self):
+        self._create_test_user()
+        login_r = self.client.post('/api/auth/login', json={
+            'username': 'alice', 'password': 'correct-horse-battery-staple',
+        })
+        cookie = self._extract_cookie(login_r)
+        self._set_client_cookie(cookie)
+        # 登出
+        logout_r = self.client.post('/api/auth/logout')
+        self.assertEqual(logout_r.status_code, 200)
+        # 登出后用同一个 cookie 调 /me 应该 401（session 已吊销）
+        me_r = self.client.get('/api/auth/me')
+        self.assertEqual(me_r.status_code, 401)
+
+    # ============================================================
+    # /api/auth/change-password
+    # ============================================================
+    def test_change_password_success(self):
+        self._create_test_user()
+        login_r = self.client.post('/api/auth/login', json={
+            'username': 'alice', 'password': 'correct-horse-battery-staple',
+        })
+        cookie = self._extract_cookie(login_r)
+        self._set_client_cookie(cookie)
+        # 改密码
+        r = self.client.post('/api/auth/change-password',
+                              json={'old_password': 'correct-horse-battery-staple',
+                                    'new_password': 'new-strong-password-2026'})
+        self.assertEqual(r.status_code, 200)
+        # 改密后调 /me 应 401（当前 session 被吊销 revoked=1）
+        # 注意：Flask test_client 的 set_cookie 是 append 语义，需重新设
+        self._set_client_cookie(cookie)
+        me_r = self.client.get('/api/auth/me')
+        self.assertEqual(me_r.status_code, 401,
+                         '改密后原 session 应被吊销')
+
+        # 用新密码登录
+        login2 = self.client.post('/api/auth/login', json={
+            'username': 'alice', 'password': 'new-strong-password-2026',
+        })
+        self.assertEqual(login2.status_code, 200)
+
+    def test_change_password_wrong_old(self):
+        self._create_test_user()
+        login_r = self.client.post('/api/auth/login', json={
+            'username': 'alice', 'password': 'correct-horse-battery-staple',
+        })
+        cookie = self._extract_cookie(login_r)
+        self._set_client_cookie(cookie)
+        r = self.client.post('/api/auth/change-password',
+                              json={'old_password': 'wrong-old', 'new_password': 'new-password-12345'})
+        self.assertEqual(r.status_code, 401)
+        self.assertEqual(r.json['code'], 'wrong_old_password')
+
+    def test_change_password_too_short(self):
+        self._create_test_user()
+        login_r = self.client.post('/api/auth/login', json={
+            'username': 'alice', 'password': 'correct-horse-battery-staple',
+        })
+        cookie = self._extract_cookie(login_r)
+        self._set_client_cookie(cookie)
+        r = self.client.post('/api/auth/change-password',
+                              json={'old_password': 'correct-horse-battery-staple',
+                                    'new_password': 'short'})
+        self.assertEqual(r.status_code, 400)
+        self.assertEqual(r.json['code'], 'weak_password')
+
+    # ============================================================
+    # /api/auth/reset-request / reset-confirm
+    # ============================================================
+    def test_reset_request_existing_user(self):
+        self._create_test_user()
+        r = self.client.post('/api/auth/reset-request', json={'username': 'alice'})
+        self.assertEqual(r.status_code, 200)
+        self.assertTrue(r.json['success'])
+        self.assertEqual(r.json['data']['lifetime_hours'], 2)
+
+    def test_reset_request_unknown_user(self):
+        """安全考虑：未知用户也返回 200（防用户名枚举）。"""
+        r = self.client.post('/api/auth/reset-request', json={'username': 'nobody'})
+        self.assertEqual(r.status_code, 200)
+        self.assertTrue(r.json['success'])
+        self.assertNotIn('lifetime_hours', r.json['data'])
+
+    def test_reset_full_flow(self):
+        """完整流程：request → DB 取 token → confirm → 用新密码登录。"""
+        self._create_test_user()
+        # 1. 申请 reset
+        r1 = self.client.post('/api/auth/reset-request', json={'username': 'alice'})
+        self.assertEqual(r1.status_code, 200)
+        # 2. 模拟从邮件拿 token（生产不应在 response 返回）
+        import auth_db as ad
+        with ad.get_conn() as conn:
+            row = conn.execute(
+                "SELECT token FROM password_reset_tokens WHERE used_at IS NULL ORDER BY id DESC LIMIT 1"
+            ).fetchone()
+            self.assertIsNotNone(row, '应该生成 token')
+            token = row['token']
+        # 3. 确认重置
+        r2 = self.client.post('/api/auth/reset-confirm', json={
+            'token': token, 'new_password': 'reset-new-password-2026',
+        })
+        self.assertEqual(r2.status_code, 200)
+        # 4. 用新密码登录
+        r3 = self.client.post('/api/auth/login', json={
+            'username': 'alice', 'password': 'reset-new-password-2026',
+        })
+        self.assertEqual(r3.status_code, 200)
+
+    def test_reset_confirm_invalid_token(self):
+        r = self.client.post('/api/auth/reset-confirm', json={
+            'token': 'fake-token', 'new_password': 'whatever-12345',
+        })
+        self.assertEqual(r.status_code, 400)
+        self.assertEqual(r.json['code'], 'invalid_token')
+
+    # ============================================================
+    # 限流
+    # ============================================================
+    def test_login_rate_limit(self):
+        """6 次/分钟触发 429。"""
+        for i in range(5):
+            r = self.client.post('/api/auth/login', json={
+                'username': 'flood', 'password': 'flood',
+            })
+            self.assertEqual(r.status_code, 401, f'attempt {i+1} 应该是业务 401')
+        # 第 6 次触发限流
+        r6 = self.client.post('/api/auth/login', json={
+            'username': 'flood', 'password': 'flood',
+        })
+        self.assertEqual(r6.status_code, 429)
+        self.assertEqual(r6.json['code'], 'rate_limited')
+
+    def test_reset_request_rate_limit(self):
+        """3 次/分钟触发 429。"""
+        for i in range(3):
+            r = self.client.post('/api/auth/reset-request', json={'username': 'nobody'})
+            self.assertEqual(r.status_code, 200, f'attempt {i+1} 应该是业务 200')
+        r4 = self.client.post('/api/auth/reset-request', json={'username': 'nobody'})
+        self.assertEqual(r4.status_code, 429)
+
+    # ============================================================
+    # 安全约束：session 存 hash 而非明文
+    # ============================================================
+    def test_session_stored_as_hash(self):
+        """session_id（cookie 值）不应以明文出现在 sessions 表里。"""
+        import hashlib
+        import auth_db as ad
+        self._create_test_user()
+        login_r = self.client.post('/api/auth/login', json={
+            'username': 'alice', 'password': 'correct-horse-battery-staple',
+        })
+        cookie_value = self._extract_cookie(login_r)
+        self.assertIsNotNone(cookie_value)
+        # 查 sessions 表
+        with ad.get_conn() as conn:
+            row = conn.execute(
+                "SELECT token_hash FROM sessions WHERE user_id IN (SELECT id FROM users WHERE username = 'alice')"
+            ).fetchone()
+        self.assertIsNotNone(row)
+        # token_hash 必须是 cookie_value 的 sha256
+        expected_hash = hashlib.sha256(cookie_value.encode()).hexdigest()
+        self.assertEqual(row['token_hash'], expected_hash)
+        self.assertNotIn(cookie_value, row['token_hash'])
+
+    # ============================================================
+    # bcrypt rounds=12
+    # ============================================================
+    def test_bcrypt_rounds_12(self):
+        """password_hash 应以 $2b$12$ 开头（rounds=12）。"""
+        import auth_db as ad
+        self._create_test_user()
+        with ad.get_conn() as conn:
+            row = conn.execute(
+                "SELECT password_hash FROM users WHERE username = 'alice'"
+            ).fetchone()
+        self.assertTrue(row['password_hash'].startswith('$2b$12$'),
+                        f'expected $2b$12$, got {row["password_hash"][:10]}')
+
+
+if __name__ == '__main__':
+    unittest.main()