diff --git a/tests/unit/test_asset_scoring_domain.py b/tests/unit/test_asset_scoring_domain.py new file mode 100755 index 000000000..5f3b7ced9 --- /dev/null +++ b/tests/unit/test_asset_scoring_domain.py @@ -0,0 +1,515 @@ +"""资产评分纯逻辑单元测试 — wave129.""" + +from __future__ import annotations + +from dataclasses import dataclass + +import pytest + +from packages.domain.asset_scoring import ( + AssetScoreDetail, + SmartSelectResult, + _bucket_by_duration, + calculate_total_score, + diverse_selection, + filter_candidates, + score_asset_detail, + score_bitrate, + score_duration, + score_resolution, + WEIGHT_QUALITY, + WEIGHT_RESOLUTION, + WEIGHT_DURATION, + WEIGHT_BITRATE, +) + +# ── 常量与权重 ────────────────────────────────────────────────────────────── + + +class TestWeights: + def test_weights_sum_to_one(self): + total = WEIGHT_QUALITY + WEIGHT_RESOLUTION + WEIGHT_DURATION + WEIGHT_BITRATE + assert abs(total - 1.0) < 0.001 + + def test_quality_is_highest_weight(self): + assert WEIGHT_QUALITY > WEIGHT_RESOLUTION + assert WEIGHT_QUALITY > WEIGHT_DURATION + assert WEIGHT_QUALITY > WEIGHT_BITRATE + + +# ── 分辨率评分 ────────────────────────────────────────────────────────────── + + +class TestScoreResolution: + def test_exact_target_full_score(self): + assert score_resolution(1920, 1080) == 1.0 + + def test_higher_than_target_full_score(self): + """4K 等高于目标分辨率也给满分.""" + assert score_resolution(3840, 2160) == 1.0 + assert score_resolution(2560, 1440) == 1.0 + + def test_seventytwop_less_than_one(self): + score = score_resolution(1280, 720) + assert 0.5 < score < 1.0 + + def test_fourheightyp_even_lower(self): + score_480 = score_resolution(854, 480) + score_720 = score_resolution(1280, 720) + assert score_480 < score_720 + + def test_none_returns_medium(self): + assert score_resolution(None, None) == 0.5 + assert score_resolution(None, 1080) == 0.5 + assert score_resolution(1920, None) == 0.5 + + def test_zero_or_negative_returns_medium(self): + assert score_resolution(0, 1080) == 0.5 + assert score_resolution(-100, 1080) == 0.5 + assert score_resolution(1920, 0) == 0.5 + + def test_custom_target_resolution(self): + score = score_resolution(1280, 720, target_width=1280, target_height=720) + assert score == 1.0 + + def test_score_between_zero_one(self): + score = score_resolution(320, 240) + assert 0.0 < score < 1.0 + + def test_very_low_resolution_not_zero(self): + """低分也不会低于 0.1.""" + score = score_resolution(160, 120) + assert score >= 0.1 + + +# ── 时长评分 ──────────────────────────────────────────────────────────────── + + +class TestScoreDuration: + def test_optimal_range_full_score(self): + """3-30秒最佳区间满分.""" + assert score_duration(3.0) == 1.0 + assert score_duration(10.0) == 1.0 + assert score_duration(30.0) == 1.0 + + def test_very_short_lower_score(self): + score_1s = score_duration(1.0) + assert 0.3 <= score_1s < 1.0 + + def test_shorter_than_optimal_lower(self): + """越短分越低.""" + score_1 = score_duration(1.0) + score_2 = score_duration(2.0) + assert score_1 < score_2 + + def test_just_below_optimal(self): + score = score_duration(2.9) + assert score < 1.0 + assert score > 0.8 # 接近满分 + + def test_too_long_penalty(self): + score_30 = score_duration(30.0) + score_60 = score_duration(60.0) + assert score_60 < score_30 + + def test_very_long_minimum_floor(self): + """超长素材最低 0.2 分.""" + score = score_duration(1000.0) + assert score >= 0.2 + + def test_none_returns_medium(self): + assert score_duration(None) == 0.5 + + def test_zero_returns_medium(self): + assert score_duration(0) == 0.5 + assert score_duration(0.0) == 0.5 + + def test_negative_returns_medium(self): + assert score_duration(-5.0) == 0.5 + + +# ── 码率评分 ──────────────────────────────────────────────────────────────── + + +class TestScoreBitrate: + def test_optimal_bitrate_full_score(self): + """2-8 Mbps 区间满分.""" + # 5 Mbps, 10秒 = 50Mbit = 6.25MB = 6,250,000 字节 + size_5mbps_10s = int(5_000_000 * 10 / 8) + assert score_bitrate(size_5mbps_10s, 10.0) == 1.0 + + # 3 Mbps, 5秒 + size_3mbps_5s = int(3_000_000 * 5 / 8) + assert score_bitrate(size_3mbps_5s, 5.0) == 1.0 + + def test_low_bitrate_lower_score(self): + """码率低得分低.""" + # 500 Kbps + size_low = int(500_000 * 10 / 8) + score = score_bitrate(size_low, 10.0) + assert 0.3 <= score < 1.0 + + def test_high_bitrate_moderate_penalty(self): + """码率过高适度扣分,最低0.5.""" + # 50 Mbps,远超 8Mbps + size_high = int(50_000_000 * 10 / 8) + score = score_bitrate(size_high, 10.0) + assert 0.5 <= score < 1.0 + + def test_none_duration_returns_medium(self): + assert score_bitrate(1_000_000, None) == 0.5 + + def test_zero_file_size_returns_medium(self): + assert score_bitrate(0, 10.0) == 0.5 + + def test_zero_duration_returns_medium(self): + assert score_bitrate(1_000_000, 0) == 0.5 + assert score_bitrate(1_000_000, -5.0) == 0.5 + + +# ── 加权总分 ──────────────────────────────────────────────────────────────── + + +class TestCalculateTotalScore: + def test_all_perfect(self): + assert calculate_total_score(1.0, 1.0, 1.0, 1.0) == 1.0 + + def test_all_zero(self): + assert calculate_total_score(0.0, 0.0, 0.0, 0.0) == 0.0 + + def test_weighted_calculation(self): + """手动验证加权计算.""" + q, r, d, b = 0.8, 0.6, 0.4, 0.2 + expected = WEIGHT_QUALITY * q + WEIGHT_RESOLUTION * r + WEIGHT_DURATION * d + WEIGHT_BITRATE * b + assert calculate_total_score(q, r, d, b) == pytest.approx(expected, rel=1e-3) + + def test_quality_dominates(self): + """质量分权重最高,质量分变化影响最大.""" + score_high_quality = calculate_total_score(1.0, 0.5, 0.5, 0.5) + score_low_quality = calculate_total_score(0.0, 0.5, 0.5, 0.5) + diff_quality = score_high_quality - score_low_quality + + score_high_res = calculate_total_score(0.5, 1.0, 0.5, 0.5) + score_low_res = calculate_total_score(0.5, 0.0, 0.5, 0.5) + diff_res = score_high_res - score_low_res + + assert diff_quality > diff_res + + def test_rounded_to_4_decimals(self): + result = calculate_total_score(0.3333, 0.3333, 0.3333, 0.3333) + assert result == round(result, 4) + + +# ── 单个素材评分详情 ──────────────────────────────────────────────────────── + + +class TestScoreAssetDetail: + def test_normal_asset(self): + detail = score_asset_detail( + asset_id="asset_001", + quality=80.0, + width=1920, + height=1080, + duration=10.0, + file_size=5_000_000, + ) + assert detail.asset_id == "asset_001" + assert 0.0 <= detail.total_score <= 1.0 + assert detail.quality_score == pytest.approx(0.8) + assert detail.resolution_score == 1.0 + assert detail.duration_score == 1.0 + assert detail.duration == 10.0 + + def test_unknown_quality_defaults(self): + detail = score_asset_detail( + asset_id="a1", + quality=None, + width=1920, + height=1080, + duration=10.0, + file_size=5_000_000, + ) + assert detail.quality_score == 0.5 + + def test_quality_hundred_is_one(self): + detail = score_asset_detail( + asset_id="a1", + quality=100.0, + width=1920, + height=1080, + duration=10.0, + file_size=1_000_000, + ) + assert detail.quality_score == 1.0 + + def test_quality_zero_is_zero(self): + detail = score_asset_detail( + asset_id="a1", + quality=0.0, + width=1920, + height=1080, + duration=10.0, + file_size=1_000_000, + ) + assert detail.quality_score == 0.0 + + def test_custom_target_resolution(self): + detail = score_asset_detail( + asset_id="a1", + quality=50.0, + width=1280, + height=720, + duration=10.0, + file_size=1_000_000, + target_width=1280, + target_height=720, + ) + assert detail.resolution_score == 1.0 + + +# ── 时长分桶 ──────────────────────────────────────────────────────────────── + + +def _make_detail(asset_id: str, duration: float | None, score: float = 0.8) -> AssetScoreDetail: + return AssetScoreDetail( + asset_id=asset_id, + total_score=score, + quality_score=score, + resolution_score=score, + duration_score=score, + bitrate_score=score, + duration=duration, + ) + + +class TestBucketByDuration: + def test_short_bucket(self): + item = _make_detail("s1", duration=3.0) + assert _bucket_by_duration(item) == "short" + + def test_short_boundary(self): + item = _make_detail("s1", duration=4.9) + assert _bucket_by_duration(item) == "short" + + def test_medium_bucket(self): + item = _make_detail("m1", duration=10.0) + assert _bucket_by_duration(item) == "medium" + + def test_medium_boundary(self): + item = _make_detail("m1", duration=14.9) + assert _bucket_by_duration(item) == "medium" + + def test_long_bucket(self): + item = _make_detail("l1", duration=20.0) + assert _bucket_by_duration(item) == "long" + + def test_long_at_boundary(self): + """>=15s 为长素材.""" + item = _make_detail("l1", duration=15.0) + assert _bucket_by_duration(item) == "long" + + def test_unknown_bucket(self): + item = _make_detail("u1", duration=None) + assert _bucket_by_duration(item) == "unknown" + + +# ── 多样性选择 ────────────────────────────────────────────────────────────── + + +class TestDiverseSelection: + def _make_scored_list(self) -> list[AssetScoreDetail]: + """构造一个包含各时长桶的测试列表,按分数降序.""" + items = [ + _make_detail("high_short", duration=3.0, score=0.95), + _make_detail("high_medium", duration=8.0, score=0.9), + _make_detail("high_long", duration=30.0, score=0.85), + _make_detail("mid_short", duration=2.0, score=0.8), + _make_detail("mid_medium", duration=10.0, score=0.75), + _make_detail("mid_long", duration=20.0, score=0.7), + _make_detail("low_short", duration=4.0, score=0.6), + _make_detail("low_medium", duration=12.0, score=0.5), + _make_detail("low_long", duration=60.0, score=0.4), + ] + items.sort(key=lambda x: x.total_score, reverse=True) + return items + + def test_empty_list_returns_empty(self): + assert diverse_selection([], 5) == [] + + def test_zero_count_returns_empty(self): + items = self._make_scored_list() + assert diverse_selection(items, 0) == [] + + def test_negative_count_returns_empty(self): + items = self._make_scored_list() + assert diverse_selection(items, -1) == [] + + def test_selects_from_multiple_buckets(self): + items = self._make_scored_list() + result = diverse_selection(items, 6) + assert len(result) == 6 + # 应该包含来自不同桶的素材 + durations = [r.duration for r in result] + has_short = any(d and d < 5.0 for d in durations) + has_medium = any(d and 5.0 <= d < 15.0 for d in durations) + has_long = any(d and d >= 15.0 for d in durations) + assert has_short and has_medium and has_long + + def test_no_duplicate_ids(self): + items = self._make_scored_list() + result = diverse_selection(items, 9) + ids = [r.asset_id for r in result] + assert len(ids) == len(set(ids)) + + def test_not_more_than_count(self): + items = self._make_scored_list() + result = diverse_selection(items, 3) + assert len(result) <= 3 + + def test_fewer_assets_than_count(self): + items = [_make_detail("a1", duration=3.0, score=0.9)] + result = diverse_selection(items, 10) + assert len(result) == 1 + + def test_only_short_bucket(self): + items = [ + _make_detail("s1", duration=1.0, score=0.9), + _make_detail("s2", duration=2.0, score=0.8), + _make_detail("s3", duration=3.0, score=0.7), + ] + result = diverse_selection(items, 3) + assert len(result) == 3 + # 只有短素材,应该都返回 + assert all(r.duration and r.duration < 5.0 for r in result) + + def test_highest_scores_priority(self): + """分数最高的素材应该优先被选中.""" + items = self._make_scored_list() + result = diverse_selection(items, 3) + # 最高分的那个应该在结果里 + assert result[0].asset_id == "high_short" + + def test_contains_top_scoring_items(self): + """结果中应该包含全局最高分的素材.""" + items = self._make_scored_list() + result = diverse_selection(items, 6) + result_ids = {r.asset_id for r in result} + # 全局最高分的应该在结果中 + assert "high_short" in result_ids + assert "high_medium" in result_ids + + +# ── 候选过滤 ──────────────────────────────────────────────────────────────── + + +@dataclass +class MockAsset: + asset_id: str + status: str = "ready" + mime_type: str = "video/mp4" + quality_score: float | None = None + + +class TestFilterCandidates: + def test_ready_video_passes(self): + assets = [MockAsset("a1", status="ready", mime_type="video/mp4")] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 1 + assert filtered == 0 + + def test_not_ready_filtered(self): + assets = [ + MockAsset("a1", status="processing"), + MockAsset("a2", status="failed"), + MockAsset("a3", status="ready"), + ] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 1 + assert candidates[0].asset_id == "a3" + assert filtered == 0 # 状态不对的不算质量过滤 + + def test_non_video_filtered(self): + assets = [ + MockAsset("a1", mime_type="image/jpeg"), + MockAsset("a2", mime_type="audio/mp3"), + MockAsset("a3", mime_type="video/mp4"), + ] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 1 + assert filtered == 0 + + def test_video_mime_prefix(self): + """所有以 video 开头的 MIME 都通过.""" + assets = [ + MockAsset("a1", mime_type="video/mp4"), + MockAsset("a2", mime_type="video/webm"), + MockAsset("a3", mime_type="video/x-matroska"), + ] + candidates, _ = filter_candidates(assets) + assert len(candidates) == 3 + + def test_low_quality_filtered_counted(self): + """质量分低于门槛的计入 filtered_out.""" + assets = [ + MockAsset("good", quality_score=80.0), + MockAsset("bad", quality_score=20.0), + ] + candidates, filtered = filter_candidates(assets, min_quality_score=30.0) + assert len(candidates) == 1 + assert candidates[0].asset_id == "good" + assert filtered == 1 + + def test_none_quality_passes(self): + """quality_score 为 None 的不做质量检查,通过.""" + assets = [MockAsset("a1", quality_score=None)] + candidates, filtered = filter_candidates(assets, min_quality_score=30.0) + assert len(candidates) == 1 + assert filtered == 0 + + def test_quality_at_threshold_passes(self): + """刚好等于门槛值的通过.""" + assets = [MockAsset("a1", quality_score=30.0)] + candidates, filtered = filter_candidates(assets, min_quality_score=30.0) + assert len(candidates) == 1 + assert filtered == 0 + + def test_empty_list(self): + candidates, filtered = filter_candidates([]) + assert candidates == [] + assert filtered == 0 + + def test_none_mime_type_treated_as_empty(self): + assets = [MockAsset("a1", mime_type=None)] # type: ignore + candidates, _ = filter_candidates(assets) + assert len(candidates) == 0 + + +# ── 数据类 ────────────────────────────────────────────────────────────────── + + +class TestDataClasses: + def test_score_detail_defaults(self): + detail = AssetScoreDetail( + asset_id="test", + total_score=0.5, + quality_score=0.5, + resolution_score=0.5, + duration_score=0.5, + bitrate_score=0.5, + duration=None, + ) + assert detail.asset_id == "test" + assert detail.total_score == 0.5 + assert detail.duration is None + + def test_smart_select_result_defaults(self): + result = SmartSelectResult( + selected_ids=["a1", "a2"], + total_candidates=10, + filtered_out=3, + avg_score=0.75, + ) + assert len(result.selected_ids) == 2 + assert result.total_candidates == 10 + assert result.filtered_out == 3 + assert result.details == [] # 默认空列表