ed09794f4d
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 2s
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 1s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 34s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 34s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m27s
CI/CD Pipeline / Unit Tests (pull_request) Failing after 1m25s
AI Code Review / AI Code Review (pull_request) Successful in 1m32s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m38s
CI/CD Pipeline / Validate - Style (pull_request) Successful in 1m59s
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Successful in 2m17s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 2m48s
CI/CD Pipeline / Validate - Security (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been cancelled
1. smart-match 排序零随机修复(主因):
- smart_select_assets 排序/多样性分桶注入 0~SCORE_RANDOM_NOISE_MAX 随机噪声,
同分/近分素材每次选出不同组合与顺序;分差>20的高质量素材保持稳定优先级
- 噪声以 asset.id 为 key 同次调用内一致;r.score 始终为无噪声原始分
- 支持 rng 注入(测试可复现);smart-match API/正式生成/模板编辑器三调用点全受益
2. 素材使用次数口径修复:
- mark_asset_used_for_generation 新增 times 参数,按成片实际渲染片段引用次数累加
- worker 回写从 task.asset_ids(请求列表,含未被plan选用的素材)改为
统计最终成片 plan 的 edit_plan_clips(同素材多片段复用按片段数累加)
- 抽 _count_plan_clip_asset_usage/_record_rendered_asset_usage 纯函数(可单测)
- plan 无有效片段时兜底 task.asset_ids 单次计数;单素材失败不阻断其他
3. 测试:22 新测试(噪声 10 + 回写计数 12);旧确定性排序断言注入零噪声 rng;
修复 test_distribute_assets 预存在 flaky(shuffle 未被零噪声 patch 覆盖)
178 lines
6.2 KiB
Python
178 lines
6.2 KiB
Python
"""
|
||
素材库自动匹配 单元测试
|
||
|
||
覆盖:
|
||
- all 模式:返回全部 ready 视频素材 ID
|
||
- random 模式:随机选取 N 个
|
||
- smart 模式:使用 smart_match 多维评分+多样性选取
|
||
- 无 ready 视频素材时返回空列表
|
||
- count=0 时返回全部(random/smart 模式)
|
||
- 非视频素材和非 ready 状态素材被过滤
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import random
|
||
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.smart_match import SCORE_RANDOM_NOISE_MAX
|
||
|
||
|
||
class _ZeroNoiseRandom(random.Random):
|
||
"""零噪声随机源:uniform(0, NOISE_MAX) 恒返回 0,smart 排序确定可复现。"""
|
||
|
||
def uniform(self, a, b):
|
||
if a == 0.0 and b == SCORE_RANDOM_NOISE_MAX:
|
||
return 0.0
|
||
return super().uniform(a, b)
|
||
|
||
|
||
_ZERO_NOISE = _ZeroNoiseRandom(0)
|
||
|
||
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 TestSelectAssetsSmartMode:
|
||
"""smart 模式:使用 smart_match 多维评分(质量40%+时长30%+新鲜度20%+未使用10%)。"""
|
||
|
||
def test_smart_sorts_by_quality_score_desc(self):
|
||
assets = [
|
||
_asset("low", "low.mp4", quality_score=30),
|
||
_asset("high", "high.mp4", quality_score=90),
|
||
_asset("mid", "mid.mp4", quality_score=60),
|
||
]
|
||
result = _select_assets_from_library(assets, mode="smart", count=0, rng=_ZERO_NOISE)
|
||
assert result == ["high", "mid", "low"]
|
||
|
||
def test_smart_duration_optimal_beats_too_short(self):
|
||
"""最优时长区间(5-30s)的素材得分高于过短素材。"""
|
||
assets = [
|
||
_asset("too_short", "short.mp4", quality_score=80, duration=1.0),
|
||
_asset("optimal", "optimal.mp4", quality_score=80, duration=15.0),
|
||
]
|
||
result = _select_assets_from_library(assets, mode="smart", count=0, rng=_ZERO_NOISE)
|
||
# Both: quality=80*0.4=32, recency/unused equal
|
||
# optimal(15s): duration_fitness=30 → total=62+
|
||
# too_short(1s): duration_fitness=20+(1/5)*80=36 → 36*0.3=10.8 → total=42.8+
|
||
assert result == ["optimal", "too_short"]
|
||
|
||
def test_smart_with_count_limits_results(self):
|
||
assets = [
|
||
_asset("a1", "v1.mp4", quality_score=90),
|
||
_asset("a2", "v2.mp4", quality_score=70),
|
||
_asset("a3", "v3.mp4", quality_score=50),
|
||
]
|
||
result = _select_assets_from_library(assets, mode="smart", count=2, rng=_ZERO_NOISE)
|
||
assert result == ["a1", "a2"]
|
||
|
||
def test_smart_null_quality_treated_as_default(self):
|
||
"""无质量分的素材按50分计算(0-100标度)。"""
|
||
assets = [
|
||
_asset("scored", "scored.mp4", quality_score=80),
|
||
_asset("unscored", "unscored.mp4", quality_score=None),
|
||
]
|
||
result = _select_assets_from_library(assets, mode="smart", count=0, rng=_ZERO_NOISE)
|
||
# scored(80): quality=80*0.4=32; unscored(None→50): quality=50*0.4=20
|
||
assert result == ["scored", "unscored"]
|
||
|
||
def test_smart_count_zero_returns_all_sorted(self):
|
||
assets = [
|
||
_asset("a1", "v1.mp4", quality_score=10),
|
||
_asset("a2", "v2.mp4", quality_score=90),
|
||
_asset("a3", "v3.mp4", quality_score=50),
|
||
]
|
||
result = _select_assets_from_library(assets, mode="smart", count=0, rng=_ZERO_NOISE)
|
||
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
|