你的下载器改了一行代码,怎么知道没搞坏其他功能?——视频下载器的测试策略
你给下载器加了一个新平台的解析器,结果 B 站的下载挂了。你改了重试逻辑,结果进度条不更新了。这种"修一个 bug 引出三个新 bug"的事,每个开发者都经历过。根源在于没有自动化测试——每次改动后只能手动点点点验证,累且不可靠。这篇文章讲怎么给视频下载器写测试:从最简单的单元测试到 Mock 网络请求、从 CI 自动运行到"解析器健康检查"——花一个下午把测试框架搭好,以后每次改代码跑一遍,心里有底。
TL;DR:下载器的测试分三层:① 单元测试(pytest):测解析器逻辑(BV 号转换、签名算法、URL 提取),Mock 掉网络请求;② 集成测试:用真实 URL 测完整下载流程,但限制只下载前几秒;③ 健康检查:定时跑 CI 验证所有平台解析器是否仍然有效。关键技巧:用
responses或pytest-httpxMock HTTP,用yt-dlp --playlist-end 1只获取信息不下载完整视频。
目录
- 一、下载器测试的难点
- 二、单元测试:测解析逻辑,Mock 网络
- 三、集成测试:真实下载但不下载全部
- 四、解析器健康检查:平台改版早知道
- 五、CI 集成:GitHub Actions 自动跑测试
- 六、测试覆盖率与持续改进
- 七、合规与温馨提示
一、下载器测试的难点
视频下载器比普通 Web 应用更难测试:
| 难点 | 说明 |
|---|---|
| 依赖外部服务 | 解析器依赖 B 站/YouTube 的 API,这些 API 随时可能变 |
| 网络不稳定 | 测试跑一半断网了,是代码 bug 还是网络问题? |
| 大文件 | 下载一个完整视频要几分钟,测试不能等这么久 |
| 需要登录态 | 部分接口需要 Cookie,测试环境没有 |
| 反爬机制 | 频繁测试可能触发风控,导致 IP 被封 |
解决方案:Mock 掉一切外部依赖——单元测试不访问真实网络,集成测试只下载少量数据,健康检查用独立 IP 低频运行。
二、单元测试:测解析逻辑,Mock 网络
2.1 Mock HTTP 请求
# test_bilibili.py
import pytest
import responses
import json
# 假设你的 B 站解析器
from my_downloader.parsers.bilibili import BilibiliParser
class TestBilibiliParser:
"""B 站解析器单元测试"""
@responses.activate
def test_bv_to_av(self):
"""测试 BV 号转 AV 号(纯计算,不需要 Mock)"""
parser = BilibiliParser()
assert parser.bv_to_av('BV1xx411c7mD') == 'av170001'
@responses.activate
def test_extract_video_id(self):
"""测试从各种 URL 中提取 BV 号"""
parser = BilibiliParser()
# 短链接
assert parser.extract_bvid('https://b23.tv/xxxxx') is not None
# 长链接
assert parser.extract_bvid(
'https://www.bilibili.com/video/BV1xx411c7mD'
) == 'BV1xx411c7mD'
@responses.activate
def test_parse_video_info(self):
"""测试解析视频信息(Mock API 响应)"""
parser = BilibiliParser()
# Mock 视频信息 API
responses.add(
responses.GET,
'https://api.bilibili.com/x/web-interface/view',
json={
'code': 0,
'data': {
'bvid': 'BV1xx411c7mD',
'title': '测试视频',
'duration': 120,
'pages': [{'cid': 123456, 'part': 'P1', 'page': 1}],
}
},
status=200,
)
info = parser.get_video_info('BV1xx411c7mD')
assert info['title'] == '测试视频'
assert info['duration'] == 120
assert len(info['pages']) == 1
@responses.activate
def test_parse_playurl(self):
"""测试解析播放地址(Mock playurl API)"""
parser = BilibiliParser()
# Mock playurl API
responses.add(
responses.GET,
'https://api.bilibili.com/x/player/wbi/v2',
json={
'code': 0,
'data': {
'dash': {
'video': [{
'id': 112,
'base_url': 'https://cdn.example.com/video.mp4',
'bandwidth': 5800000,
'width': 1920,
'height': 1080,
}],
'audio': [{
'id': 30280,
'base_url': 'https://cdn.example.com/audio.mp4',
'bandwidth': 320000,
}],
}
}
},
status=200,
)
video_url, audio_url = parser.get_playurl('BV1xx411c7mD', 123456)
assert 'cdn.example.com' in video_url
assert 'cdn.example.com' in audio_url
@responses.activate
def test_api_error_handling(self):
"""测试 API 返回错误时的处理"""
parser = BilibiliParser()
# Mock 错误响应
responses.add(
responses.GET,
'https://api.bilibili.com/x/web-interface/view',
json={'code': -404, 'message': '视频不存在'},
status=200,
)
with pytest.raises(ValueError, match='视频不存在'):
parser.get_video_info('BV0000000000')
2.2 用 pytest-httpx(推荐)
responses 库只能 Mock requests,如果你的代码用了 httpx 或 aiohttp,用 pytest-httpx:
# pip install pytest-httpx
import pytest
import httpx
from my_downloader.parsers.douyin import DouyinParser
@pytest.mark.asyncio
async def test_douyin_parse(httpx_mock):
"""测试抖音解析(Mock httpx)"""
# Mock 短链接重定向
httpx_mock.add_response(
url='https://v.douyin.com/xxxxx/',
status_code=302,
headers={
'Location': 'https://www.douyin.com/video/123456789'
},
)
# Mock 视频信息 API
httpx_mock.add_response(
url__contains='douyin.com/aweme/v1/web/aweme/detail',
json={
'aweme_detail': {
'desc': '测试视频',
'video': {
'play_addr': {
'url_list': [
'https://cdn.douyin.com/video.mp4'
]
}
}
}
},
)
parser = DouyinParser()
result = await parser.parse('https://v.douyin.com/xxxxx/')
assert result['title'] == '测试视频'
assert 'cdn.douyin.com' in result['video_url']
2.3 测试签名算法
class TestWBISignature:
"""WBI 签名算法测试"""
def test_mix_key_calculation(self):
"""测试 mix_key 拼接逻辑"""
img_key = '7cd084941338484aae1ad9425b84077b'
sub_key = '4932caff0ff746eab6f01bf08b70ac45'
mix_key = calculate_mix_key(img_key, sub_key)
# 已知的正确结果(需要预先计算好)
expected = '7c48aa1d5b073a2af0f7e6b0b04c5'
assert mix_key == expected
def test_wbi_sign(self):
"""测试完整签名流程"""
params = {
'bvid': 'BV1xx411c7mD',
'cid': '123456',
'fnval': '4048',
}
mix_key = '7c48aa1d5b073a2af0f7e6b0b04c5'
signed = sign_params(params, mix_key, wts=1735689600)
assert 'w_rid' in signed
assert 'wts' in signed
assert signed['wts'] == 1735689600
assert len(signed['w_rid']) == 32 # MD5 是 32 字符
三、集成测试:真实下载但不下载全部
3.1 只获取信息不下载
class TestIntegration:
"""集成测试——用真实网络,但限制下载量"""
@pytest.mark.integration
@pytest.mark.slow
def test_bilibili_real_parse(self):
"""真实测试 B 站解析(只获取信息,不下载视频)"""
import yt_dlp
ydl_opts = {
'quiet': True,
'skip_download': True, # 关键:不下载
'playlist_end': 1, # 只处理第一个视频
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(
'https://www.bilibili.com/video/BV1GJ411x7h7',
download=False
)
assert info is not None
assert 'title' in info
assert info['title'] != ''
@pytest.mark.integration
def test_download_first_few_seconds(self, tmp_path):
"""测试下载前几秒(验证完整链路)"""
import subprocess
url = 'https://example.com/video.mp4'
output = str(tmp_path / 'test.mp4')
# 只下载前 5 秒
subprocess.run([
'ffmpeg', '-i', url, '-t', '5',
'-c', 'copy', output, '-y'
], check=True, timeout=30)
assert os.path.exists(output)
assert os.path.getsize(output) > 1000
3.2 测试下载器的完整流程
@pytest.mark.integration
def test_full_download_pipeline(tmp_path):
"""测试完整下载管线:解析 → 下载 → 合并 → 校验"""
from my_downloader import Downloader
dl = Downloader(output_dir=str(tmp_path))
# 使用一个稳定的测试视频(可以是自己上传的)
test_url = 'https://www.youtube.com/watch?v=jNQXAC9IVRw' # YouTube 第一个视频
try:
result = dl.download(test_url)
assert result['success'] is True
assert os.path.exists(result['filepath'])
assert os.path.getsize(result['filepath']) > 1024
# 校验视频有效性
assert validate_video(result['filepath'])
except Exception as e:
# 网络问题或 API 变更,不算测试失败
pytest.skip(f"集成测试跳过(外部服务不可用): {e}")
四、解析器健康检查:平台改版早知道
4.1 健康检查脚本
#!/usr/bin/env python3
"""解析器健康检查——每天跑一次,发现平台改版第一时间告警"""
import sys
import json
import requests
from datetime import datetime
# 每个平台的测试用例
TEST_CASES = [
{
'name': 'B站-普通视频',
'parser': 'bilibili',
'url': 'https://www.bilibili.com/video/BV1GJ411x7h7',
'checks': ['title', 'video_url', 'audio_url'],
},
{
'name': 'B站-番剧',
'parser': 'bilibili',
'url': 'https://www.bilibili.com/bangumi/play/ep123456',
'checks': ['title', 'video_url'],
},
{
'name': 'YouTube-普通视频',
'parser': 'youtube',
'url': 'https://www.youtube.com/watch?v=jNQXAC9IVRw',
'checks': ['title', 'video_url', 'audio_url'],
},
{
'name': '抖音-网页版',
'parser': 'douyin',
'url': 'https://www.douyin.com/video/723456789',
'checks': ['video_url'],
},
]
def run_health_check(test_case):
"""运行单个健康检查"""
result = {
'name': test_case['name'],
'parser': test_case['parser'],
'url': test_case['url'],
'status': 'unknown',
'error': None,
'timestamp': datetime.now().isoformat(),
}
try:
# 动态导入解析器
parser_module = __import__(
f'my_downloader.parsers.{test_case["parser"]}',
fromlist=['parse']
)
data = parser_module.parse(test_case['url'])
# 检查必需字段
for field in test_case['checks']:
if field not in data or not data[field]:
result['status'] = 'failed'
result['error'] = f'缺少字段: {field}'
return result
result['status'] = 'ok'
except Exception as e:
result['status'] = 'failed'
result['error'] = str(e)[:200]
return result
def main():
results = []
for tc in TEST_CASES:
print(f"检查: {tc['name']}...", end=' ')
r = run_health_check(tc)
results.append(r)
print('✓' if r['status'] == 'ok' else f'✗ {r["error"]}')
# 保存结果
report = {
'date': datetime.now().isoformat(),
'total': len(results),
'ok': sum(1 for r in results if r['status'] == 'ok'),
'failed': sum(1 for r in results if r['status'] == 'failed'),
'results': results,
}
with open('health_report.json', 'w') as f:
json.dump(report, f, ensure_ascii=False, indent=2)
# 有失败时发告警
failed = [r for r in results if r['status'] == 'failed']
if failed:
send_alert(f"解析器健康检查失败: {len(failed)} 个平台")
sys.exit(1)
print(f"\n全部通过: {report['ok']}/{report['total']}")
if __name__ == '__main__':
main()
五、CI 集成:GitHub Actions 自动跑测试
# .github/workflows/test.yml
name: 测试
on:
push:
branches: [main]
pull_request:
branches: [main]
schedule:
# 每天凌晨跑一次健康检查
- cron: '0 2 * * *'
jobs:
unit-test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: 安装依赖
run: pip install -r requirements.txt pytest pytest-cov
- name: 单元测试
run: pytest tests/ -v --cov=my_downloader --cov-report=term
- name: 覆盖率报告
run: pytest tests/ --cov=my_downloader --cov-report=xml
integration-test:
runs-on: ubuntu-latest
# 只在 schedule 或手动触发时跑(避免每次 PR 都跑)
if: github.event_name == 'schedule' || github.event_name == 'workflow_dispatch'
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: 安装依赖
run: |
sudo apt-get update && sudo apt-get install -y ffmpeg
pip install -r requirements.txt
- name: 集成测试
run: pytest tests/ -v -m integration
health-check:
runs-on: ubuntu-latest
if: github.event_name == 'schedule'
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: 安装依赖
run: pip install -r requirements.txt
- name: 健康检查
run: python scripts/health_check.py
env:
BILIBILI_COOKIE: ${{ secrets.BILIBILI_COOKIE }}
YOUTUBE_COOKIE: ${{ secrets.YOUTUBE_COOKIE }}
- name: 告警(如果失败)
if: failure()
uses: appleboy/telegram-action@master
with:
to: ${{ secrets.TELEGRAM_CHAT_ID }}
token: ${{ secrets.TELEGRAM_BOT_TOKEN }}
message: "解析器健康检查失败!请检查 CI 日志。"
pytest 配置
# pytest.ini
[pytest]
markers =
slow: 慢速测试
integration: 集成测试(需要网络)
unit: 单元测试
# 默认只跑单元测试
addopts = -m "not integration"
六、测试覆盖率与持续改进
# 生成覆盖率报告
pytest --cov=my_downloader --cov-report=html tests/
# 打开 htmlcov/index.html 查看
# 推荐覆盖率目标:
# 解析器核心逻辑:> 80%
# URL 提取/转换:> 90%
# 工具函数:> 70%
# 整体:> 60%(下载器天然有大量难以测试的 IO 代码)
好的测试应该:
1. 快(单元测试 < 1 秒,全部测试 < 5 分钟)
2. 独立(不依赖外部服务,除了专门的集成测试)
3. 可重复(同一条测试跑 10 次结果一样)
4. 有意义的断言(不只是 assert True)
七、合规与温馨提示
- 测试时使用的视频 URL 应该是测试专用或已获授权的内容
- 集成测试和健康检查会访问真实平台 API,请合理设置频率(每天 1-2 次即可)
- 不要把包含 Cookie 的测试代码提交到公开仓库
- 更多讨论见 下载视频算侵权吗?聊聊个人备份与版权的那条线
测试这件事,投入和回报严重不成正比——投入的时间是"今天下午",回报的时间是"未来每次改代码不用手动验证"。我的建议是:先从解析器核心逻辑开始写单元测试,因为这部分最容易出 bug 也最容易 Mock。等覆盖率达到 60% 以上,再加集成测试和 CI。别追求 100% 覆盖率,那不现实也不划算。
本文由 VidDown 技术博客原创发布。VidDown 的解析引擎有完整的单元测试和健康检查体系——每个平台解析器在发布前都经过自动化测试验证。访问 VidDown 了解更多。