提示

返回博客列表

你的下载器改了一行代码,怎么知道没搞坏其他功能?——视频下载器的测试策略

你的下载器改了一行代码,怎么知道没搞坏其他功能?——视频下载器的测试策略

你给下载器加了一个新平台的解析器,结果 B 站的下载挂了。你改了重试逻辑,结果进度条不更新了。这种"修一个 bug 引出三个新 bug"的事,每个开发者都经历过。根源在于没有自动化测试——每次改动后只能手动点点点验证,累且不可靠。这篇文章讲怎么给视频下载器写测试:从最简单的单元测试到 Mock 网络请求、从 CI 自动运行到"解析器健康检查"——花一个下午把测试框架搭好,以后每次改代码跑一遍,心里有底。

TL;DR:下载器的测试分三层:① 单元测试(pytest):测解析器逻辑(BV 号转换、签名算法、URL 提取),Mock 掉网络请求;② 集成测试:用真实 URL 测完整下载流程,但限制只下载前几秒;③ 健康检查:定时跑 CI 验证所有平台解析器是否仍然有效。关键技巧:用 responsespytest-httpx Mock HTTP,用 yt-dlp --playlist-end 1 只获取信息不下载完整视频。

目录

一、下载器测试的难点

视频下载器比普通 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,如果你的代码用了 httpxaiohttp,用 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)

七、合规与温馨提示


测试这件事,投入和回报严重不成正比——投入的时间是"今天下午",回报的时间是"未来每次改代码不用手动验证"。我的建议是:先从解析器核心逻辑开始写单元测试,因为这部分最容易出 bug 也最容易 Mock。等覆盖率达到 60% 以上,再加集成测试和 CI。别追求 100% 覆盖率,那不现实也不划算。

本文由 VidDown 技术博客原创发布。VidDown 的解析引擎有完整的单元测试和健康检查体系——每个平台解析器在发布前都经过自动化测试验证。访问 VidDown 了解更多。

想亲手试试?用 VidDown 一键解析下载

粘贴视频链接即可解析,多平台支持、网页端即用;下载桌面客户端解锁海外平台本地解析,开通会员更享不限次下载。

顶部