"""SmartAssetSelector 智能素材选择服务单元测试.""" from __future__ import annotations import unittest from dataclasses import dataclass from app.services.smart_asset_selector import ( _MEDIUM_BUCKET_MAX, _SHORT_BUCKET_MAX, SmartAssetSelector, ) @dataclass class MockAsset: """模拟 Asset 实体.""" id: str quality_score: float | None = None width: int | None = None height: int | None = None duration: float | None = None file_size: int = 0 mime_type: str = "video/mp4" status: str = "ready" @property def status_value(self) -> str: return self.status class TestSmartAssetSelectorScoring(unittest.TestCase): """评分维度测试.""" def setUp(self): self.selector = SmartAssetSelector() def test_quality_score_normalization(self): """质量分正确归一化到 0-1.""" asset_high = MockAsset(id="1", quality_score=90.0) asset_low = MockAsset(id="2", quality_score=30.0) asset_none = MockAsset(id="3", quality_score=None) detail_high = self.selector._score_asset(asset_high) detail_low = self.selector._score_asset(asset_low) detail_none = self.selector._score_asset(asset_none) # 90分 → 0.9 × 0.5权重 = 0.45 基础贡献 self.assertAlmostEqual(detail_high.quality_score, 0.9, delta=0.01) # 30分 → 0.3 × 0.5权重 = 0.15 基础贡献 self.assertAlmostEqual(detail_low.quality_score, 0.3, delta=0.01) # 无质量分给默认 0.5 self.assertAlmostEqual(detail_none.quality_score, 0.5, delta=0.01) def test_resolution_score_1080p_full(self): """1080p 分辨率得满分.""" asset = MockAsset(id="1", width=1920, height=1080) detail = self.selector._score_asset(asset) self.assertAlmostEqual(detail.resolution_score, 1.0, delta=0.01) def test_resolution_score_4k_full(self): """4K 也得满分(高于目标分辨率不扣分).""" asset = MockAsset(id="1", width=3840, height=2160) detail = self.selector._score_asset(asset) self.assertAlmostEqual(detail.resolution_score, 1.0, delta=0.01) def test_resolution_score_720p_lower(self): """720p 低于 1080p,得分低于 1.""" asset = MockAsset(id="1", width=1280, height=720) detail = self.selector._score_asset(asset) self.assertLess(detail.resolution_score, 1.0) self.assertGreater(detail.resolution_score, 0.3) def test_resolution_score_none(self): """分辨率未知给中评分.""" asset = MockAsset(id="1", width=None, height=None) detail = self.selector._score_asset(asset) self.assertAlmostEqual(detail.resolution_score, 0.5, delta=0.01) def test_duration_score_optimal(self): """最佳时长区间内得满分.""" asset = MockAsset(id="1", duration=10.0) detail = self.selector._score_asset(asset) self.assertAlmostEqual(detail.duration_score, 1.0, delta=0.01) def test_duration_score_too_short(self): """时长过短扣分.""" asset = MockAsset(id="1", duration=1.0) detail = self.selector._score_asset(asset) self.assertLess(detail.duration_score, 1.0) def test_duration_score_too_long(self): """时长过长扣分.""" asset = MockAsset(id="1", duration=120.0) detail = self.selector._score_asset(asset) self.assertLess(detail.duration_score, 1.0) def test_duration_score_none(self): """时长未知给中评分.""" asset = MockAsset(id="1", duration=None) detail = self.selector._score_asset(asset) self.assertAlmostEqual(detail.duration_score, 0.5, delta=0.01) def test_total_score_weighted_sum(self): """总分是各维度的加权和.""" asset = MockAsset( id="1", quality_score=100.0, # 1.0 × 0.5 = 0.5 width=1920, # 1.0 × 0.2 = 0.2 height=1080, duration=10.0, # 1.0 × 0.2 = 0.2 file_size=10_000_000, # ~8Mbps,10秒 → 约 1.0 × 0.1 = 0.1 ) detail = self.selector._score_asset(asset) # 理论上接近 1.0 self.assertGreater(detail.total_score, 0.85) self.assertLessEqual(detail.total_score, 1.0) class TestSmartAssetSelectorSelection(unittest.TestCase): """选择逻辑测试.""" def setUp(self): self.selector = SmartAssetSelector(min_quality_score=0) # 测试时关闭质量门槛 def _make_assets(self, count: int, base_quality: float = 80.0) -> list[MockAsset]: assets = [] for i in range(count): assets.append( MockAsset( id=f"asset_{i}", quality_score=base_quality - i * 5, # 质量递减 width=1920, height=1080, duration=10.0 + i, file_size=5_000_000 + i * 100_000, ) ) return assets def test_select_all_when_count_zero(self): """count=0 时返回全部符合条件的.""" assets = self._make_assets(10) result = self.selector.select(assets, count=0) self.assertEqual(len(result.selected_ids), 10) self.assertEqual(result.total_candidates, 10) def test_select_top_n(self): """返回指定数量的 top N.""" assets = self._make_assets(10) result = self.selector.select(assets, count=3) self.assertEqual(len(result.selected_ids), 3) # 最高分的应该是 asset_0(质量分最高) self.assertEqual(result.selected_ids[0], "asset_0") def test_select_more_than_available(self): """请求数量超过候选数量时返回全部.""" assets = self._make_assets(5) result = self.selector.select(assets, count=10) self.assertEqual(len(result.selected_ids), 5) def test_filter_non_ready(self): """非 ready 状态的素材被过滤.""" assets = [ MockAsset(id="1", quality_score=90.0, status="ready"), MockAsset(id="2", quality_score=80.0, status="processing"), MockAsset(id="3", quality_score=70.0, status="ready"), ] result = self.selector.select(assets, count=0) self.assertEqual(len(result.selected_ids), 2) self.assertIn("1", result.selected_ids) self.assertIn("3", result.selected_ids) self.assertNotIn("2", result.selected_ids) def test_filter_non_video(self): """非视频素材被过滤.""" assets = [ MockAsset(id="1", quality_score=90.0, mime_type="video/mp4"), MockAsset(id="2", quality_score=80.0, mime_type="image/jpeg"), MockAsset(id="3", quality_score=70.0, mime_type="video/quicktime"), ] result = self.selector.select(assets, count=0) self.assertEqual(len(result.selected_ids), 2) def test_min_quality_filter(self): """最低质量分门槛过滤.""" selector = SmartAssetSelector(min_quality_score=60.0) assets = [ MockAsset(id="1", quality_score=90.0), MockAsset(id="2", quality_score=50.0), # 低于门槛 MockAsset(id="3", quality_score=70.0), MockAsset(id="4", quality_score=30.0), # 低于门槛 ] result = selector.select(assets, count=0) self.assertEqual(len(result.selected_ids), 2) self.assertEqual(result.filtered_out, 2) self.assertIn("1", result.selected_ids) self.assertIn("3", result.selected_ids) def test_empty_input(self): """空输入返回空结果.""" result = self.selector.select([], count=5) self.assertEqual(result.selected_ids, []) self.assertEqual(result.total_candidates, 0) self.assertEqual(result.avg_score, 0.0) def test_sorted_by_score_descending(self): """结果按总分降序排列.""" assets = self._make_assets(5) result = self.selector.select(assets, count=0, ensure_diversity=False) scores = [d.total_score for d in result.details] # 应该是降序 self.assertEqual(scores, sorted(scores, reverse=True)) class TestSmartAssetSelectorDiversity(unittest.TestCase): """多样性选择测试.""" def setUp(self): self.selector = SmartAssetSelector(min_quality_score=0) def _make_assets(self, count: int, base_quality: float = 80.0) -> list[MockAsset]: assets = [] for i in range(count): assets.append( MockAsset( id=f"asset_{i}", quality_score=base_quality - i * 5, width=1920, height=1080, duration=10.0 + i, file_size=5_000_000 + i * 100_000, ) ) return assets def test_diversity_all_short(self): """全是短素材时不报错,正常返回.""" assets = [] for i in range(10): assets.append( MockAsset( id=f"short_{i}", quality_score=80.0 + i, width=1920, height=1080, duration=2.0 + i * 0.1, # 都 < 5s file_size=1_000_000, ) ) result = self.selector.select(assets, count=5, ensure_diversity=True) self.assertEqual(len(result.selected_ids), 5) def test_diversity_mixed_buckets(self): """混合时长素材时,各桶都有代表.""" assets = [] # 短素材(质量分高) for i in range(5): assets.append( MockAsset( id=f"short_{i}", quality_score=95.0 - i, width=1920, height=1080, duration=3.0, file_size=2_000_000, ) ) # 中素材(质量分中等) for i in range(5): assets.append( MockAsset( id=f"medium_{i}", quality_score=85.0 - i, width=1920, height=1080, duration=10.0, file_size=5_000_000, ) ) # 长素材(质量分低) for i in range(5): assets.append( MockAsset( id=f"long_{i}", quality_score=75.0 - i, width=1920, height=1080, duration=60.0, file_size=20_000_000, ) ) result = self.selector.select(assets, count=6, ensure_diversity=True) selected = result.selected_ids # 6个素材,每个桶至少有1个(基础配额 max(1, 6//3)=2) short_count = sum(1 for sid in selected if sid.startswith("short_")) medium_count = sum(1 for sid in selected if sid.startswith("medium_")) long_count = sum(1 for sid in selected if sid.startswith("long_")) # 每个桶至少1个 self.assertGreaterEqual(short_count, 1) self.assertGreaterEqual(medium_count, 1) self.assertGreaterEqual(long_count, 1) self.assertEqual(len(selected), 6) def test_diversity_disabled_returns_top(self): """关闭多样性时,直接返回 top N(可能全是短素材).""" assets = [] # 短素材(质量分最高) for i in range(10): assets.append( MockAsset( id=f"short_{i}", quality_score=95.0 - i, width=1920, height=1080, duration=3.0, file_size=2_000_000, ) ) # 长素材(质量分低) for i in range(5): assets.append( MockAsset( id=f"long_{i}", quality_score=70.0, width=1920, height=1080, duration=60.0, file_size=20_000_000, ) ) result = self.selector.select(assets, count=5, ensure_diversity=False) selected = result.selected_ids # 全是短素材(因为质量分高) self.assertTrue(all(s.startswith("short_") for s in selected)) def test_avg_score_calculated(self): """平均分正确计算.""" assets = self._make_assets(3) result = self.selector.select(assets, count=3, ensure_diversity=False) expected_avg = sum(d.total_score for d in result.details) / 3 self.assertAlmostEqual(result.avg_score, expected_avg, delta=0.001) class TestSmartAssetSelectorEdgeCases(unittest.TestCase): """边界情况测试.""" def setUp(self): self.selector = SmartAssetSelector(min_quality_score=0) def _make_assets(self, count: int, base_quality: float = 80.0) -> list[MockAsset]: assets = [] for i in range(count): assets.append( MockAsset( id=f"asset_{i}", quality_score=base_quality - i * 5, width=1920, height=1080, duration=10.0 + i, file_size=5_000_000 + i * 100_000, ) ) return assets def test_single_asset(self): """单个素材正常返回.""" assets = [MockAsset(id="1", quality_score=80.0, width=1920, height=1080, duration=10.0)] result = self.selector.select(assets, count=1) self.assertEqual(len(result.selected_ids), 1) self.assertEqual(result.selected_ids[0], "1") def test_zero_width_height(self): """宽高为0时按未知处理.""" asset = MockAsset(id="1", width=0, height=0) detail = self.selector._score_asset(asset) self.assertAlmostEqual(detail.resolution_score, 0.5, delta=0.01) def test_negative_duration(self): """负时长按未知处理.""" asset = MockAsset(id="1", duration=-5.0) detail = self.selector._score_asset(asset) self.assertAlmostEqual(detail.duration_score, 0.5, delta=0.01) def test_zero_file_size_with_duration(self): """文件大小为0时码率评分中等.""" asset = MockAsset(id="1", file_size=0, duration=10.0) detail = self.selector._score_asset(asset) self.assertAlmostEqual(detail.bitrate_score, 0.5, delta=0.01) def test_bitrate_score_optimal(self): """最佳码率范围得满分.""" # 5 Mbps × 10秒 = 6.25 MB → file_size = 6,250,000 bytes asset = MockAsset(id="1", file_size=6_250_000, duration=10.0) detail = self.selector._score_asset(asset) self.assertAlmostEqual(detail.bitrate_score, 1.0, delta=0.01) def test_details_match_selected_ids(self): """details 列表和 selected_ids 顺序一致.""" assets = self._make_assets(5) result = self.selector.select(assets, count=3, ensure_diversity=False) self.assertEqual(len(result.details), 3) for i, aid in enumerate(result.selected_ids): self.assertEqual(result.details[i].asset_id, aid) if __name__ == "__main__": unittest.main()