saas-backend
75ec9db439
- ai_client/dashscope_client 新增 last_video_error 字段，记录每次失败的 error_code/user_message/status_code/detail/model 等结构化信息 - 新增 _classify_video_error 错误分类器，自动识别： - 真人肖像拦截 (portrait_intercept) → 提示用户移除真人图片 - 鉴权失败 (auth_error) / 配额不足 (quota_exceeded) / 限流 (rate_limit) - 模型/Endpoint 不存在 (model_not_found) / 参数错误 (invalid_param) - 网络超时 (network_error) / 任务失败 (task_failed) / 未知 (unknown) - call_video_generation 失败时通过 get_last_video_error() 暴露详细错误 - worker _step_render 不再笼统抛"返回为空"，改为读取错误详情并展示具体用户提示 - 增强日志：完整 payload 打印（截断 prompt）、轮询 HTTP 错误记录完整 body、 下载/超时时记录具体失败原因 - 修复重试策略：4xx 错误不再重试（参数/鉴权错误重试无意义），仅 5xx/网络错误重试 - DashScope 客户端同样增加 last_video_error 和错误分类，保持语义一致 - 新增单测覆盖：400真人拦截、401鉴权失败、轮询failed、ai_service错误透传、dashscope 401
659 lines
28 KiB
Python
659 lines
28 KiB
Python
"""#2106 DoubaoClient.video_generation 单测，覆盖 submit/poll/download 主路径和失败分支。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
from packages.shared.ai_client import DoubaoClient
|
||
|
||
|
||
def _make_client(**overrides):
|
||
client = DoubaoClient.__new__(DoubaoClient)
|
||
client.api_key = overrides.get("api_key", "test-key")
|
||
client.base_url = overrides.get("base_url", "https://ark.cn-beijing.volces.com/api/v3")
|
||
client.model = "doubao-model"
|
||
client.vision_model = "doubao-vision"
|
||
client.timeout = overrides.get("timeout", 30)
|
||
client.max_retries = overrides.get("max_retries", 0)
|
||
return client
|
||
|
||
|
||
def _fake_time_factory(base=1000.0, jump_after=2, jump=1e9):
|
||
"""返回一个 time.time() 替身：前 jump_after 次返回 base+offset，之后返回巨大值让 deadline 立即触发。
|
||
|
||
避免 Python logging 内部也调 time.time() 导致 StopIteration。
|
||
"""
|
||
state = {"n": 0}
|
||
|
||
def _t():
|
||
n = state["n"]
|
||
state["n"] += 1
|
||
if n < jump_after:
|
||
return base + n
|
||
return base + jump + n
|
||
|
||
return _t
|
||
|
||
|
||
class TestVideoGenerationHappyPath:
|
||
def test_happy_path_generates_and_downloads(self, tmp_path):
|
||
client = _make_client()
|
||
|
||
fake_task_resp = MagicMock()
|
||
fake_task_resp.json.return_value = {"id": "task-001"}
|
||
fake_task_resp.raise_for_status = MagicMock()
|
||
fake_task_resp.status_code = 200
|
||
fake_task_resp.text = ""
|
||
|
||
fake_poll_resp = MagicMock()
|
||
fake_poll_resp.json.return_value = {
|
||
"status": "succeeded",
|
||
"content": {"video_url": "https://cdn.example.com/v.mp4"},
|
||
}
|
||
fake_poll_resp.raise_for_status = MagicMock()
|
||
fake_poll_resp.status_code = 200
|
||
fake_poll_resp.text = ""
|
||
|
||
class FakeStreamResponse:
|
||
def __init__(self):
|
||
self._chunks = [b"FAKE", b"MP4", b"DATA"]
|
||
self._it = iter(self._chunks)
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *a):
|
||
return False
|
||
|
||
def raise_for_status(self):
|
||
return None
|
||
|
||
def iter_bytes(self, chunk_size=None):
|
||
return self._it
|
||
|
||
calls = {"post": 0, "get": 0}
|
||
|
||
def fake_post(url, **kwargs):
|
||
calls["post"] += 1
|
||
return fake_task_resp
|
||
|
||
def fake_get(url, **kwargs):
|
||
calls["get"] += 1
|
||
if "/tasks/task-001" in url:
|
||
return fake_poll_resp
|
||
raise AssertionError(f"unexpected GET (not stream): {url}")
|
||
|
||
fake_uuid = MagicMock()
|
||
fake_uuid.hex = "abcd1234"
|
||
|
||
with (
|
||
patch("packages.shared.ai_client.httpx.post", side_effect=fake_post),
|
||
patch("packages.shared.ai_client.httpx.get", side_effect=fake_get),
|
||
patch("packages.shared.ai_client.httpx.stream", return_value=FakeStreamResponse()),
|
||
patch("packages.shared.ai_client.time.sleep", return_value=None),
|
||
patch("packages.shared.ai_client.time.time", side_effect=_fake_time_factory(jump_after=2)),
|
||
patch("packages.shared.ai_client.uuid.uuid4", return_value=fake_uuid),
|
||
patch("packages.shared.ai_client.get_shared_settings") as mock_settings,
|
||
):
|
||
mock_settings.return_value = MagicMock(
|
||
doubao_video_poll_interval=0,
|
||
doubao_video_timeout=60,
|
||
doubao_video_model="doubao-seedance-2-5-260628",
|
||
)
|
||
out = client.video_generation(
|
||
prompt=" 镜头一 ",
|
||
image_url="https://img/x.jpg",
|
||
duration=5,
|
||
ratio="9:16",
|
||
resolution="720p",
|
||
output_dir=str(tmp_path),
|
||
)
|
||
assert out is not None and isinstance(out, dict)
|
||
assert Path(out["video_path"]).exists()
|
||
assert Path(out["video_path"]).name == "seedance_task-001_abcd1234.mp4"
|
||
assert Path(out["video_path"]).read_bytes() == b"FAKEMP4DATA"
|
||
assert calls["post"] == 1
|
||
assert calls["get"] == 1
|
||
|
||
|
||
class TestVideoGenerationFailures:
|
||
def test_returns_none_when_unavailable(self, tmp_path):
|
||
client = _make_client(api_key="")
|
||
assert client.video_generation("p", output_dir=str(tmp_path)) is None
|
||
|
||
def test_returns_none_on_empty_prompt(self, tmp_path):
|
||
client = _make_client()
|
||
assert client.video_generation(" ", output_dir=str(tmp_path)) is None
|
||
|
||
def test_returns_none_when_create_returns_no_id(self, tmp_path):
|
||
client = _make_client(max_retries=0)
|
||
fake_resp = MagicMock()
|
||
fake_resp.json.return_value = {"error": "bad"}
|
||
fake_resp.raise_for_status = MagicMock()
|
||
with (
|
||
patch("packages.shared.ai_client.httpx.post", return_value=fake_resp),
|
||
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
|
||
patch("packages.shared.ai_client.time.sleep", return_value=None),
|
||
patch("packages.shared.ai_client.time.time", side_effect=_fake_time_factory()),
|
||
):
|
||
mock_s.return_value = MagicMock(
|
||
doubao_video_poll_interval=1, doubao_video_timeout=60, doubao_video_model="seedance"
|
||
)
|
||
assert client.video_generation("p", output_dir=str(tmp_path)) is None
|
||
|
||
def test_returns_none_when_poll_returns_failed(self, tmp_path):
|
||
client = _make_client(max_retries=0)
|
||
create_resp = MagicMock()
|
||
create_resp.json.return_value = {"id": "t2"}
|
||
create_resp.raise_for_status = MagicMock()
|
||
poll_resp = MagicMock()
|
||
poll_resp.json.return_value = {"status": "failed", "error": {"code": "C1", "message": "bad"}}
|
||
poll_resp.raise_for_status = MagicMock()
|
||
with (
|
||
patch("packages.shared.ai_client.httpx.post", return_value=create_resp),
|
||
patch("packages.shared.ai_client.httpx.get", return_value=poll_resp),
|
||
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
|
||
patch("packages.shared.ai_client.time.sleep", return_value=None),
|
||
patch("packages.shared.ai_client.time.time", side_effect=_fake_time_factory()),
|
||
):
|
||
mock_s.return_value = MagicMock(
|
||
doubao_video_poll_interval=0, doubao_video_timeout=10, doubao_video_model="seedance"
|
||
)
|
||
assert client.video_generation("p", output_dir=str(tmp_path)) is None
|
||
|
||
def test_returns_none_when_download_raises(self, tmp_path):
|
||
client = _make_client(max_retries=0)
|
||
create_resp = MagicMock()
|
||
create_resp.json.return_value = {"id": "t3"}
|
||
create_resp.raise_for_status = MagicMock()
|
||
poll_resp = MagicMock()
|
||
poll_resp.json.return_value = {"status": "succeeded", "content": {"video_url": "https://cdn/v.mp4"}}
|
||
poll_resp.raise_for_status = MagicMock()
|
||
|
||
class BadStream:
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *a):
|
||
return False
|
||
|
||
def raise_for_status(self):
|
||
raise RuntimeError("network down")
|
||
|
||
def iter_bytes(self, **kw):
|
||
return iter([])
|
||
|
||
with (
|
||
patch("packages.shared.ai_client.httpx.post", return_value=create_resp),
|
||
patch("packages.shared.ai_client.httpx.get", return_value=poll_resp),
|
||
patch("packages.shared.ai_client.httpx.stream", return_value=BadStream()),
|
||
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
|
||
patch("packages.shared.ai_client.time.sleep", return_value=None),
|
||
patch("packages.shared.ai_client.time.time", side_effect=_fake_time_factory()),
|
||
):
|
||
mock_s.return_value = MagicMock(
|
||
doubao_video_poll_interval=0, doubao_video_timeout=10, doubao_video_model="seedance"
|
||
)
|
||
assert client.video_generation("p", output_dir=str(tmp_path)) is None
|
||
|
||
|
||
class TestVideoGenerationRetryAndPoll:
|
||
def test_create_retries_then_succeeds(self, tmp_path):
|
||
client = _make_client(max_retries=1)
|
||
|
||
ok_resp = MagicMock()
|
||
ok_resp.json.return_value = {"id": "t-retry"}
|
||
ok_resp.raise_for_status = MagicMock()
|
||
poll_resp = MagicMock()
|
||
poll_resp.json.return_value = {"status": "expired"}
|
||
poll_resp.raise_for_status = MagicMock()
|
||
|
||
calls = {"post": 0}
|
||
|
||
def fake_post(url, **kwargs):
|
||
calls["post"] += 1
|
||
if calls["post"] == 1:
|
||
raise httpx.HTTPError("network")
|
||
return ok_resp
|
||
|
||
with (
|
||
patch("packages.shared.ai_client.httpx") as mock_httpx,
|
||
patch("packages.shared.ai_client.time.sleep", return_value=None),
|
||
patch("packages.shared.ai_client.time.time", side_effect=_fake_time_factory(jump_after=3)),
|
||
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
|
||
):
|
||
mock_s.return_value = MagicMock(
|
||
doubao_video_poll_interval=0, doubao_video_timeout=1, doubao_video_model="seedance"
|
||
)
|
||
mock_httpx.HTTPError = httpx.HTTPError
|
||
mock_httpx.post.side_effect = fake_post
|
||
mock_httpx.get.return_value = poll_resp
|
||
assert client.video_generation("p", output_dir=str(tmp_path)) is None
|
||
assert calls["post"] == 2
|
||
|
||
def test_succeeded_but_no_video_url_returns_none(self, tmp_path):
|
||
client = _make_client(max_retries=0)
|
||
create_resp = MagicMock()
|
||
create_resp.json.return_value = {"id": "t-nourl"}
|
||
create_resp.raise_for_status = MagicMock()
|
||
poll_resp = MagicMock()
|
||
poll_resp.json.return_value = {"status": "succeeded", "content": {}}
|
||
poll_resp.raise_for_status = MagicMock()
|
||
with (
|
||
patch("packages.shared.ai_client.httpx.post", return_value=create_resp),
|
||
patch("packages.shared.ai_client.httpx.get", return_value=poll_resp),
|
||
patch("packages.shared.ai_client.time.sleep", return_value=None),
|
||
patch("packages.shared.ai_client.time.time", side_effect=_fake_time_factory()),
|
||
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
|
||
):
|
||
mock_s.return_value = MagicMock(
|
||
doubao_video_poll_interval=0, doubao_video_timeout=10, doubao_video_model="seedance"
|
||
)
|
||
assert client.video_generation("p", output_dir=str(tmp_path)) is None
|
||
|
||
|
||
class TestAiServiceCallVideoGeneration:
|
||
def test_returns_none_on_exception(self):
|
||
from packages.shared import ai_service
|
||
|
||
with patch("packages.shared.ai_service.get_doubao_client") as mock_get:
|
||
mock_client = MagicMock()
|
||
mock_client.is_available = True
|
||
mock_client.video_generation.side_effect = RuntimeError("boom")
|
||
mock_get.return_value = mock_client
|
||
assert ai_service.call_video_generation("p") is None
|
||
|
||
|
||
class TestVideoGenerationPollLoop:
|
||
def test_poll_queued_then_running_then_succeeded(self, tmp_path):
|
||
client = _make_client(max_retries=0)
|
||
create_resp = MagicMock()
|
||
create_resp.json.return_value = {"id": "t-wait"}
|
||
create_resp.raise_for_status = MagicMock()
|
||
|
||
queued = MagicMock(json=MagicMock(return_value={"status": "queued"}))
|
||
queued.raise_for_status = MagicMock()
|
||
running = MagicMock(json=MagicMock(return_value={"status": "running"}))
|
||
running.raise_for_status = MagicMock()
|
||
ok = MagicMock(
|
||
json=MagicMock(return_value={"status": "succeeded", "content": {"video_url": "https://cdn/x.mp4"}})
|
||
)
|
||
ok.raise_for_status = MagicMock()
|
||
poll_seq = [queued, running, ok]
|
||
|
||
class EmptyChunkStream:
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *a):
|
||
return False
|
||
|
||
def raise_for_status(self):
|
||
return None
|
||
|
||
def iter_bytes(self, chunk_size=None):
|
||
yield b""
|
||
yield b"D"
|
||
yield b""
|
||
yield b"ATA"
|
||
|
||
get_calls = {"n": 0}
|
||
|
||
def fake_get(url, **kw):
|
||
if "/tasks/t-wait" in url:
|
||
resp = poll_seq[min(get_calls["n"], len(poll_seq) - 1)]
|
||
get_calls["n"] += 1
|
||
return resp
|
||
raise AssertionError(url)
|
||
|
||
sleeps = []
|
||
# jump_after 要足够大：deadline 计算一次 + 3次 while 条件判断 = 4 次
|
||
with (
|
||
patch("packages.shared.ai_client.httpx.post", return_value=create_resp),
|
||
patch("packages.shared.ai_client.httpx.get", side_effect=fake_get),
|
||
patch("packages.shared.ai_client.httpx.stream", return_value=EmptyChunkStream()),
|
||
patch("packages.shared.ai_client.time.sleep", side_effect=lambda s: sleeps.append(s)),
|
||
patch("packages.shared.ai_client.time.time", side_effect=_fake_time_factory(jump_after=5, jump=1)),
|
||
patch("packages.shared.ai_client.uuid.uuid4", return_value=MagicMock(hex="ef012345")),
|
||
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
|
||
):
|
||
mock_s.return_value = MagicMock(
|
||
doubao_video_poll_interval=0, doubao_video_timeout=200, doubao_video_model="seedance"
|
||
)
|
||
out = client.video_generation("p", output_dir=str(tmp_path))
|
||
assert out is not None and isinstance(out, dict)
|
||
assert Path(out["video_path"]).read_bytes() == b"DATA"
|
||
# queued 和 running 各 sleep 一次
|
||
assert len(sleeps) >= 2
|
||
|
||
def test_poll_exception_does_not_crash(self, tmp_path):
|
||
client = _make_client(max_retries=0)
|
||
create_resp = MagicMock(json=MagicMock(return_value={"id": "t-err"}))
|
||
create_resp.raise_for_status = MagicMock()
|
||
ok = MagicMock(
|
||
json=MagicMock(return_value={"status": "succeeded", "content": {"video_url": "https://cdn/e.mp4"}})
|
||
)
|
||
ok.raise_for_status = MagicMock()
|
||
|
||
class OkStream:
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *a):
|
||
return False
|
||
|
||
def raise_for_status(self):
|
||
return None
|
||
|
||
def iter_bytes(self, chunk_size=None):
|
||
yield b"OK"
|
||
|
||
poll_calls = {"n": 0}
|
||
|
||
def fake_get(url, **kw):
|
||
poll_calls["n"] += 1
|
||
if poll_calls["n"] == 1:
|
||
raise httpx.HTTPError("transient")
|
||
return ok
|
||
|
||
with (
|
||
patch("packages.shared.ai_client.httpx.post", return_value=create_resp),
|
||
patch("packages.shared.ai_client.httpx.get", side_effect=fake_get),
|
||
patch("packages.shared.ai_client.httpx.stream", return_value=OkStream()),
|
||
patch("packages.shared.ai_client.time.sleep", return_value=None),
|
||
patch("packages.shared.ai_client.time.time", side_effect=_fake_time_factory(jump_after=3)),
|
||
patch("packages.shared.ai_client.uuid.uuid4", return_value=MagicMock(hex="11111111")),
|
||
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
|
||
):
|
||
mock_s.return_value = MagicMock(
|
||
doubao_video_poll_interval=0, doubao_video_timeout=100, doubao_video_model="seedance"
|
||
)
|
||
out = client.video_generation("p", output_dir=str(tmp_path))
|
||
assert out is not None and isinstance(out, dict)
|
||
assert Path(out["video_path"]).exists()
|
||
assert poll_calls["n"] == 2
|
||
|
||
def test_default_output_dir_and_audio_watermark(self, tmp_path, monkeypatch):
|
||
"""不传 output_dir 时落到 /tmp；generate_audio/watermark=True 也能正常提交。"""
|
||
client = _make_client()
|
||
|
||
create_resp = MagicMock(json=MagicMock(return_value={"id": "t-default"}))
|
||
create_resp.raise_for_status = MagicMock()
|
||
poll_resp = MagicMock(
|
||
json=MagicMock(return_value={"status": "succeeded", "content": {"video_url": "https://cdn/d.mp4"}})
|
||
)
|
||
poll_resp.raise_for_status = MagicMock()
|
||
|
||
# 用 tmp_path 伪造 /tmp 避免污染真 /tmp
|
||
monkeypatch.setattr("packages.shared.ai_client.os.makedirs", lambda d, exist_ok=True: None)
|
||
|
||
class S:
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *a):
|
||
return False
|
||
|
||
def raise_for_status(self):
|
||
return None
|
||
|
||
def iter_bytes(self, chunk_size=None):
|
||
yield b"D"
|
||
|
||
# 捕获 POST payload 断言
|
||
captured = {}
|
||
|
||
def fake_post(url, **kw):
|
||
captured["json"] = kw.get("json")
|
||
return create_resp
|
||
|
||
def fake_get(url, **kw):
|
||
return poll_resp
|
||
|
||
def fake_open(path, mode):
|
||
# 返回一个 MagicMock file，模拟写入
|
||
f = MagicMock()
|
||
f.__enter__ = MagicMock(return_value=f)
|
||
f.__exit__ = MagicMock(return_value=False)
|
||
captured["path"] = path
|
||
return f
|
||
|
||
monkeypatch.setattr("packages.shared.ai_client.httpx.post", fake_post)
|
||
monkeypatch.setattr("packages.shared.ai_client.httpx.get", fake_get)
|
||
monkeypatch.setattr("packages.shared.ai_client.httpx.stream", lambda *a, **kw: S())
|
||
monkeypatch.setattr("builtins.open", fake_open)
|
||
monkeypatch.setattr("packages.shared.ai_client.os.path.getsize", lambda p: 99)
|
||
|
||
with (
|
||
patch("packages.shared.ai_client.time.sleep", return_value=None),
|
||
patch("packages.shared.ai_client.time.time", side_effect=_fake_time_factory()),
|
||
patch("packages.shared.ai_client.uuid.uuid4", return_value=MagicMock(hex="00000001")),
|
||
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
|
||
):
|
||
mock_s.return_value = MagicMock(
|
||
doubao_video_poll_interval=0,
|
||
doubao_video_timeout=60,
|
||
doubao_video_model="seedance",
|
||
)
|
||
out = client.video_generation(
|
||
"p", duration=3, ratio="1:1", resolution="480p", generate_audio=True, watermark=True
|
||
)
|
||
assert out is not None and isinstance(out, dict)
|
||
assert out["video_path"] == "/tmp/seedance_t-default_00000001.mp4"
|
||
assert captured["json"]["generate_audio"] is True
|
||
assert captured["json"]["watermark"] is True
|
||
assert captured["json"]["ratio"] == "1:1"
|
||
assert captured["json"]["resolution"] == "480p"
|
||
|
||
|
||
class TestGetDoubaoClientSingleton:
|
||
def test_singleton_lazy_init(self):
|
||
from packages.shared import ai_client
|
||
|
||
prev = ai_client._client
|
||
try:
|
||
ai_client._client = None
|
||
c1 = ai_client.get_doubao_client()
|
||
c2 = ai_client.get_doubao_client()
|
||
assert c1 is c2
|
||
assert isinstance(c1, ai_client.DoubaoClient)
|
||
finally:
|
||
ai_client._client = prev
|
||
|
||
|
||
class TestVideoGenerationCancelled:
|
||
def test_poll_cancelled_returns_none(self, tmp_path):
|
||
client = _make_client(max_retries=0)
|
||
create_resp = MagicMock(json=MagicMock(return_value={"id": "t-can"}))
|
||
create_resp.raise_for_status = MagicMock()
|
||
poll_resp = MagicMock(json=MagicMock(return_value={"status": "cancelled"}))
|
||
poll_resp.raise_for_status = MagicMock()
|
||
with (
|
||
patch("packages.shared.ai_client.httpx.post", return_value=create_resp),
|
||
patch("packages.shared.ai_client.httpx.get", return_value=poll_resp),
|
||
patch("packages.shared.ai_client.time.sleep", return_value=None),
|
||
patch("packages.shared.ai_client.time.time", side_effect=_fake_time_factory()),
|
||
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
|
||
):
|
||
mock_s.return_value = MagicMock(
|
||
doubao_video_poll_interval=0, doubao_video_timeout=10, doubao_video_model="seedance"
|
||
)
|
||
assert client.video_generation("p", output_dir=str(tmp_path)) is None
|
||
|
||
|
||
# ============ #2157 _resolve_video_model_id 模型ID映射单测 ============
|
||
|
||
|
||
class TestResolveVideoModelId:
|
||
"""覆盖 _resolve_video_model_id 各分支（#2157 P0 修复）。"""
|
||
|
||
def _import_target(self):
|
||
from packages.shared.ai_client import _resolve_video_model_id
|
||
|
||
return _resolve_video_model_id
|
||
|
||
def test_none_uses_default(self):
|
||
fn = self._import_target()
|
||
with patch("packages.shared.ai_client.get_shared_settings") as ms:
|
||
ms.return_value = MagicMock(doubao_video_model="doubao-seedance-2-5-260628")
|
||
assert fn(None) == "doubao-seedance-2-5-260628"
|
||
|
||
def test_empty_uses_default(self):
|
||
fn = self._import_target()
|
||
with patch("packages.shared.ai_client.get_shared_settings") as ms:
|
||
ms.return_value = MagicMock(doubao_video_model="doubao-seedance-2-5-260628")
|
||
assert fn(" ") == "doubao-seedance-2-5-260628"
|
||
|
||
def test_doubao_prefix_passthrough(self):
|
||
fn = self._import_target()
|
||
assert fn("doubao-seedance-2-5-260628") == "doubao-seedance-2-5-260628"
|
||
|
||
def test_ep_prefix_passthrough(self):
|
||
fn = self._import_target()
|
||
assert fn("ep-20260721114705-b568m") == "ep-20260721114705-b568m"
|
||
|
||
def test_seedance_2_5_alias(self):
|
||
fn = self._import_target()
|
||
assert fn("seedance-2.5") == "doubao-seedance-2-5-260628"
|
||
|
||
def test_seedance_2_0_alias(self):
|
||
fn = self._import_target()
|
||
assert fn("seedance-2.0") == "doubao-seedance-2-0-260128"
|
||
|
||
def test_seedance_2_0_fast_alias(self):
|
||
fn = self._import_target()
|
||
assert fn("seedance-2.0-fast") == "doubao-seedance-2-0-fast-260128"
|
||
|
||
def test_seedance_2_0_mini_alias(self):
|
||
fn = self._import_target()
|
||
assert fn("seedance-2.0-mini") == "doubao-seedance-2-0-mini-260615"
|
||
|
||
def test_wan_3_0_returns_dashscope_provider(self):
|
||
from packages.shared.ai_client import _resolve_video_provider_and_id
|
||
|
||
prov, mid, cfg = _resolve_video_provider_and_id("wan-3.0")
|
||
assert prov == "dashscope"
|
||
assert mid == "wan3.0-video"
|
||
assert cfg.get("billing_mode") == "per_second"
|
||
|
||
def test_seedance_2_5_uppercase(self):
|
||
fn = self._import_target()
|
||
assert fn("Seedance-2.5") == "doubao-seedance-2-5-260628"
|
||
|
||
def test_seedance_dot_normalize(self):
|
||
fn = self._import_target()
|
||
# dot 形式 "seedance-2.5" 直接命中 domain config 的 key（与 2-5 同等）
|
||
assert fn("seedance-2.5") == "doubao-seedance-2-5-260628"
|
||
|
||
def test_unknown_model_falls_back_to_default_seedance_2_5(self, caplog):
|
||
fn = self._import_target()
|
||
import logging
|
||
|
||
# 未知 model key 会通过 get_viral_video_model_config 回落到 seedance-2.5
|
||
with caplog.at_level(logging.WARNING, logger="shared.ai_client"):
|
||
assert fn("some-random-model") == "doubao-seedance-2-5-260628"
|
||
|
||
|
||
# ── #2165 详细错误信息和 last_video_error ─────────────────────────
|
||
|
||
|
||
class TestVideoGenerationLastError:
|
||
def test_create_400_portrait_returns_user_message(self, tmp_path):
|
||
"""HTTP 400 + 真人拦截关键词 → portrait_intercept 错误码，用户提示友好。"""
|
||
client = _make_client(max_retries=0)
|
||
create_resp = MagicMock()
|
||
create_resp.status_code = 400
|
||
create_resp.text = '{"error":{"code":"ContentRisk","message":"Real person face detected in reference image, portrait blocked"}}'
|
||
create_resp.json.return_value = {"error": {"code": "ContentRisk", "message": "..."}}
|
||
create_resp.raise_for_status.side_effect = httpx.HTTPStatusError(
|
||
"bad", request=MagicMock(), response=create_resp
|
||
)
|
||
with (
|
||
patch("packages.shared.ai_client.httpx.post", return_value=create_resp),
|
||
patch("packages.shared.ai_client.time.sleep", return_value=None),
|
||
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
|
||
):
|
||
mock_s.return_value = MagicMock(
|
||
doubao_video_poll_interval=0, doubao_video_timeout=1, doubao_video_model="seedance"
|
||
)
|
||
result = client.video_generation("p", output_dir=str(tmp_path), image_url="https://img/x.jpg")
|
||
assert result is None
|
||
err = client.get_last_video_error()
|
||
assert err["error_code"] == "portrait_intercept"
|
||
assert "真人" in err["user_message"]
|
||
assert err["status_code"] == 400
|
||
|
||
def test_create_401_returns_auth_error(self, tmp_path):
|
||
client = _make_client(max_retries=0)
|
||
create_resp = MagicMock()
|
||
create_resp.status_code = 401
|
||
create_resp.text = '{"error":{"message":"Unauthorized"}}'
|
||
create_resp.json.return_value = {"error": {"message": "Unauthorized"}}
|
||
create_resp.raise_for_status.side_effect = httpx.HTTPStatusError(
|
||
"auth", request=MagicMock(), response=create_resp
|
||
)
|
||
with (
|
||
patch("packages.shared.ai_client.httpx.post", return_value=create_resp),
|
||
patch("packages.shared.ai_client.time.sleep", return_value=None),
|
||
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
|
||
):
|
||
mock_s.return_value = MagicMock(
|
||
doubao_video_poll_interval=0, doubao_video_timeout=1, doubao_video_model="seedance"
|
||
)
|
||
result = client.video_generation("p", output_dir=str(tmp_path))
|
||
assert result is None
|
||
err = client.get_last_video_error()
|
||
assert err["error_code"] == "auth_error"
|
||
assert err["status_code"] == 401
|
||
|
||
def test_poll_failed_returns_task_failed_error(self, tmp_path):
|
||
"""轮询 status=failed 时应记录 task_failed 错误并含 detail。"""
|
||
client = _make_client(max_retries=0)
|
||
create_resp = MagicMock()
|
||
create_resp.status_code = 200
|
||
create_resp.json.return_value = {"id": "t-fail"}
|
||
create_resp.raise_for_status = MagicMock()
|
||
poll_resp = MagicMock()
|
||
poll_resp.status_code = 200
|
||
poll_resp.json.return_value = {
|
||
"status": "failed",
|
||
"error": {"code": "InvalidParam", "message": "resolution invalid"},
|
||
}
|
||
poll_resp.raise_for_status = MagicMock()
|
||
with (
|
||
patch("packages.shared.ai_client.httpx.post", return_value=create_resp),
|
||
patch("packages.shared.ai_client.httpx.get", return_value=poll_resp),
|
||
patch("packages.shared.ai_client.time.sleep", return_value=None),
|
||
patch("packages.shared.ai_client.time.time", side_effect=_fake_time_factory()),
|
||
patch("packages.shared.ai_client.get_shared_settings") as mock_s,
|
||
):
|
||
mock_s.return_value = MagicMock(
|
||
doubao_video_poll_interval=0, doubao_video_timeout=10, doubao_video_model="seedance"
|
||
)
|
||
result = client.video_generation("p", output_dir=str(tmp_path))
|
||
assert result is None
|
||
err = client.get_last_video_error()
|
||
assert err["error_code"] == "task_failed"
|
||
assert "InvalidParam" in err.get("detail", "") or err["status_code"] == 200
|
||
|
||
|
||
class TestAiServiceLastVideoError:
|
||
def test_call_video_generation_returns_none_sets_error(self):
|
||
"""失败后 get_last_video_error 应返回结构化错误信息。"""
|
||
from packages.shared import ai_service
|
||
|
||
mock_client = MagicMock()
|
||
mock_client.is_available = True
|
||
mock_client.last_video_error = {"error_code": "unknown", "user_message": "test"}
|
||
mock_client.get_last_video_error.return_value = {"error_code": "unknown", "user_message": "test"}
|
||
mock_client.video_generation.return_value = None
|
||
with patch("packages.shared.ai_service.get_doubao_client", return_value=mock_client):
|
||
assert ai_service.call_video_generation("p") is None
|
||
err = ai_service.get_last_video_error()
|
||
assert err["error_code"] == "unknown"
|
||
assert "user_message" in err
|