refactor!: uv 管理环境 + 彻底删除技术债(Flask 与 requests 全部移除)

## uv 环境管理
- pyproject.toml 声明 dependencies + [dependency-groups].dev,uv.lock 精确锁定
- 删除 requirements.txt / requirements-fastapi.txt / requirements-dev.txt
- .python-version = 3.12;Dockerfile 改用 uv sync --frozen(可复现构建)
- 新增 jinja2(首页渲染)

## 删除 Flask 技术债
- 从 main.py 用 ast 精确剥离 56 个 Flask 节点(路由/中间件/错误处理/启动),
  框架无关的服务层落为 services/runtime.py(MusicAPIService、下载/打包 worker、
  任务监控、缓存清扫),零 Flask 依赖
- 删除 main.py(4527 行)与 auth_routes.py(1050 行)
- auth_system.py 移除会话/请求上下文层(login_session/current_user/login_required/
  admin_required/_client_ip 等),保留数据与业务层
- 新增 core/net.py(IP 解析)、web/auth_state.py(登录锁定/2FA 票据/发码限流)
- FastAPI 侧改直接依赖 services.runtime,不再 import main
- web/app.py 去掉 Flask fallback;入口默认根路径模式(API_PREFIX 可选)

## 删除 requests 技术债
- 新增 core/sync_http.py:requests 风格的同步门面(底层 aiohttp 连接池,
  流式读取经有界队列背压),含 Session/headers/(connect,read) 元组超时兼容
- core/utils、auth_system、podcast_api、music_downloader、services 全部切换
- 实测真实下载 5.2MB 音频完整成功

## 质量门禁归零
- ruff:0 错误(B904 全部修复,不再忽略)
- pyright:0 错误(类型收窄修复)
- pytest:140/140(含 golden 契约回归、root 模式 66 项路径覆盖)
- 真实 uvicorn 启动:/health、/、/static、受保护端鉴权全部正确

新增 tests/golden/responses.json:删除 Flask 前捕获的 32 例响应快照,
FastAPI 逐字段复现——删除巨石后行为回归仍有硬证据。
This commit is contained in:
snowy-yang committed 2026-09-26 11:10:17 +08:00
1 parent f5d1e3c3a9
commit 8446166120
41 files changed
+4836 -6060

No files matched your search

+540
View File
@@ -0,0 +1,540 @@
[
{
"name": "health",
"auth": "none",
"method": "GET",
"path": "/health",
"payload": null,
"status": 200,
"body": {
"data": {
"cookie_count": 3,
"cookie_status": "valid",
"downloads_dir": "configured",
"service": "running",
"timestamp": 1790391125,
"version": "2.4.0"
},
"message": "API服务运行正常",
"status": 200,
"success": true
}
},
{
"name": "api_info",
"auth": "none",
"method": "GET",
"path": "/api/info",
"payload": null,
"status": 200,
"body": {
"data": {
"config": {
"downloads_dir": "configured",
"max_body_bytes": "2MB(备份上传 200MB)"
},
"description": "云村小助手 · 网易云音乐解析/下载服务",
"endpoints": {
"/album": "GET/POST - 获取专辑详情",
"/api/info": "GET - API信息",
"/api/playlist/play": "GET/POST - 公开歌单播放API(id/level/limit)",
"/download": "POST - 下载音乐(付费动作仅接受 POST)",
"/global/playlist": "GET/POST - 全局账号歌单详情",
"/global/playlists": "GET/POST - 全局账号歌单列表",
"/health": "GET - 健康检查",
"/playlist": "GET/POST - 获取歌单详情",
"/podcast/download": "POST - 播客单集下载",
"/podcast/download/batch_zip": "POST - 整台播客打包ZIP",
"/podcast/hot": "GET/POST - 热门播客与分类",
"/podcast/parse": "GET/POST - 播客解析(电台/单集)",
"/podcast/programs": "GET/POST - 播客单集列表(分页)",
"/podcast/search": "GET/POST - 播客搜索",
"/search": "GET/POST - 搜索音乐",
"/song": "GET/POST - 获取歌曲信息"
},
"name": "云村小助手API服务",
"supported_qualities": [
"standard",
"exhigh",
"lossless",
"hires",
"sky",
"jyeffect",
"jymaster",
"dolby"
],
"version": "2.4.0"
},
"message": "API信息获取成功",
"status": 200,
"success": true
}
},
{
"name": "toplist_unauth",
"auth": "none",
"method": "POST",
"path": "/toplist",
"payload": {},
"status": 401,
"body": {
"error_code": "unauthorized",
"message": "请先登录后再使用解析功能",
"status": 401,
"success": false
}
},
{
"name": "user_status_unauth",
"auth": "none",
"method": "POST",
"path": "/user/status",
"payload": {},
"status": 200,
"body": {
"data": {
"cookie": {
"is_valid": false
},
"user": null
},
"message": "个人账号未登录",
"status": 200,
"success": true
}
},
{
"name": "user_logout",
"auth": "none",
"method": "POST",
"path": "/user/logout",
"payload": {},
"status": 200,
"body": {
"data": {
"is_valid": false
},
"message": "已退出登录",
"status": 200,
"success": true
}
},
{
"name": "auth_trial",
"auth": "none",
"method": "POST",
"path": "/auth/trial",
"payload": {},
"status": 400,
"body": {
"message": "注册即送 3 次免费下载次数,请直接注册账号",
"status": 400,
"success": false
}
},
{
"name": "auth_sendcode_badpurpose",
"auth": "none",
"method": "POST",
"path": "/auth/send_code",
"payload": {
"email": "x@y.com",
"purpose": "bad"
},
"status": 400,
"body": {
"message": "验证码用途不正确",
"status": 400,
"success": false
}
},
{
"name": "auth_register_shortuser",
"auth": "none",
"method": "POST",
"path": "/auth/register",
"payload": {
"username": "ab",
"password": "12345678",
"email": "a@b.com",
"code": "x",
"agree": "1"
},
"status": 400,
"body": {
"message": "用户名需 3-20 位,仅支持中文、字母、数字、下划线,且不能纯下划线",
"status": 400,
"success": false
}
},
{
"name": "auth_register_noagree",
"auth": "none",
"method": "POST",
"path": "/auth/register",
"payload": {
"username": "abcdef",
"password": "12345678",
"email": "a@b.com",
"code": "x",
"agree": "0"
},
"status": 400,
"body": {
"message": "请先阅读并同意《用户协议》与《隐私政策》",
"status": 400,
"success": false
}
},
{
"name": "auth_register_badcode",
"auth": "none",
"method": "POST",
"path": "/auth/register",
"payload": {
"username": "abcdef",
"password": "12345678",
"email": "a@b.com",
"code": "zz",
"agree": "1"
},
"status": 400,
"body": {
"message": "邮箱验证码错误或已过期",
"status": 400,
"success": false
}
},
{
"name": "auth_redeem_unauth",
"auth": "none",
"method": "POST",
"path": "/auth/redeem",
"payload": {
"code": "X"
},
"status": 401,
"body": {
"error_code": "unauthorized",
"message": "请先登录",
"status": 401,
"success": false
}
},
{
"name": "auth_forgot_shortpwd",
"auth": "none",
"method": "POST",
"path": "/auth/forgot_password",
"payload": {
"email": "a@b.com",
"code": "x",
"password": "short"
},
"status": 400,
"body": {
"message": "新密码长度需在 8-128 位之间",
"status": 400,
"success": false
}
},
{
"name": "admin_logs_forbidden",
"auth": "user",
"method": "POST",
"path": "/admin/logs",
"payload": {},
"status": 403,
"body": {
"message": "需要管理员权限,请使用管理员账号登录",
"status": 403,
"success": false
}
},
{
"name": "admin_site_users_forbidden",
"auth": "user",
"method": "POST",
"path": "/admin/site/users",
"payload": {},
"status": 403,
"body": {
"error_code": "forbidden",
"message": "需要管理员权限",
"status": 403,
"success": false
}
},
{
"name": "admin_qr_forbidden",
"auth": "user",
"method": "POST",
"path": "/admin/qr/create",
"payload": {},
"status": 403,
"body": {
"message": "需要管理员权限,请使用管理员账号登录",
"status": 403,
"success": false
}
},
{
"name": "global_playlists_forbidden",
"auth": "user",
"method": "POST",
"path": "/global/playlists",
"payload": {},
"status": 403,
"body": {
"error_code": "forbidden",
"message": "需要管理员权限",
"status": 403,
"success": false
}
},
{
"name": "download_unauth",
"auth": "none",
"method": "POST",
"path": "/download",
"payload": {
"id": "1"
},
"status": 401,
"body": {
"error_code": "unauthorized",
"message": "请先登录后再使用解析功能",
"status": 401,
"success": false
}
},
{
"name": "download_badid",
"auth": "user",
"method": "POST",
"path": "/download",
"payload": {
"id": "abc"
},
"status": 400,
"body": {
"message": "无效的歌曲 ID",
"status": 400,
"success": false
}
},
{
"name": "download_badquality",
"auth": "user",
"method": "POST",
"path": "/download",
"payload": {
"id": "347230",
"quality": "nope"
},
"status": 400,
"body": {
"message": "无效的音质参数,支持: standard, exhigh, lossless, hires, sky, jyeffect, jymaster, dolby",
"status": 400,
"success": false
}
},
{
"name": "task_list_unauth",
"auth": "none",
"method": "POST",
"path": "/api/task/list",
"payload": {
"device_id": "x"
},
"status": 401,
"body": {
"error_code": "unauthorized",
"message": "请先登录后再使用解析功能",
"status": 401,
"success": false
}
},
{
"name": "task_cancel_unauth",
"auth": "none",
"method": "POST",
"path": "/api/task/cancel",
"payload": {
"task_id": "x"
},
"status": 401,
"body": {
"error_code": "unauthorized",
"message": "请先登录后再使用解析功能",
"status": 401,
"success": false
}
},
{
"name": "queue",
"auth": "none",
"method": "POST",
"path": "/api/queue",
"payload": {},
"status": 200,
"body": {
"data": {
"avg_batch_seconds": null,
"batch_active": 0,
"batch_capacity": 3,
"batch_waiting": 0,
"download_active": 0,
"download_capacity": 10,
"download_waiting": 0
},
"message": "获取排队状态成功",
"status": 200,
"success": true
}
},
{
"name": "delivery_404",
"auth": "user",
"method": "POST",
"path": "/download/result",
"payload": {
"task_id": "nope"
},
"status": 404,
"body": {
"message": "任务不存在或已过期",
"status": 404,
"success": false
}
},
{
"name": "delivery_batch_404",
"auth": "user",
"method": "POST",
"path": "/download/batch_zip/result",
"payload": {
"task_id": "nope"
},
"status": 404,
"body": {
"message": "任务不存在或已过期",
"status": 404,
"success": false
}
},
{
"name": "fe_error_nomsg",
"auth": "none",
"method": "POST",
"path": "/api/fe_error",
"payload": {},
"status": 400,
"body": {
"message": "缺少 msg 参数",
"status": 400,
"success": false
}
},
{
"name": "play_badkey",
"auth": "none",
"method": "POST",
"path": "/api/playlist/play",
"payload": {
"id": 3778678
},
"status": 401,
"body": {
"message": "需要有效的 API 密钥(请联系站点管理员获取)",
"status": 401,
"success": false
}
},
{
"name": "podcast_programs_badid",
"auth": "user",
"method": "POST",
"path": "/podcast/programs",
"payload": {
"radio_id": "abc"
},
"status": 400,
"body": {
"message": "无效的电台 ID",
"status": 400,
"success": false
}
},
{
"name": "song_badid",
"auth": "user",
"method": "POST",
"path": "/song",
"payload": {
"id": "abc"
},
"status": 400,
"body": {
"message": "无效的歌曲ID,请输入数字ID或分享链接",
"status": 400,
"success": false
}
},
{
"name": "comments_badid",
"auth": "user",
"method": "POST",
"path": "/comments",
"payload": {
"id": "abc"
},
"status": 400,
"body": {
"message": "无效的歌曲ID,请输入数字ID或分享链接",
"status": 400,
"success": false
}
},
{
"name": "playlist_badid",
"auth": "user",
"method": "POST",
"path": "/playlist",
"payload": {
"id": ""
},
"status": 400,
"body": {
"message": "参数 'playlist_id' 不能为空",
"status": 400,
"success": false
}
},
{
"name": "album_badid",
"auth": "user",
"method": "POST",
"path": "/album",
"payload": {
"id": "abc"
},
"status": 400,
"body": {
"message": "无效的专辑ID,请输入数字ID或分享链接",
"status": 400,
"success": false
}
},
{
"name": "index",
"auth": "none",
"method": "GET",
"path": "/",
"payload": null,
"status": 200,
"body": null,
"html_len": 466261,
"html_sha256": "5ba897c2f07b43b6c34e998505fea385cb6a6c8bfd5115d014f80044a38c6200"
}
]
+10 -10
View File
@@ -42,7 +42,7 @@ def app_env(tmp_path, monkeypatch):
monkeypatch.setattr('web.deps.secret_key', lambda: SECRET)
# 清空登录失败锁定/2FA 的模块级内存态,避免测试间串扰
import auth_routes
import web.auth_state as auth_routes
auth_routes._LOGIN_FAILS.clear()
auth_routes._LOGIN_FAIL_LOGGED.clear()
auth_routes._ADMIN_2FA.clear()
@@ -52,14 +52,14 @@ def app_env(tmp_path, monkeypatch):
def test_wrong_password(app_env):
r = app_env.post('/api/v1/auth/login', json={'username': 'normaluser', 'password': 'nope'})
r = app_env.post('/auth/login', json={'username': 'normaluser', 'password': 'nope'})
assert r.status_code == 400
assert r.json()['success'] is False
assert 'session' not in r.cookies
def test_user_login_issues_working_cookie(app_env):
r = app_env.post('/api/v1/auth/login', json={'username': 'normaluser', 'password': USER_PW})
r = app_env.post('/auth/login', json={'username': 'normaluser', 'password': USER_PW})
assert r.status_code == 200
body = r.json()
assert body['success'] is True
@@ -67,13 +67,13 @@ def test_user_login_issues_working_cookie(app_env):
assert 'session' in r.cookies
# 用签发的 Cookie 访问受保护端点,应放行
r2 = app_env.post('/api/v1/auth/heartbeat', json={})
r2 = app_env.post('/auth/heartbeat', json={})
assert r2.status_code == 200
assert r2.json()['data']['duration_text'] == '97 次'
def test_admin_login_returns_2fa_ticket(app_env):
r = app_env.post('/api/v1/auth/login', json={'username': 'bossuser', 'password': ADMIN_PW})
r = app_env.post('/auth/login', json={'username': 'bossuser', 'password': ADMIN_PW})
assert r.status_code == 200
data = r.json()['data']
assert data['admin_2fa'] is True
@@ -82,12 +82,12 @@ def test_admin_login_returns_2fa_ticket(app_env):
def test_logout_bumps_login_seq_and_clears(app_env):
app_env.post('/api/v1/auth/login', json={'username': 'normaluser', 'password': USER_PW})
app_env.post('/auth/login', json={'username': 'normaluser', 'password': USER_PW})
conn = auth.get_conn()
seq_before = conn.execute(
"SELECT login_seq FROM users WHERE username='normaluser'").fetchone()[0]
conn.close()
r = app_env.post('/api/v1/auth/logout', json={})
r = app_env.post('/auth/logout', json={})
assert r.status_code == 200
conn = auth.get_conn()
seq_after = conn.execute(
@@ -95,7 +95,7 @@ def test_logout_bumps_login_seq_and_clears(app_env):
conn.close()
assert seq_after == seq_before + 1
# 旧 Cookie 失效
r2 = app_env.post('/api/v1/auth/heartbeat', json={})
r2 = app_env.post('/auth/heartbeat', json={})
assert r2.status_code == 401
@@ -104,8 +104,8 @@ def test_heartbeat_expired_credits(app_env):
conn.execute("UPDATE users SET duration_seconds=0 WHERE username='normaluser'")
conn.commit()
conn.close()
r = app_env.post('/api/v1/auth/login', json={'username': 'normaluser', 'password': USER_PW})
r = app_env.post('/auth/login', json={'username': 'normaluser', 'password': USER_PW})
assert r.status_code == 200
r2 = app_env.post('/api/v1/auth/heartbeat', json={})
r2 = app_env.post('/auth/heartbeat', json={})
assert r2.status_code == 401
assert r2.json().get('error_code') == 'expired'
+8 -8
View File
@@ -48,7 +48,7 @@ def _login(env, username):
base64 padding('=') 的值加双引号,导致服务端签名解析失败(真实浏览器不会),
故测试避开 jar。
"""
r = env.post('/api/v1/auth/login', json={'username': username, 'password': PW})
r = env.post('/auth/login', json={'username': username, 'password': PW})
assert r.status_code == 200, r.text
raw = r.headers.get('set-cookie', '')
cookie = ''
@@ -73,7 +73,7 @@ def _open_ledger_rows():
def test_download_requires_login(env):
r = env.post('/api/v1/download', json={'id': '347230'})
r = env.post('/download', json={'id': '347230'})
assert r.status_code == 401
@@ -81,7 +81,7 @@ def test_download_bad_id_refunds(env):
ck, _ = _login(env, 'payer')
H = {'Cookie': f'session={ck}'}
before = _balance('payer')
r = env.post('/api/v1/download', json={'id': 'not-a-number'}, headers=H)
r = env.post('/download', json={'id': 'not-a-number'}, headers=H)
assert r.status_code == 400
assert '无效' in r.json()['message']
# 预扣已退回,且无 open 账本行残留
@@ -93,7 +93,7 @@ def test_download_bad_quality_refunds(env):
ck, _ = _login(env, 'payer')
H = {'Cookie': f'session={ck}'}
before = _balance('payer')
r = env.post('/api/v1/download', json={'id': '347230', 'quality': 'nope'}, headers=H)
r = env.post('/download', json={'id': '347230', 'quality': 'nope'}, headers=H)
assert r.status_code == 400
assert _balance('payer') == before
assert _open_ledger_rows() == []
@@ -103,7 +103,7 @@ def test_download_bad_format_refunds(env):
ck, _ = _login(env, 'payer')
H = {'Cookie': f'session={ck}'}
before = _balance('payer')
r = env.post('/api/v1/download', json={'id': '347230', 'format': 'xml'}, headers=H)
r = env.post('/download', json={'id': '347230', 'format': 'xml'}, headers=H)
assert r.status_code == 400
assert _balance('payer') == before
@@ -111,7 +111,7 @@ def test_download_bad_format_refunds(env):
def test_download_insufficient_credits(env):
ck, _ = _login(env, 'broke')
H = {'Cookie': f'session={ck}'}
r = env.post('/api/v1/download', json={'id': '347230'}, headers=H)
r = env.post('/download', json={'id': '347230'}, headers=H)
assert r.status_code == 401
assert r.json().get('error_code') == 'expired'
# 未开账
@@ -124,7 +124,7 @@ def test_batch_zip_too_many_refunds(env):
H = {'Cookie': f'session={ck}'}
before = _balance('rich')
ids = ','.join(str(1000 + i) for i in range(501))
r = env.post('/api/v1/download/batch_zip', json={'ids': ids}, headers=H)
r = env.post('/download/batch_zip', json={'ids': ids}, headers=H)
assert r.status_code == 400
assert '500' in r.json()['message']
# 预扣(封顶 500)已全额退回,余额回到 600
@@ -136,6 +136,6 @@ def test_batch_zip_missing_ids_refunds(env):
ck, _ = _login(env, 'payer')
H = {'Cookie': f'session={ck}'}
before = _balance('payer')
r = env.post('/api/v1/download/batch_zip', json={}, headers=H)
r = env.post('/download/batch_zip', json={}, headers=H)
assert r.status_code == 400
assert _balance('payer') == before
+109
View File
@@ -0,0 +1,109 @@
# -*- coding: utf-8 -*-
"""黄金契约回归:FastAPI 严格复现 Flask 时代的响应。
tests/golden/responses.json 是**删除 Flask 之前**从 Flask 应用捕获的快照
(状态码 + 响应体)。本测试让 FastAPI(根路径模式)跑同一批请求,
逐字段断言一致——这样删除 Flask 巨石后,行为回归仍有硬证据。
说明:
- 会话 Cookie 在运行时用当前库的 login_seq 动态构造(快照不存 Cookie)。
- 请求为只读/确定性调用,不含实时上游数据(那些由 parity/单测覆盖)。
"""
import hashlib
import json
import sqlite3
import time
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
import auth_system
from core import session as cs
_GOLDEN = Path(__file__).parent / 'golden' / 'responses.json'
SECRET = 'golden-secret'
def _load_cases():
return json.loads(_GOLDEN.read_text(encoding='utf-8'))
def _session_for(role):
"""按当前库中该角色用户的 login_seq 造一份合法会话 Cookie。"""
conn = sqlite3.connect('data/site.db')
conn.row_factory = sqlite3.Row
try:
where = "role = 'admin'" if role == 'admin' else "role != 'admin' AND status = 1"
row = conn.execute(
f'SELECT id, login_seq FROM users WHERE {where} AND login_seq IS NOT NULL LIMIT 1'
).fetchone()
finally:
conn.close()
if row is None:
return None
return cs.dumps(SECRET, {
'site_uid': int(row['id']), 'site_uid_seq': int(row['login_seq']),
'site_uid_ts': int(time.time()),
})
@pytest.fixture()
def client(tmp_path, monkeypatch):
# 站点库仍用生产 data/site.db(只读),密钥 patch 成固定值以便造 Cookie
monkeypatch.setattr('web.deps.secret_key', lambda: SECRET)
monkeypatch.setattr(auth_system, 'get_secret_key', lambda: SECRET)
from web.app import create_app
return TestClient(create_app(api_prefix=''))
# 响应对比前剔除的易变字段(时钟)
_VOLATILE = {'timestamp'}
def _strip(obj):
if isinstance(obj, dict):
return {k: _strip(v) for k, v in obj.items() if k not in _VOLATILE}
if isinstance(obj, list):
return [_strip(v) for v in obj]
return obj
def _request(client, method, path, payload, cookie):
headers = {'Cookie': f'session={cookie}'} if cookie else {}
if method == 'GET':
return client.get(path, headers=headers)
return client.post(path, json=payload or {}, headers=headers)
def test_golden_contract(client):
"""逐例复现 Flask 快照。"""
cases = _load_cases()
failures = []
for c in cases:
if c.get('html_len'):
r = client.get(c['path'])
if r.status_code != c['status'] or len(r.content) != c['html_len']:
failures.append(f"{c['name']}: html {r.status_code}/{len(r.content)} "
f"!= {c['status']}/{c['html_len']}")
continue
cookie = None
if c['auth'] == 'admin':
cookie = _session_for('admin')
elif c['auth'] == 'user':
cookie = _session_for('user')
r = _request(client, c['method'], c['path'], c['payload'], cookie)
body = r.json() if r.headers.get('content-type', '').startswith('application/json') else None
if r.status_code != c['status']:
failures.append(f"{c['name']}: status {r.status_code} != {c['status']}")
elif _strip(body) != _strip(c['body']):
failures.append(f"{c['name']}: body\n fastapi={body}\n golden ={c['body']}")
assert not failures, '黄金契约回归失败:\n' + '\n'.join(failures)
def test_index_html_stable(client):
"""首页渲染逐字节稳定(模板未改时)。"""
cases = {c['name']: c for c in _load_cases()}
c = cases['index']
r = client.get('/')
assert hashlib.sha256(r.content).hexdigest() == c['html_sha256']
+11 -11
View File
@@ -30,7 +30,7 @@ def env(tmp_path, monkeypatch):
monkeypatch.setattr(db, 'DB_PATH', db_path)
auth.init_db()
monkeypatch.setattr('web.deps.secret_key', lambda: SECRET)
import auth_routes
import web.auth_state as auth_routes
auth_routes._SEND_CODE_IP.clear()
from web.app import create_app
return TestClient(create_app())
@@ -47,7 +47,7 @@ def _seed_code(email: str, code: str, purpose: str = 'register'):
def test_register_requires_agreement(env):
r = env.post('/api/v1/auth/register', json={
r = env.post('/auth/register', json={
'username': 'newuser', 'password': 'Passw0rd12', 'email': 'n@example.com',
'code': '000000', 'agree': '0'})
assert r.json()['message'].startswith('请先阅读并同意')
@@ -55,7 +55,7 @@ def test_register_requires_agreement(env):
def test_register_success(env):
_seed_code('new@example.com', '123456')
r = env.post('/api/v1/auth/register', json={
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
@@ -65,18 +65,18 @@ def test_register_success(env):
assert 'session' in r.cookies
# 会话可用
assert env.post('/api/v1/auth/heartbeat', json={}).status_code == 200
assert env.post('/auth/heartbeat', json={}).status_code == 200
# 用户名/邮箱已被占用
_seed_code('new@example.com', '222222')
r2 = env.post('/api/v1/auth/register', json={
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('/api/v1/auth/register', json={
r = env.post('/auth/register', json={
'username': 'another', 'password': 'Passw0rd12', 'email': 'a@example.com',
'code': '999999', 'agree': '1'})
assert r.json()['message'] == '邮箱验证码错误或已过期'
@@ -86,12 +86,12 @@ 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={
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('/api/v1/auth/login', json={'username': 'forgetme', 'password': 'NewPass456'})
r2 = env.post('/auth/login', json={'username': 'forgetme', 'password': 'NewPass456'})
assert r2.status_code == 200 and r2.json()['success'] is True
@@ -101,7 +101,7 @@ def test_forgot_password_admin_blocked(env):
conn.commit()
conn.close()
_seed_code('boss@example.com', '111111', purpose='reset')
r = env.post('/api/v1/auth/forgot_password', json={
r = env.post('/auth/forgot_password', json={
'email': 'boss@example.com', 'code': '111111', 'password': 'NewPass456'})
# 管理员不走自助找回:统一防枚举文案
assert r.json()['message'] == '邮箱验证码错误或已过期'
@@ -111,8 +111,8 @@ 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})
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
+1 -1
View File
@@ -62,6 +62,6 @@ def test_root_page_served(root_client):
def test_no_api_v1_prefix_in_root_mode(root_client):
"""root 模式下不再暴露 /api/v1 前缀(避免与 Flask 时代语义混淆)。"""
"""根路径模式下不再有 /api/v1 前缀(Flask 时代的比对前缀已废弃)。"""
r = root_client.post('/api/v1/song', json={})
assert r.status_code == 404
+33 -25
View File
@@ -1,54 +1,57 @@
# -*- coding: utf-8 -*-
"""Flask 兼容会话编解码的单元测试(不依赖网络)。"""
"""Flask 兼容会话编解码的单元测试。
验证 core.session 与 itsdangerous(Flask 会话所用库)**双向互通**:
Flask 时代签发的会话 Cookie 必须仍能被本站解析(用户不必重新登录)。
Flask 本身已从依赖中移除,这里直接用 itsdangerous 构造/校验
(Flask 会话即 itsdangerous URLSafeTimedSerializer:盐 cookie-session、HMAC-SHA1)。
"""
import hashlib
import time
import pytest
from flask import Flask
from flask.sessions import SecureCookieSessionInterface
from itsdangerous import URLSafeTimedSerializer
from core import session as cs
SECRET = 'unit-test-secret-key'
@pytest.fixture()
def flask_serializer():
app = Flask(__name__)
app.secret_key = SECRET
return SecureCookieSessionInterface().get_signing_serializer(app)
def _flask_serializer():
"""等价于 Flask 的会话序列化器(同盐、同摘要算法、同 key 派生)。"""
return URLSafeTimedSerializer(
SECRET, salt='cookie-session',
signer_kwargs={'key_derivation': 'hmac', 'digest_method': hashlib.sha1})
def _flask_cookie(ser, data):
return ser.dumps(data)
def test_flask_to_core(flask_serializer):
"""Flask 签发的会话 Cookie 能被 core.session 解出。"""
cookie = _flask_cookie(flask_serializer, {'site_uid': 26, 'site_uid_seq': 5})
def test_itsdangerous_to_core():
"""itsdangerous(Flask)签发的会话 Cookie 能被 core.session 解出。"""
cookie = _flask_serializer().dumps({'site_uid': 26, 'site_uid_seq': 5})
data = cs.loads(SECRET, cookie)
assert data == {'site_uid': 26, 'site_uid_seq': 5}
assert cs.get_uid(data) == 26
def test_core_to_flask(flask_serializer):
"""core.session 签发(含时间戳)的 Cookie 能被 Flask 解出。"""
def test_core_to_itsdangerous():
"""core.session 签发的 Cookie 能被 itsdangerous 解出。"""
cookie = cs.dumps(SECRET, {'site_uid': 74, 'site_uid_seq': 8})
assert flask_serializer.loads(cookie) == {'site_uid': 74, 'site_uid_seq': 8}
assert _flask_serializer().loads(cookie) == {'site_uid': 74, 'site_uid_seq': 8}
def test_core_to_flask_expiry(flask_serializer):
"""时间戳格式正确:Flask 能按 max_age 校验(未过期可解)。"""
def test_core_timestamp_valid_for_timed_check():
"""时间戳编码与 itsdangerous 一致:max_age 校验可用(未过期)。"""
cookie = cs.dumps(SECRET, {'site_uid': 1})
assert flask_serializer.loads(cookie, max_age=3600) == {'site_uid': 1}
assert _flask_serializer().loads(cookie, max_age=3600) == {'site_uid': 1}
def test_tampered_returns_none(flask_serializer):
cookie = _flask_cookie(flask_serializer, {'site_uid': 26})
def test_tampered_returns_none():
cookie = _flask_serializer().dumps({'site_uid': 26})
assert cs.loads(SECRET, cookie[:-3] + 'xyz') is None
def test_wrong_key_returns_none(flask_serializer):
cookie = _flask_cookie(flask_serializer, {'site_uid': 26})
def test_wrong_key_returns_none():
cookie = _flask_serializer().dumps({'site_uid': 26})
assert cs.loads('wrong-secret', cookie) is None
@@ -65,3 +68,8 @@ def test_roundtrip_selfconsistency():
def test_get_uid_none():
assert cs.get_uid(None) is None
assert cs.get_uid({}) is None
@pytest.mark.parametrize('uid', [1, 26, 99999])
def test_various_uids(uid):
assert cs.get_uid(cs.loads(SECRET, cs.dumps(SECRET, {'site_uid': uid}))) == uid