xiaoxia-agent
c7a34fb297
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (push) Successful in 0s
CI/CD Pipeline / Check push changed paths (push) Successful in 1s
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Build Staging Web Image (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Failing after 1m0s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 2m25s
CI/CD Pipeline / Integration Tests (push) Successful in 4m8s
CI/CD Pipeline / Validate - Python (mypy + alembic) (push) Successful in 5m16s
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Retag skipped Staging API Image (push) Has been cancelled
CI/CD Pipeline / Retag skipped Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Retag skipped Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
CI/CD Pipeline / CI Gate (push) Has been cancelled
CI/CD Pipeline / Validate - Style (push) Has been cancelled
CI/CD Pipeline / Validate - Security (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
P1 修复： 1. 抖音链接提取：后端自动从分享文本（如"这个视频 https://v.douyin.com/xxx/ 快来看看"） 中正则提取 http(s) URL，不再要求用户只粘贴纯链接； 裸域名（v.douyin.com/xxx）自动补 https://；ftp/file 等非 http(s) 协议拒绝。 2. 抖音文案提取换方案：优先走火山引擎 MediaKit asr-subtitles API（云端 ASR）， 利用 yt-dlp 解析无水印直链（不下载整段视频）→ 提交 MediaKit → 轮询拿字幕； MediaKit 未配置/失败时回退到原"下载视频+本地 ASR"路径；cookies 503 友好错误保留。 3. 文案库 P1-1 排查结论：后端 GET /api/v1/scripts CRUD 正常，scripts 表结构/数据 在 070/077/078 迁移链路中无数据丢失；"数据不显示"应是前端未正确切换到新接口， 已在回复中说明。 测试：新增 13 个 URL 提取单测；原有 15576 单测全绿；ruff clean.
433 lines
17 KiB
Python
433 lines
17 KiB
Python
"""验证 extract-from-douyin 在各种失败场景返回正确的 HTTP 状态码（绝不能 500）"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
import types
|
||
from unittest import mock
|
||
|
||
import pytest
|
||
from app.auth import AuthenticatedUser
|
||
from fastapi import HTTPException, status
|
||
|
||
|
||
class _FakeUser:
|
||
id = "u-test"
|
||
is_member = False
|
||
member_type = None
|
||
|
||
|
||
@pytest.fixture
|
||
def fake_user():
|
||
return AuthenticatedUser(user=_FakeUser())
|
||
|
||
|
||
class _FakeYDLBase:
|
||
"""通用假 yt-dlp 基类（支持上下文管理器 with 语法）"""
|
||
|
||
extract_info_result = None
|
||
extract_info_raises = None
|
||
prepare_filename_result = "/tmp/fake.mp4"
|
||
|
||
def __init__(self, *a, **kw):
|
||
pass
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *a):
|
||
return False
|
||
|
||
def extract_info(self, url, download=True):
|
||
if self.__class__.extract_info_raises:
|
||
raise self.__class__.extract_info_raises
|
||
return self.__class__.extract_info_result
|
||
|
||
def prepare_filename(self, info):
|
||
return self.__class__.prepare_filename_result
|
||
|
||
|
||
def _install_fake_ytdlp(fake_ydl_class, *, download_error_cls=None):
|
||
"""把假 yt-dlp 注入 sys.modules，函数内 import yt_dlp 会拿到我们的假版本"""
|
||
fake_mod = types.ModuleType("yt_dlp")
|
||
fake_mod.YoutubeDL = fake_ydl_class
|
||
if download_error_cls is None:
|
||
download_error_cls = type("DownloadError", (Exception,), {})
|
||
fake_mod.DownloadError = download_error_cls
|
||
utils_mod = types.ModuleType("yt_dlp.utils")
|
||
utils_mod.DownloadError = download_error_cls
|
||
fake_mod.utils = utils_mod
|
||
sys.modules["yt_dlp"] = fake_mod
|
||
sys.modules["yt_dlp.utils"] = utils_mod
|
||
return fake_mod
|
||
|
||
|
||
def _import_target():
|
||
from app.api.routes import scripts_ai
|
||
|
||
return scripts_ai
|
||
|
||
|
||
def _fake_mk_unavailable():
|
||
"""Mock MediaKitClient 不可用，强制走下载+本地 ASR 路径。"""
|
||
fake_mk = mock.MagicMock()
|
||
fake_mk.is_available = False
|
||
return mock.patch("app.api.routes.scripts_ai.get_mediakit_client", return_value=fake_mk)
|
||
|
||
|
||
def test_download_http404_returns_400_not_500(fake_user):
|
||
"""无效短链 / 视频 404 → 应返回 400 业务错误，不能 500"""
|
||
scripts_ai = _import_target()
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/test123/")
|
||
|
||
class DownloadError(Exception):
|
||
pass
|
||
|
||
class FailingYDL(_FakeYDLBase):
|
||
extract_info_raises = DownloadError("ERROR: Unable to download webpage: HTTP Error 404: Not Found")
|
||
|
||
_install_fake_ytdlp(FailingYDL, download_error_cls=DownloadError)
|
||
|
||
with _fake_mk_unavailable():
|
||
with pytest.raises(HTTPException) as exc:
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
assert (
|
||
exc.value.status_code == status.HTTP_400_BAD_REQUEST
|
||
), f"应为400，实际 {exc.value.status_code}: {exc.value.detail}"
|
||
assert "无法解析" in exc.value.detail or "抖音" in exc.value.detail
|
||
|
||
|
||
def test_download_network_error_returns_502_not_500(fake_user):
|
||
"""网络错误 / 上游异常 → 502，不能 500"""
|
||
scripts_ai = _import_target()
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
|
||
|
||
class DownloadError(Exception):
|
||
pass
|
||
|
||
class NetErrYDL(_FakeYDLBase):
|
||
# 路径A（元信息解析）会吞异常返回 None；路径B（下载）抛网络错误
|
||
@staticmethod
|
||
def _raise():
|
||
raise DownloadError("ERROR: Connection reset by peer")
|
||
|
||
def extract_info(self, url, download=True):
|
||
# 元信息探测返回 None（拿不到直链），下载时再抛
|
||
if not download:
|
||
return None
|
||
self._raise()
|
||
|
||
_install_fake_ytdlp(NetErrYDL, download_error_cls=DownloadError)
|
||
|
||
with _fake_mk_unavailable():
|
||
with pytest.raises(HTTPException) as exc:
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
assert exc.value.status_code == status.HTTP_502_BAD_GATEWAY
|
||
|
||
|
||
def test_info_none_returns_400(fake_user):
|
||
"""yt-dlp 返回 None info → 400"""
|
||
scripts_ai = _import_target()
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
|
||
|
||
class NoneInfoYDL(_FakeYDLBase):
|
||
def extract_info(self, url, download=True):
|
||
# 元信息探测返回 None；下载也返回 None
|
||
return None
|
||
|
||
_install_fake_ytdlp(NoneInfoYDL)
|
||
|
||
with _fake_mk_unavailable():
|
||
with pytest.raises(HTTPException) as exc:
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
assert exc.value.status_code == status.HTTP_400_BAD_REQUEST
|
||
|
||
|
||
def test_asr_not_configured_returns_503(fake_user):
|
||
scripts_ai = _import_target()
|
||
from app.services.script_asr_service import ASRNotConfiguredError
|
||
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
|
||
|
||
class OkYDL(_FakeYDLBase):
|
||
def extract_info(self, url, download=True):
|
||
# 元信息返回 None（不走 MediaKit）；下载返回正常 info
|
||
if not download:
|
||
return None
|
||
return {"id": "x", "duration": 10, "title": "t"}
|
||
|
||
_install_fake_ytdlp(OkYDL)
|
||
with (
|
||
_fake_mk_unavailable(),
|
||
mock.patch.object(scripts_ai.os.path, "isfile", return_value=True),
|
||
mock.patch.object(scripts_ai.os.path, "getsize", return_value=1024),
|
||
mock.patch.object(scripts_ai, "transcribe_to_text", side_effect=ASRNotConfiguredError("未配置")),
|
||
):
|
||
with pytest.raises(HTTPException) as exc:
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
assert exc.value.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
|
||
|
||
|
||
def test_asr_failure_returns_502(fake_user):
|
||
scripts_ai = _import_target()
|
||
from app.services.script_asr_service import ASRTranscriptionError
|
||
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
|
||
|
||
class OkYDL(_FakeYDLBase):
|
||
def extract_info(self, url, download=True):
|
||
if not download:
|
||
return None
|
||
return {"id": "x", "duration": 10, "title": "t"}
|
||
|
||
_install_fake_ytdlp(OkYDL)
|
||
with (
|
||
_fake_mk_unavailable(),
|
||
mock.patch.object(scripts_ai.os.path, "isfile", return_value=True),
|
||
mock.patch.object(scripts_ai.os.path, "getsize", return_value=1024),
|
||
mock.patch.object(scripts_ai, "transcribe_to_text", side_effect=ASRTranscriptionError("识别失败")),
|
||
):
|
||
with pytest.raises(HTTPException) as exc:
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
assert exc.value.status_code == status.HTTP_502_BAD_GATEWAY
|
||
|
||
|
||
def test_asr_unexpected_error_returns_502_not_500(fake_user):
|
||
"""ASR 抛未预期异常也应被兜住，不能 500"""
|
||
scripts_ai = _import_target()
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
|
||
|
||
class OkYDL(_FakeYDLBase):
|
||
def extract_info(self, url, download=True):
|
||
if not download:
|
||
return None
|
||
return {"id": "x", "duration": 10, "title": "t"}
|
||
|
||
_install_fake_ytdlp(OkYDL)
|
||
with (
|
||
_fake_mk_unavailable(),
|
||
mock.patch.object(scripts_ai.os.path, "isfile", return_value=True),
|
||
mock.patch.object(scripts_ai.os.path, "getsize", return_value=1024),
|
||
mock.patch.object(scripts_ai, "transcribe_to_text", side_effect=RuntimeError("ffmpeg crashed")),
|
||
):
|
||
with pytest.raises(HTTPException) as exc:
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
assert exc.value.status_code == status.HTTP_502_BAD_GATEWAY, f"应为502，实际 {exc.value.status_code}"
|
||
|
||
|
||
def test_missing_downloaded_file_returns_502_not_500(fake_user):
|
||
"""yt-dlp 返回 info 但文件未落地（isfile False）→ 502"""
|
||
scripts_ai = _import_target()
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
|
||
|
||
class OkYDL(_FakeYDLBase):
|
||
def extract_info(self, url, download=True):
|
||
if not download:
|
||
return None
|
||
return {"id": "x", "duration": 10, "title": "t"}
|
||
|
||
_install_fake_ytdlp(OkYDL)
|
||
with (
|
||
_fake_mk_unavailable(),
|
||
mock.patch.object(scripts_ai.os.path, "isfile", return_value=False),
|
||
):
|
||
with pytest.raises(HTTPException) as exc:
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
assert exc.value.status_code != 500
|
||
assert "下载异常" in exc.value.detail or "文件" in exc.value.detail
|
||
|
||
|
||
def test_any_unexpected_error_does_not_return_500_raw(fake_user):
|
||
"""兜底：prepare_filename 抛未预期异常也应被捕获，返回500 code但含业务detail"""
|
||
scripts_ai = _import_target()
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
|
||
|
||
class BuggyYDL(_FakeYDLBase):
|
||
def extract_info(self, url, download=True):
|
||
if not download:
|
||
return None
|
||
return {"id": "x", "duration": "not_a_number", "title": "t"}
|
||
|
||
def prepare_filename(self, info):
|
||
raise RuntimeError("some internal bug")
|
||
|
||
_install_fake_ytdlp(BuggyYDL)
|
||
with _fake_mk_unavailable():
|
||
with pytest.raises(HTTPException) as exc:
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
assert "抖音" in exc.value.detail or "失败" in exc.value.detail or exc.value.status_code != 500
|
||
|
||
|
||
# ── cookies 相关测试 ──────────────────────────────────────────────
|
||
|
||
|
||
def test_cookies_error_returns_503_friendly_message(fake_user):
|
||
"""cookies 缺失/过期 → 返回 503 + 友好文案，不暴露原始错误"""
|
||
scripts_ai = _import_target()
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/test123/")
|
||
|
||
class DownloadError(Exception):
|
||
pass
|
||
|
||
class CookiesYDL(_FakeYDLBase):
|
||
def extract_info(self, url, download=True):
|
||
raise DownloadError(
|
||
"ERROR: [Douyin] 7623712911260650802: Fresh cookies (not necessarily logged in) are needed"
|
||
)
|
||
|
||
_install_fake_ytdlp(CookiesYDL, download_error_cls=DownloadError)
|
||
|
||
with _fake_mk_unavailable():
|
||
with pytest.raises(HTTPException) as exc:
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
assert exc.value.status_code == status.HTTP_503_SERVICE_UNAVAILABLE, f"应为503，实际 {exc.value.status_code}"
|
||
assert (
|
||
"暂时不可用" in exc.value.detail or "稍后重试" in exc.value.detail
|
||
), f"应有友好提示，实际: {exc.value.detail}"
|
||
assert "Fresh cookies" not in exc.value.detail
|
||
|
||
|
||
def test_cookies_error_in_generic_except_also_returns_503(fake_user):
|
||
"""cookies 错误绕过 DownloadError 时，兜底异常分支也应识别并返回 503"""
|
||
scripts_ai = _import_target()
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
|
||
|
||
class CookieBugYDL(_FakeYDLBase):
|
||
def extract_info(self, url, download=True):
|
||
raise RuntimeError("Fresh cookies are needed to access this video")
|
||
|
||
_install_fake_ytdlp(CookieBugYDL)
|
||
|
||
with _fake_mk_unavailable():
|
||
with pytest.raises(HTTPException) as exc:
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
assert exc.value.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
|
||
|
||
|
||
def test_ydl_opts_includes_cookiefile_when_file_exists(fake_user):
|
||
"""cookies 文件存在时，ydl_opts 应包含 cookiefile 指向该路径（在下载分支）"""
|
||
scripts_ai = _import_target()
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
|
||
|
||
captured_opts_download = {}
|
||
|
||
class CaptureOptsYDL(_FakeYDLBase):
|
||
def __init__(self, opts):
|
||
# 下载分支会触发 download=True；元信息探测 download=False
|
||
# 元信息也会传 cookiefile，但我们只在下载分支记录（更接近真实）
|
||
super().__init__()
|
||
self._opts = opts
|
||
# 总是记录最后一次的 opts，方便断言
|
||
captured_opts_download.clear()
|
||
captured_opts_download.update(opts)
|
||
|
||
def extract_info(self, url, download=True):
|
||
if not download:
|
||
return None # 元信息失败，走下载分支
|
||
return {"id": "x", "duration": 5, "title": "t"}
|
||
|
||
_install_fake_ytdlp(CaptureOptsYDL)
|
||
|
||
with (
|
||
_fake_mk_unavailable(),
|
||
mock.patch.object(scripts_ai, "_resolve_cookies_file", return_value="/tmp/fake_cookies.txt"),
|
||
mock.patch.object(scripts_ai.os.path, "isfile", return_value=True),
|
||
mock.patch.object(scripts_ai.os.path, "getsize", return_value=1024),
|
||
mock.patch.object(scripts_ai, "transcribe_to_text", return_value="ok"),
|
||
):
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
|
||
assert captured_opts_download.get("cookiefile") == "/tmp/fake_cookies.txt", f"cookiefile 应被设置，opts={captured_opts_download}"
|
||
|
||
|
||
def test_ydl_opts_no_cookiefile_when_file_missing(fake_user):
|
||
"""cookies 文件不存在时，ydl_opts 不应包含 cookiefile 键"""
|
||
scripts_ai = _import_target()
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
|
||
|
||
captured_opts_download = {}
|
||
|
||
class CaptureOptsYDL(_FakeYDLBase):
|
||
def __init__(self, opts):
|
||
super().__init__()
|
||
captured_opts_download.clear()
|
||
captured_opts_download.update(opts)
|
||
|
||
def extract_info(self, url, download=True):
|
||
if not download:
|
||
return None
|
||
return {"id": "x", "duration": 5, "title": "t"}
|
||
|
||
_install_fake_ytdlp(CaptureOptsYDL)
|
||
|
||
with (
|
||
_fake_mk_unavailable(),
|
||
mock.patch.object(scripts_ai, "_resolve_cookies_file", return_value=None),
|
||
mock.patch.object(scripts_ai.os.path, "isfile", return_value=True),
|
||
mock.patch.object(scripts_ai.os.path, "getsize", return_value=1024),
|
||
mock.patch.object(scripts_ai, "transcribe_to_text", return_value="ok"),
|
||
):
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
|
||
assert "cookiefile" not in captured_opts_download, f"cookies 文件缺失时不应设置 cookiefile，opts={captured_opts_download}"
|
||
|
||
|
||
def test_generic_download_error_hides_raw_message(fake_user):
|
||
"""非 cookies 非 404 的通用下载错误 → 502，且不暴露 yt-dlp 原始错误文本"""
|
||
scripts_ai = _import_target()
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="https://v.douyin.com/abc/")
|
||
|
||
class DownloadError(Exception):
|
||
pass
|
||
|
||
class GenErrYDL(_FakeYDLBase):
|
||
def extract_info(self, url, download=True):
|
||
raise DownloadError("ERROR: some internal yt-dlp weird failure with trace")
|
||
|
||
_install_fake_ytdlp(GenErrYDL, download_error_cls=DownloadError)
|
||
|
||
with _fake_mk_unavailable():
|
||
with pytest.raises(HTTPException) as exc:
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
assert exc.value.status_code == status.HTTP_502_BAD_GATEWAY
|
||
assert "下载失败" in exc.value.detail
|
||
assert "weird failure" not in exc.value.detail, "不应暴露 yt-dlp 内部错误文本"
|
||
|
||
|
||
def test_share_text_input_extracts_url_correctly(fake_user):
|
||
"""分享文本（含前后说明文字）应能正确提取 URL"""
|
||
scripts_ai = _import_target()
|
||
share_text = "这个视频太搞笑了 https://v.douyin.com/abcdeFG/ 快来看看！#搞笑 #日常"
|
||
body = scripts_ai.ExtractFromDouyinRequest(url=share_text)
|
||
|
||
class OkYDL(_FakeYDLBase):
|
||
def extract_info(self, url, download=True):
|
||
if not download:
|
||
return {"url": "https://example.com/direct.mp4", "duration": 5}
|
||
return {"id": "x", "duration": 5, "title": "t"}
|
||
|
||
_install_fake_ytdlp(OkYDL)
|
||
fake_mk = mock.MagicMock()
|
||
fake_mk.is_available = True
|
||
fake_mk.asr_submit.return_value = "tk1"
|
||
fake_mk.asr_poll.return_value = ("识别成功的文案", 5.0)
|
||
with (
|
||
mock.patch("app.api.routes.scripts_ai.get_mediakit_client", return_value=fake_mk),
|
||
mock.patch.object(scripts_ai, "_ytdlp_extract_video_url", return_value=("https://example.com/direct.mp4", 5.0)),
|
||
):
|
||
resp = scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
assert resp.source_url == "https://v.douyin.com/abcdeFG/"
|
||
assert resp.text == "识别成功的文案"
|
||
assert resp.duration_seconds == 5.0
|
||
|
||
|
||
def test_non_douyin_share_text_returns_400(fake_user):
|
||
"""粘贴非抖音分享链接 → 400"""
|
||
scripts_ai = _import_target()
|
||
body = scripts_ai.ExtractFromDouyinRequest(url="看看这个 https://www.bilibili.com/video/BV1xx 哈哈哈")
|
||
|
||
with _fake_mk_unavailable():
|
||
with pytest.raises(HTTPException) as exc:
|
||
scripts_ai.extract_from_douyin(request=body, current_user=fake_user, db=mock.MagicMock())
|
||
assert exc.value.status_code == 400
|