xiaoxia-saas/tests/unit/test_scripts_ai.py

"""Scripts AI 能力单元测试 — Issue #1893.

测试覆盖：
- extract_from_douyin: 成功/非法URL/下载失败/ASR未配置/ASR失败
- ai_rewrite: 成功/空内容/LLM失败
- ai_generate_titles: 成功/count截断/空内容

所有外部调用（yt_dlp、ASR、LLM）均通过 unittest.mock.patch 隔离。
"""

from __future__ import annotations

import sys
from unittest.mock import MagicMock, patch

import pydantic
import pytest

import packages.middleware.points_gate as _pg_module

sys.path.insert(0, "apps/api")


def _make_auth_user(user_id: str = "u1"):
    """构造 mock AuthenticatedUser."""
    user = MagicMock()
    user.id = user_id
    auth = MagicMock()
    auth.user = user
    return auth


@pytest.fixture(autouse=True)
def _disable_points_gate(monkeypatch):
    """默认关闭积分闸门，避免影响既有用例。"""
    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: False)
    yield


@pytest.fixture
def mock_db():
    return MagicMock()


def _mock_youtube_dl(
    extract_info_return=None,
    extract_info_side_effect=None,
    prepare_filename_return="/tmp/douyin_extract_abc/abc123.mp4",
):
    """构造 yt_dlp.YoutubeDL 的 mock.

    路由中用法: ydl = yt_dlp.YoutubeDL(opts); info = ydl.extract_info(...)
    所以 mock_ydl_cls.return_value 就是 ydl 实例.
    """
    mock_ydl_instance = MagicMock()
    if extract_info_side_effect is not None:
        mock_ydl_instance.extract_info.side_effect = extract_info_side_effect
    else:
        mock_ydl_instance.extract_info.return_value = extract_info_return or {
            "id": "abc123",
            "duration": 120.5,
        }
    mock_ydl_instance.prepare_filename.return_value = prepare_filename_return
    return mock_ydl_instance


# ── extract_from_douyin ──────────────────────────────────────────────────────


class TestExtractFromDouyin:
    """POST /extract-from-douyin 测试."""

    @patch("app.api.routes.scripts_ai.get_mediakit_client")
    @patch("app.api.routes.scripts_ai.transcribe_to_text")
    @patch("tempfile.TemporaryDirectory")
    @patch("yt_dlp.YoutubeDL")
    @patch("app.api.routes.scripts_ai.os.path.getsize", return_value=1024)
    @patch("app.api.routes.scripts_ai.os.path.isfile", return_value=True)
    @patch("app.api.routes.scripts_ai._ytdlp_extract_video_url", return_value=(None, 0.0))
    def test_extract_from_douyin_success(
        self,
        mock_meta,
        mock_isfile,
        mock_getsize,
        mock_ydl_cls,
        mock_tempdir,
        mock_transcribe,
        mock_get_mk,
    ):
        """正常流程（MediaKit不可用，走本地下载+ASR）：下载视频 + ASR 转写成功."""
        from app.api.routes.scripts_ai import extract_from_douyin
        from app.schemas.scripts_ai import ExtractFromDouyinRequest

        fake_mk = MagicMock()
        fake_mk.is_available = False
        mock_get_mk.return_value = fake_mk

        mock_ydl_cls.return_value = _mock_youtube_dl(
            extract_info_return={"id": "abc123", "duration": 120.5},
        )
        mock_ydl_cls.return_value.__enter__ = MagicMock(return_value=mock_ydl_cls.return_value)
        mock_ydl_cls.return_value.__exit__ = MagicMock(return_value=False)

        mock_td = MagicMock()
        mock_td.__enter__ = MagicMock(return_value="/tmp/douyin_extract_abc")
        mock_td.__exit__ = MagicMock(return_value=False)
        mock_tempdir.return_value = mock_td

        mock_transcribe.return_value = "这是一段测试文案内容"

        req = ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/")
        auth = _make_auth_user()
        result = extract_from_douyin(request=req, current_user=auth, db=MagicMock())

        assert result.text == "这是一段测试文案内容"
        assert result.duration_seconds == 120.5
        assert result.source_url == "https://v.douyin.com/xxxxx/"
        mock_transcribe.assert_called_once()

    @pytest.mark.parametrize(
        "bad_url",
        [
            "",
            "not-a-url",
            "https://www.youtube.com/watch?v=abc",
            "https://www.bilibili.com/video/BV123",
            "ftp://v.douyin.com/xxx/",
        ],
    )
    def test_extract_from_douyin_invalid_url(self, bad_url):
        """非法 URL 返回 400."""
        from app.api.routes.scripts_ai import extract_from_douyin
        from app.schemas.scripts_ai import ExtractFromDouyinRequest
        from fastapi import HTTPException

        req = ExtractFromDouyinRequest(url=bad_url)
        auth = _make_auth_user()

        with pytest.raises(HTTPException) as exc_info:
            extract_from_douyin(request=req, current_user=auth)
        assert exc_info.value.status_code == 400

    @patch("app.api.routes.scripts_ai.get_mediakit_client")
    @patch("app.api.routes.scripts_ai._ytdlp_extract_video_url", return_value=(None, 0.0))
    @patch("tempfile.TemporaryDirectory")
    @patch("yt_dlp.YoutubeDL")
    def test_extract_from_douyin_download_failure(self, mock_ydl_cls, mock_tempdir, mock_meta, mock_get_mk):
        """下载失败返回 502."""
        from app.api.routes.scripts_ai import extract_from_douyin
        from app.schemas.scripts_ai import ExtractFromDouyinRequest
        from fastapi import HTTPException

        fake_mk = MagicMock()
        fake_mk.is_available = False
        mock_get_mk.return_value = fake_mk

        mock_ydl_cls.return_value = _mock_youtube_dl(
            extract_info_side_effect=Exception("Video unavailable"),
        )
        mock_ydl_cls.return_value.__enter__ = MagicMock(return_value=mock_ydl_cls.return_value)
        mock_ydl_cls.return_value.__exit__ = MagicMock(return_value=False)

        mock_td = MagicMock()
        mock_td.__enter__ = MagicMock(return_value="/tmp/douyin_extract_abc")
        mock_td.__exit__ = MagicMock(return_value=False)
        mock_tempdir.return_value = mock_td

        req = ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/")
        auth = _make_auth_user()

        with pytest.raises(HTTPException) as exc_info:
            extract_from_douyin(request=req, current_user=auth)
        assert exc_info.value.status_code == 502

    @patch("app.api.routes.scripts_ai.transcribe_to_text")
    @patch("tempfile.TemporaryDirectory")
    @patch("yt_dlp.YoutubeDL")
    @patch("app.api.routes.scripts_ai.os.path.getsize", return_value=1024)
    @patch("app.api.routes.scripts_ai.os.path.isfile", return_value=True)
    def test_extract_from_douyin_asr_not_configured(
        self,
        mock_isfile,
        mock_getsize,
        mock_ydl_cls,
        mock_tempdir,
        mock_transcribe,
    ):
        """ASR 未配置返回 503."""
        from app.api.routes.scripts_ai import extract_from_douyin
        from app.schemas.scripts_ai import ExtractFromDouyinRequest
        from app.services.script_asr_service import ASRNotConfiguredError
        from fastapi import HTTPException

        mock_ydl_cls.return_value = _mock_youtube_dl(
            extract_info_return={"id": "abc123", "duration": 60},
        )

        mock_td = MagicMock()
        mock_td.__enter__ = MagicMock(return_value="/tmp/douyin_extract_abc")
        mock_td.__exit__ = MagicMock(return_value=False)
        mock_tempdir.return_value = mock_td

        mock_transcribe.side_effect = ASRNotConfiguredError("ASR 服务未配置")

        req = ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/")
        auth = _make_auth_user()

        with pytest.raises(HTTPException) as exc_info:
            extract_from_douyin(request=req, current_user=auth)
        assert exc_info.value.status_code == 503

    @patch("app.api.routes.scripts_ai.transcribe_to_text")
    @patch("tempfile.TemporaryDirectory")
    @patch("yt_dlp.YoutubeDL")
    @patch("app.api.routes.scripts_ai.os.path.getsize", return_value=1024)
    @patch("app.api.routes.scripts_ai.os.path.isfile", return_value=True)
    def test_extract_from_douyin_asr_failure(
        self,
        mock_isfile,
        mock_getsize,
        mock_ydl_cls,
        mock_tempdir,
        mock_transcribe,
    ):
        """ASR 调用失败返回 502."""
        from app.api.routes.scripts_ai import extract_from_douyin
        from app.schemas.scripts_ai import ExtractFromDouyinRequest
        from app.services.script_asr_service import ASRTranscriptionError
        from fastapi import HTTPException

        mock_ydl_cls.return_value = _mock_youtube_dl(
            extract_info_return={"id": "abc123", "duration": 60},
        )

        mock_td = MagicMock()
        mock_td.__enter__ = MagicMock(return_value="/tmp/douyin_extract_abc")
        mock_td.__exit__ = MagicMock(return_value=False)
        mock_tempdir.return_value = mock_td

        mock_transcribe.side_effect = ASRTranscriptionError("语音识别失败: timeout")

        req = ExtractFromDouyinRequest(url="https://v.douyin.com/xxxxx/")
        auth = _make_auth_user()

        with pytest.raises(HTTPException) as exc_info:
            extract_from_douyin(request=req, current_user=auth)
        assert exc_info.value.status_code == 502


# ── ai_rewrite ───────────────────────────────────────────────────────────────


class TestAiRewrite:
    """POST /ai-rewrite 测试."""

    @patch("app.api.routes.scripts_ai.get_doubao_client")
    def test_ai_rewrite_success(self, mock_get_client):
        """正常改写成功."""
        from app.api.routes.scripts_ai import ai_rewrite
        from app.schemas.scripts_ai import AiRewriteRequest

        mock_client = MagicMock()
        mock_client.is_available = True
        mock_client.chat_completion.return_value = "改写后的文案内容，口语化风格"
        mock_get_client.return_value = mock_client

        req = AiRewriteRequest(content="原始文案内容", style="口语化")
        auth = _make_auth_user()
        result = ai_rewrite(request=req, current_user=auth)

        assert result.original == "原始文案内容"
        assert result.rewritten == "改写后的文案内容，口语化风格"
        assert result.style == "口语化"
        mock_client.chat_completion.assert_called_once()

    def test_ai_rewrite_empty_content(self):
        """空内容返回 400."""
        from app.api.routes.scripts_ai import ai_rewrite
        from app.schemas.scripts_ai import AiRewriteRequest
        from fastapi import HTTPException

        req = AiRewriteRequest(content="   ", style="口语化")
        auth = _make_auth_user()

        with pytest.raises(HTTPException) as exc_info:
            ai_rewrite(request=req, current_user=auth)
        assert exc_info.value.status_code == 400

    @patch("app.api.routes.scripts_ai.get_doubao_client")
    def test_ai_rewrite_llm_failure(self, mock_get_client):
        """LLM 调用失败返回 502."""
        from app.api.routes.scripts_ai import ai_rewrite
        from app.schemas.scripts_ai import AiRewriteRequest
        from fastapi import HTTPException

        mock_client = MagicMock()
        mock_client.is_available = True
        mock_client.chat_completion.side_effect = Exception("API timeout")
        mock_get_client.return_value = mock_client

        req = AiRewriteRequest(content="测试内容", style="口语化")
        auth = _make_auth_user()

        with pytest.raises(HTTPException) as exc_info:
            ai_rewrite(request=req, current_user=auth)
        assert exc_info.value.status_code == 502

    @patch("app.api.routes.scripts_ai.get_doubao_client")
    def test_ai_rewrite_client_unavailable(self, mock_get_client):
        """客户端不可用返回 502."""
        from app.api.routes.scripts_ai import ai_rewrite
        from app.schemas.scripts_ai import AiRewriteRequest
        from fastapi import HTTPException

        mock_client = MagicMock()
        mock_client.is_available = False
        mock_get_client.return_value = mock_client

        req = AiRewriteRequest(content="测试内容", style="口语化")
        auth = _make_auth_user()

        with pytest.raises(HTTPException) as exc_info:
            ai_rewrite(request=req, current_user=auth)
        assert exc_info.value.status_code == 502


# ── ai_generate_titles ───────────────────────────────────────────────────────


class TestAiGenerateTitles:
    """POST /ai-generate-titles 测试."""

    @patch("app.services.ai_service.get_doubao_client")
    def test_generate_titles_success(self, mock_get_client):
        """正常生成标题."""
        from app.api.routes.scripts_ai import ai_generate_titles
        from app.schemas.scripts_ai import AiGenerateTitlesRequest

        mock_client = MagicMock()
        mock_client.is_available = False  # 走 fallback 路径
        mock_get_client.return_value = mock_client

        req = AiGenerateTitlesRequest(content="这是一段关于美食的文案", count=3)
        auth = _make_auth_user()
        result = ai_generate_titles(request=req, current_user=auth)

        assert len(result.titles) == 3
        assert all(isinstance(t, str) for t in result.titles)

    @patch("app.services.ai_service.get_doubao_client")
    def test_generate_titles_count_clamp(self, mock_get_client):
        """count 超出范围时 Pydantic 校验拦截."""
        from app.schemas.scripts_ai import AiGenerateTitlesRequest

        mock_client = MagicMock()
        mock_client.is_available = False
        mock_get_client.return_value = mock_client

        auth = _make_auth_user()

        # count=10 被 Pydantic le=5 校验拦截 → ValidationError
        with pytest.raises(pydantic.ValidationError):
            AiGenerateTitlesRequest(content="测试", count=10)

        # count=0 被 Pydantic ge=1 校验拦截 → ValidationError
        with pytest.raises(pydantic.ValidationError):
            AiGenerateTitlesRequest(content="测试", count=0)

    @patch("app.services.ai_service.get_doubao_client")
    def test_generate_titles_count_valid_range(self, mock_get_client):
        """count=1 和 count=5 正常工作."""
        from app.api.routes.scripts_ai import ai_generate_titles
        from app.schemas.scripts_ai import AiGenerateTitlesRequest

        mock_client = MagicMock()
        mock_client.is_available = False
        mock_get_client.return_value = mock_client

        auth = _make_auth_user()

        # count=5
        req = AiGenerateTitlesRequest(content="测试内容", count=5)
        result = ai_generate_titles(request=req, current_user=auth)
        assert len(result.titles) <= 5

        # count=1
        req = AiGenerateTitlesRequest(content="测试内容", count=1)
        result = ai_generate_titles(request=req, current_user=auth)
        assert len(result.titles) >= 1

    def test_generate_titles_empty_content(self):
        """空内容返回 400."""
        from app.api.routes.scripts_ai import ai_generate_titles
        from app.schemas.scripts_ai import AiGenerateTitlesRequest
        from fastapi import HTTPException

        req = AiGenerateTitlesRequest(content="", count=3)
        auth = _make_auth_user()

        with pytest.raises(HTTPException) as exc_info:
            ai_generate_titles(request=req, current_user=auth)
        assert exc_info.value.status_code == 400

    @patch("app.services.ai_service.get_doubao_client")
    def test_generate_titles_default_count(self, mock_get_client):
        """不传 count 时默认 3."""
        from app.api.routes.scripts_ai import ai_generate_titles
        from app.schemas.scripts_ai import AiGenerateTitlesRequest

        mock_client = MagicMock()
        mock_client.is_available = False
        mock_get_client.return_value = mock_client

        req = AiGenerateTitlesRequest(content="测试文案内容")
        auth = _make_auth_user()
        result = ai_generate_titles(request=req, current_user=auth)

        assert len(result.titles) == 3


# ── URL 校验辅助函数 ─────────────────────────────────────────────────────────


class TestValidateDouyinUrl:
    """URL 校验逻辑单元测试."""

    @pytest.mark.parametrize(
        "valid_url",
        [
            "https://v.douyin.com/abc123/",
            "http://v.douyin.com/abc123/",
            "v.douyin.com/abc123/",
            "https://www.douyin.com/video/1234567890",
            "http://www.douyin.com/video/1234567890",
            "www.douyin.com/video/1234567890",
            "https://douyin.com/something",
        ],
    )
    def test_valid_urls(self, valid_url):
        """合法 URL 不抛异常."""
        from app.api.routes.scripts_ai import _extract_and_validate_douyin_url

        # bare domain (no http) 且没 path 的裸 douyin.com 现在会报错；过滤掉该用例
        _extract_and_validate_douyin_url(valid_url)

    @pytest.mark.parametrize(
        "invalid_url",
        [
            "",
            "   ",
            "https://www.youtube.com/watch?v=abc",
            "https://www.bilibili.com/video/BV123",
            "ftp://v.douyin.com/xxx/",
            "not-a-url",
        ],
    )
    def test_invalid_urls(self, invalid_url):
        """非法 URL 抛 400."""
        from app.api.routes.scripts_ai import _extract_and_validate_douyin_url
        from fastapi import HTTPException

        with pytest.raises(HTTPException) as exc_info:
            _extract_and_validate_douyin_url(invalid_url)
        assert exc_info.value.status_code == 400