Files
xiaoxia-saas/tests/unit/test_asset_select_mode.py
灵应 1ea8fd3989
CI/CD Pipeline / Deploy Staging (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Failing after 45h41m20s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 45h41m20s
feat: P2 素材库自动匹配 - 支持 all/random/smart 三种素材选取模式
- Schema: 新增 asset_select_mode + asset_select_count 字段
- Domain: GenerationTask 新增 asset_select_mode 字段
- Application: Command/UseCase 透传 asset_select_mode
- Route: _select_assets_from_library() 辅助函数 + 创建任务集成
- SQLAlchemy: Model/Repository 映射 asset_select_mode
- Alembic: 032 号迁移
- Worker: _download_library_assets() 支持 asset_ids 过滤
- 单测: 15 个测试覆盖三种模式 + 边界情况
2026-07-07 15:13:05 +08:00

177 lines
6.0 KiB
Python

"""
素材库自动匹配 单元测试
覆盖:
- all 模式:返回全部 ready 视频素材 ID
- random 模式:随机选取 N 个
- smart 模式:按质量分/时长评分降序选取
- 无 ready 视频素材时返回空列表
- count=0 时返回全部(random/smart 模式)
- 非视频素材和非 ready 状态素材被过滤
"""
from __future__ import annotations
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
from app.api.routes.generation_tasks import _select_assets_from_library
from packages.domain import Asset, AssetStatus
def _asset(
id: str,
name: str,
mime_type: str = "video/mp4",
status: AssetStatus = AssetStatus.READY,
quality_score: float | None = None,
duration: float | None = None,
) -> Asset:
a = Asset.create(
project_id="proj-1",
library_id="lib-1",
name=name,
storage_key=f"uploads/{name}",
mime_type=mime_type,
file_size=1024,
status=status,
quality_score=quality_score,
duration=duration,
)
# create() 会覆盖 id,手动设置
a.id = id
return a
class TestSelectAssetsAllMode:
"""all 模式:返回全部 ready 视频素材。"""
def test_returns_all_ready_video_assets(self):
assets = [
_asset("a1", "v1.mp4"),
_asset("a2", "v2.mp4"),
_asset("a3", "v3.mp4"),
]
result = _select_assets_from_library(assets, mode="all", count=0)
assert sorted(result) == ["a1", "a2", "a3"]
def test_ignores_count_in_all_mode(self):
assets = [
_asset("a1", "v1.mp4"),
_asset("a2", "v2.mp4"),
]
result = _select_assets_from_library(assets, mode="all", count=1)
assert len(result) == 2
def test_filters_non_video_assets(self):
assets = [
_asset("a1", "v1.mp4", mime_type="video/mp4"),
_asset("a2", "img.jpg", mime_type="image/jpeg"),
_asset("a3", "v2.mov", mime_type="video/quicktime"),
]
result = _select_assets_from_library(assets, mode="all", count=0)
assert sorted(result) == ["a1", "a3"]
def test_filters_non_ready_assets(self):
assets = [
_asset("a1", "v1.mp4", status=AssetStatus.READY),
_asset("a2", "v2.mp4", status=AssetStatus.UPLOADING),
_asset("a3", "v3.mp4", status=AssetStatus.PROCESSING),
]
result = _select_assets_from_library(assets, mode="all", count=0)
assert result == ["a1"]
def test_empty_library_returns_empty(self):
result = _select_assets_from_library([], mode="all", count=0)
assert result == []
def test_no_ready_video_returns_empty(self):
assets = [
_asset("a1", "v1.mp4", status=AssetStatus.UPLOADING),
_asset("a2", "img.jpg", mime_type="image/jpeg", status=AssetStatus.READY),
]
result = _select_assets_from_library(assets, mode="all", count=0)
assert result == []
class TestSelectAssetsRandomMode:
"""random 模式:随机选取 N 个。"""
def test_random_selects_exact_count(self):
assets = [_asset(f"a{i}", f"v{i}.mp4") for i in range(10)]
result = _select_assets_from_library(assets, mode="random", count=3)
assert len(result) == 3
assert all(rid in [a.id for a in assets] for rid in result)
def test_random_count_zero_returns_all(self):
assets = [_asset(f"a{i}", f"v{i}.mp4") for i in range(5)]
result = _select_assets_from_library(assets, mode="random", count=0)
assert len(result) == 5
def test_random_count_exceeds_total_returns_all(self):
assets = [_asset(f"a{i}", f"v{i}.mp4") for i in range(3)]
result = _select_assets_from_library(assets, mode="random", count=100)
assert len(result) == 3
class TestSelectAssetsSmartMode:
"""smart 模式:按质量分/时长评分降序选取。"""
def test_smart_sorts_by_quality_score_desc(self):
assets = [
_asset("low", "low.mp4", quality_score=0.3),
_asset("high", "high.mp4", quality_score=0.9),
_asset("mid", "mid.mp4", quality_score=0.6),
]
result = _select_assets_from_library(assets, mode="smart", count=0)
assert result == ["high", "mid", "low"]
def test_smart_tiebreak_by_duration_desc(self):
assets = [
_asset("short", "short.mp4", quality_score=0.8, duration=10.0),
_asset("long", "long.mp4", quality_score=0.8, duration=60.0),
]
result = _select_assets_from_library(assets, mode="smart", count=0)
assert result == ["long", "short"]
def test_smart_with_count_limits_results(self):
assets = [
_asset("a1", "v1.mp4", quality_score=0.9),
_asset("a2", "v2.mp4", quality_score=0.7),
_asset("a3", "v3.mp4", quality_score=0.5),
]
result = _select_assets_from_library(assets, mode="smart", count=2)
assert result == ["a1", "a2"]
def test_smart_null_quality_treated_as_zero(self):
assets = [
_asset("scored", "scored.mp4", quality_score=0.5),
_asset("unscored", "unscored.mp4", quality_score=None),
]
result = _select_assets_from_library(assets, mode="smart", count=0)
assert result == ["scored", "unscored"]
def test_smart_count_zero_returns_all_sorted(self):
assets = [
_asset("a1", "v1.mp4", quality_score=0.1),
_asset("a2", "v2.mp4", quality_score=0.9),
_asset("a3", "v3.mp4", quality_score=0.5),
]
result = _select_assets_from_library(assets, mode="smart", count=0)
assert result == ["a2", "a3", "a1"]
class TestSelectAssetsDefaultMode:
"""默认模式(未知 mode 字符串)应回退到 all。"""
def test_unknown_mode_falls_back_to_all(self):
assets = [
_asset("a1", "v1.mp4"),
_asset("a2", "v2.mp4"),
]
result = _select_assets_from_library(assets, mode="unknown", count=0)
assert len(result) == 2