diff --git a/tests/unit/domain/test_asset_scoring.py b/tests/unit/domain/test_asset_scoring.py new file mode 100755 index 000000000..983a3f21d --- /dev/null +++ b/tests/unit/domain/test_asset_scoring.py @@ -0,0 +1,530 @@ +"""asset_scoring 单元测试 - wave162 + +覆盖: +- 分辨率评分 score_resolution +- 时长评分 score_duration +- 码率评分 score_bitrate +- 加权总分 calculate_total_score +- 单个素材评分 score_asset_detail +- 时长分桶 _bucket_by_duration +- 多样性选择 diverse_selection +- 候选过滤 filter_candidates +""" + +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, +) + +# ============================================================ +# score_resolution +# ============================================================ + + +class TestScoreResolution: + def test_none_width_returns_mid(self): + assert score_resolution(None, 1080) == 0.5 + + def test_none_height_returns_mid(self): + assert score_resolution(1920, None) == 0.5 + + def test_zero_dimension_returns_mid(self): + assert score_resolution(0, 1080) == 0.5 + assert score_resolution(1920, 0) == 0.5 + assert score_resolution(-1, 1080) == 0.5 + + def test_exact_target_returns_1(self): + assert score_resolution(1920, 1080) == 1.0 + + def test_higher_than_target_returns_1(self): + assert score_resolution(3840, 2160) == 1.0 # 4K + assert score_resolution(2560, 1440) == 1.0 # 2K + + def test_lower_than_target_linear_decay(self): + # 720p = 1280*720 / 1920*1080 = 0.444 ratio + # score = 0.3 + 0.7 * 0.444 = 0.611 + score = score_resolution(1280, 720) + assert 0.55 < score < 0.7 + + def test_very_low_has_floor(self): + # 最低不低于 0.1 + score = score_resolution(100, 100) + assert score >= 0.1 + + def test_480p_still_reasonable(self): + score = score_resolution(640, 480) + assert 0.3 < score < 0.5 + + def test_custom_target(self): + score = score_resolution(1280, 720, target_width=1280, target_height=720) + assert score == 1.0 + + def test_between_0_and_1(self): + for w, h in [(1920, 1080), (1280, 720), (640, 480), (3840, 2160)]: + s = score_resolution(w, h) + assert 0.0 <= s <= 1.0 + + +# ============================================================ +# score_duration +# ============================================================ + + +class TestScoreDuration: + def test_none_returns_mid(self): + assert score_duration(None) == 0.5 + + def test_zero_or_negative_returns_mid(self): + assert score_duration(0) == 0.5 + assert score_duration(-1) == 0.5 + + def test_optimal_range_returns_1(self): + assert score_duration(3.0) == 1.0 + assert score_duration(10.0) == 1.0 + assert score_duration(30.0) == 1.0 + assert score_duration(15.0) == 1.0 + + def test_short_duration_linear_decay(self): + # 1.5s: ratio = 1.5/3 = 0.5, score = 0.3 + 0.7*0.5 = 0.65 + score = score_duration(1.5) + assert score == pytest.approx(0.65) + + def test_very_short_above_floor(self): + score = score_duration(0.1) + assert 0.3 <= score < 0.5 + + def test_long_duration_penalty(self): + # 40s: excess=10, penalty=10/10*0.1=0.1, score=0.9 + score = score_duration(40.0) + assert score == pytest.approx(0.9) + + def test_very_long_minimum_floor(self): + # 超过很多,最低 0.2 + score = score_duration(1000.0) + assert score >= 0.2 + assert score < 0.5 + + def test_just_below_optimal(self): + score = score_duration(2.9) + assert 0.9 < score < 1.0 + + def test_just_above_optimal(self): + score = score_duration(30.1) + assert 0.9 < score < 1.0 + + +# ============================================================ +# score_bitrate +# ============================================================ + + +class TestScoreBitrate: + def test_no_file_size_returns_mid(self): + assert score_bitrate(0, 10.0) == 0.5 + + def test_no_duration_returns_mid(self): + assert score_bitrate(1000000, None) == 0.5 + assert score_bitrate(1000000, 0) == 0.5 + assert score_bitrate(1000000, -1) == 0.5 + + def test_optimal_range_returns_1(self): + # 5 Mbps for 10s = 5*10^6 * 10 / 8 = 6,250,000 bytes + size_5mbps_10s = int(5_000_000 * 10 / 8) + assert score_bitrate(size_5mbps_10s, 10.0) == 1.0 + + def test_low_bitrate_decay(self): + # 500 Kbps for 10s + size_500kbps = int(500_000 * 10 / 8) + score = score_bitrate(size_500kbps, 10.0) + assert 0.3 < score < 0.7 + + def test_high_bitrate_moderate_penalty(self): + # 16 Mbps (2x optimal high), excess=1.0, penalty=min(0.5, 1.0*0.2)=0.2 + # score = 0.8 + size_16mbps = int(16_000_000 * 10 / 8) + score = score_bitrate(size_16mbps, 10.0) + assert 0.7 < score < 0.9 + + def test_very_high_bitrate_floor(self): + # 极高码率,最低 0.5 + huge_size = 10**9 # 1GB for 1s = 8Gbps + score = score_bitrate(huge_size, 1.0) + assert score >= 0.5 + + def test_between_0_and_1(self): + for size, dur in [(1000, 1), (1000000, 10), (100000000, 5)]: + s = score_bitrate(size, dur) + assert 0.0 <= s <= 1.0 + + +# ============================================================ +# calculate_total_score +# ============================================================ + + +class TestCalculateTotalScore: + def test_all_perfect_equals_1(self): + assert calculate_total_score(1.0, 1.0, 1.0, 1.0) == 1.0 + + def test_all_zero_equals_0(self): + assert calculate_total_score(0.0, 0.0, 0.0, 0.0) == 0.0 + + def test_weighted_sum(self): + # 0.5*0.5 + 0.2*0.5 + 0.2*0.5 + 0.1*0.5 = 0.25+0.1+0.1+0.05 = 0.5 + assert calculate_total_score(0.5, 0.5, 0.5, 0.5) == pytest.approx(0.5) + + def test_quality_has_highest_weight(self): + # 只提高质量分,对比只提高其他 + q_high = calculate_total_score(1.0, 0.0, 0.0, 0.0) + r_high = calculate_total_score(0.0, 1.0, 0.0, 0.0) + assert q_high > r_high # 0.5 > 0.2 + + def test_bitrate_has_lowest_weight(self): + b_high = calculate_total_score(0.0, 0.0, 0.0, 1.0) + q_high = calculate_total_score(1.0, 0.0, 0.0, 0.0) + assert b_high < q_high # 0.1 < 0.5 + + def test_rounded_to_4_decimals(self): + result = calculate_total_score(0.3333, 0.3333, 0.3333, 0.3333) + assert round(result, 4) == result + + +# ============================================================ +# score_asset_detail +# ============================================================ + + +class TestScoreAssetDetail: + def test_returns_detail_object(self): + detail = score_asset_detail( + asset_id="a1", + quality=80.0, + width=1920, + height=1080, + duration=10.0, + file_size=5_000_000, + ) + assert isinstance(detail, AssetScoreDetail) + assert detail.asset_id == "a1" + assert 0.0 <= detail.total_score <= 1.0 + + def test_perfect_asset_high_score(self): + detail = score_asset_detail( + asset_id="perfect", + quality=100.0, + width=1920, + height=1080, + duration=10.0, + file_size=6_250_000, # 5Mbps for 10s + ) + assert detail.total_score > 0.9 + + def test_quality_none_defaults_mid(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_normalized(self): + detail = score_asset_detail( + asset_id="a1", + quality=50.0, + width=1920, + height=1080, + duration=10.0, + file_size=5_000_000, + ) + assert detail.quality_score == pytest.approx(0.5) + + def test_custom_target_resolution(self): + detail = score_asset_detail( + asset_id="a1", + quality=100.0, + width=1280, + height=720, + duration=10.0, + file_size=5_000_000, + target_width=1280, + target_height=720, + ) + assert detail.resolution_score == 1.0 + + def test_total_score_matches_components(self): + detail = score_asset_detail( + asset_id="a1", + quality=80.0, + width=1920, + height=1080, + duration=10.0, + file_size=5_000_000, + ) + expected = calculate_total_score( + detail.quality_score, + detail.resolution_score, + detail.duration_score, + detail.bitrate_score, + ) + assert detail.total_score == pytest.approx(expected, abs=0.001) + + +# ============================================================ +# _bucket_by_duration +# ============================================================ + + +class TestBucketByDuration: + def test_none_is_unknown(self): + item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, None) + assert _bucket_by_duration(item) == "unknown" + + def test_short(self): + item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 3.0) + assert _bucket_by_duration(item) == "short" + + def test_short_boundary(self): + item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 4.9) + assert _bucket_by_duration(item) == "short" + + def test_medium(self): + item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 10.0) + assert _bucket_by_duration(item) == "medium" + + def test_medium_boundary(self): + item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 5.0) + assert _bucket_by_duration(item) == "medium" + + def test_medium_upper_boundary(self): + item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 14.9) + assert _bucket_by_duration(item) == "medium" + + def test_long(self): + item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 20.0) + assert _bucket_by_duration(item) == "long" + + def test_long_boundary(self): + item = AssetScoreDetail("a", 0.5, 0.5, 0.5, 0.5, 0.5, 15.0) + assert _bucket_by_duration(item) == "long" + + +# ============================================================ +# diverse_selection +# ============================================================ + + +def _make_detail(asset_id: str, score: float, duration: float) -> 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 TestDiverseSelection: + def test_empty_input_returns_empty(self): + assert diverse_selection([], 5) == [] + + def test_zero_count_returns_empty(self): + items = [_make_detail("a1", 0.9, 10.0)] + assert diverse_selection(items, 0) == [] + + def test_negative_count_returns_empty(self): + items = [_make_detail("a1", 0.9, 10.0)] + assert diverse_selection(items, -1) == [] + + def test_fewer_items_than_count(self): + items = [_make_detail("a1", 0.9, 10.0), _make_detail("a2", 0.8, 3.0)] + result = diverse_selection(items, 10) + assert len(result) == 2 + + def test_picks_top_from_each_bucket(self): + # 3个桶各有3个素材,选3个 + items = [ + _make_detail("s1", 0.95, 2.0), + _make_detail("m1", 0.9, 10.0), + _make_detail("l1", 0.85, 20.0), + _make_detail("s2", 0.8, 3.0), + _make_detail("m2", 0.75, 8.0), + _make_detail("l2", 0.7, 25.0), + ] + result = diverse_selection(items, 3) + assert len(result) == 3 + ids = [d.asset_id for d in result] + assert "s1" in ids + assert "m1" in ids + assert "l1" in ids + + def test_base_quota_when_count_large(self): + # count=6, base_quota=max(1, 6//3)=2 + items = [ + _make_detail("s1", 1.0, 2.0), + _make_detail("s2", 0.9, 3.0), + _make_detail("s3", 0.8, 4.0), + _make_detail("m1", 0.95, 10.0), + _make_detail("m2", 0.85, 12.0), + _make_detail("l1", 0.92, 20.0), + _make_detail("l2", 0.82, 30.0), + ] + result = diverse_selection(items, 6) + assert len(result) == 6 + ids = [d.asset_id for d in result] + # 每桶至少2个 + short_count = sum(1 for d in result if d.duration and d.duration < 5) + assert short_count >= 2 + + def test_remaining_filled_by_global_score(self): + # 只有2个桶有内容,count=5,配额用完后剩余从全局取 + items = [ + _make_detail("s1", 1.0, 2.0), + _make_detail("s2", 0.9, 3.0), + _make_detail("m1", 0.95, 10.0), + _make_detail("m2", 0.8, 12.0), + _make_detail("s3", 0.7, 4.0), + _make_detail("s4", 0.6, 1.0), + _make_detail("m3", 0.5, 8.0), + ] + result = diverse_selection(items, 5) + assert len(result) == 5 + # 最高分的都应该在 + ids = [d.asset_id for d in result] + assert "s1" in ids + assert "m1" in ids + + def test_single_bucket(self): + items = [ + _make_detail("s1", 1.0, 2.0), + _make_detail("s2", 0.9, 3.0), + _make_detail("s3", 0.8, 4.0), + ] + result = diverse_selection(items, 2) + assert len(result) == 2 + assert result[0].asset_id == "s1" + assert result[1].asset_id == "s2" + + def test_unknown_duration_fallback(self): + # 已知素材不够时用未知时长的补充 + items = [ + _make_detail("s1", 1.0, 2.0), + _make_detail("u1", 0.95, None), + _make_detail("u2", 0.9, None), + ] + result = diverse_selection(items, 3) + assert len(result) == 3 + ids = [d.asset_id for d in result] + assert "s1" in ids + assert "u1" in ids + + def test_no_duplicates(self): + items = [ + _make_detail("s1", 1.0, 2.0), + _make_detail("m1", 0.9, 10.0), + ] + result = diverse_selection(items, 5) + ids = [d.asset_id for d in result] + assert len(ids) == len(set(ids)) + + +# ============================================================ +# filter_candidates +# ============================================================ + + +@dataclass +class FakeAsset: + status: str = "ready" + mime_type: str = "video/mp4" + quality_score: float | None = 50.0 + + +class TestFilterCandidates: + def test_ready_video_passes(self): + assets = [FakeAsset()] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 1 + assert filtered == 0 + + def test_non_ready_filtered(self): + assets = [FakeAsset(status="uploading"), FakeAsset(status="processing")] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 0 + assert filtered == 0 # 被状态过滤的不计入质量门槛 + + def test_non_video_filtered(self): + assets = [FakeAsset(mime_type="image/jpeg"), FakeAsset(mime_type="audio/mp3")] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 0 + + def test_low_quality_filtered(self): + assets = [FakeAsset(quality_score=10.0), FakeAsset(quality_score=80.0)] + candidates, filtered = filter_candidates(assets, min_quality_score=30.0) + assert len(candidates) == 1 + assert filtered == 1 + + def test_quality_none_passes(self): + assets = [FakeAsset(quality_score=None)] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 1 + assert filtered == 0 + + def test_exactly_min_quality_passes(self): + assets = [FakeAsset(quality_score=30.0)] + candidates, filtered = filter_candidates(assets, min_quality_score=30.0) + assert len(candidates) == 1 + + def test_custom_min_quality(self): + assets = [ + FakeAsset(quality_score=40.0), + FakeAsset(quality_score=60.0), + FakeAsset(quality_score=80.0), + ] + candidates, filtered = filter_candidates(assets, min_quality_score=50.0) + assert len(candidates) == 2 + assert filtered == 1 + + def test_empty_input(self): + candidates, filtered = filter_candidates([]) + assert candidates == [] + assert filtered == 0 + + def test_mime_type_none(self): + # None 的 mime_type 也应该被过滤掉(不是video开头) + asset = FakeAsset(mime_type="") + candidates, _ = filter_candidates([asset]) + assert len(candidates) == 0 + + def test_with_enum_status(self): + from enum import Enum + + class StatusEnum(Enum): + READY = "ready" + UPLOADING = "uploading" + + @dataclass + class EnumAsset: + status: StatusEnum = StatusEnum.READY + mime_type: str = "video/mp4" + quality_score: float = 50.0 + + assets = [EnumAsset()] + candidates, filtered = filter_candidates(assets) + assert len(candidates) == 1