"""第78波：Duplication 查重领域模型纯逻辑单测。""" import pytest from packages.domain.duplication import DuplicateSegment, DuplicationRecord # ============================================================ # DuplicateSegment.create 测试 # ============================================================ class TestDuplicateSegmentCreate: def test_create_basic(self): seg = DuplicateSegment.create( source_start=10.0, source_end=20.0, matched_video_id="mv1", matched_video_name="匹配视频", matched_start=5.0, matched_end=15.0, similarity=85.5, ) assert seg.id assert seg.source_start == 10.0 assert seg.source_end == 20.0 assert seg.matched_video_id == "mv1" assert seg.matched_video_name == "匹配视频" assert seg.matched_start == 5.0 assert seg.matched_end == 15.0 assert seg.similarity == 85.5 def test_create_invalid_source_start_negative(self): with pytest.raises(ValueError, match="invalid source segment range"): DuplicateSegment.create( source_start=-1.0, source_end=10.0, matched_video_id="mv1", matched_video_name="n", matched_start=0.0, matched_end=10.0, similarity=50.0, ) def test_create_invalid_source_end_lte_start(self): with pytest.raises(ValueError, match="invalid source segment range"): DuplicateSegment.create( source_start=20.0, source_end=10.0, matched_video_id="mv1", matched_video_name="n", matched_start=0.0, matched_end=10.0, similarity=50.0, ) def test_create_source_end_equal_start_invalid(self): with pytest.raises(ValueError, match="invalid source segment range"): DuplicateSegment.create( source_start=10.0, source_end=10.0, matched_video_id="mv1", matched_video_name="n", matched_start=0.0, matched_end=10.0, similarity=50.0, ) def test_create_invalid_matched_start_negative(self): with pytest.raises(ValueError, match="invalid matched segment range"): DuplicateSegment.create( source_start=0.0, source_end=10.0, matched_video_id="mv1", matched_video_name="n", matched_start=-1.0, matched_end=10.0, similarity=50.0, ) def test_create_invalid_matched_end_lte_start(self): with pytest.raises(ValueError, match="invalid matched segment range"): DuplicateSegment.create( source_start=0.0, source_end=10.0, matched_video_id="mv1", matched_video_name="n", matched_start=15.0, matched_end=5.0, similarity=50.0, ) def test_create_similarity_zero(self): seg = DuplicateSegment.create( source_start=0.0, source_end=5.0, matched_video_id="mv1", matched_video_name="n", matched_start=0.0, matched_end=5.0, similarity=0.0, ) assert seg.similarity == 0.0 def test_create_similarity_hundred(self): seg = DuplicateSegment.create( source_start=0.0, source_end=5.0, matched_video_id="mv1", matched_video_name="n", matched_start=0.0, matched_end=5.0, similarity=100.0, ) assert seg.similarity == 100.0 def test_create_similarity_negative_raises(self): with pytest.raises(ValueError, match="similarity must be between 0 and 100"): DuplicateSegment.create( source_start=0.0, source_end=5.0, matched_video_id="mv1", matched_video_name="n", matched_start=0.0, matched_end=5.0, similarity=-1.0, ) def test_create_similarity_over_100_raises(self): with pytest.raises(ValueError, match="similarity must be between 0 and 100"): DuplicateSegment.create( source_start=0.0, source_end=5.0, matched_video_id="mv1", matched_video_name="n", matched_start=0.0, matched_end=5.0, similarity=101.0, ) def test_create_id_unique(self): s1 = DuplicateSegment.create(0, 5, "m", "n", 0, 5, 50.0) s2 = DuplicateSegment.create(0, 5, "m", "n", 0, 5, 50.0) assert s1.id != s2.id # ============================================================ # DuplicationRecord.create 测试 # ============================================================ class TestDuplicationRecordCreate: def test_create_minimal(self): rec = DuplicationRecord.create( user_id="u1", filename="test.mp4", file_size=102400, storage_key="oss://key", ) assert rec.id assert rec.user_id == "u1" assert rec.filename == "test.mp4" assert rec.file_size == 102400 assert rec.storage_key == "oss://key" assert rec.duration_seconds == 0.0 assert rec.status == "pending" assert rec.duplicate_rate is None assert rec.duplicate_count == 0 assert rec.segments == [] assert rec.error_message == "" def test_create_with_duration(self): rec = DuplicationRecord.create( user_id="u1", filename="test.mp4", file_size=1024, storage_key="oss://k", duration_seconds=120.5, ) assert rec.duration_seconds == 120.5 def test_create_empty_user_id_raises(self): with pytest.raises(ValueError, match="user_id cannot be empty"): DuplicationRecord.create( user_id=" ", filename="t.mp4", file_size=100, storage_key="k", ) def test_create_empty_filename_raises(self): with pytest.raises(ValueError, match="filename cannot be empty"): DuplicationRecord.create( user_id="u1", filename="", file_size=100, storage_key="k", ) def test_create_zero_file_size_raises(self): with pytest.raises(ValueError, match="file_size must be positive"): DuplicationRecord.create( user_id="u1", filename="t.mp4", file_size=0, storage_key="k", ) def test_create_negative_file_size_raises(self): with pytest.raises(ValueError, match="file_size must be positive"): DuplicationRecord.create( user_id="u1", filename="t.mp4", file_size=-100, storage_key="k", ) def test_create_strips_whitespace(self): rec = DuplicationRecord.create( user_id=" u1 ", filename=" test.mp4 ", file_size=100, storage_key=" oss://k ", ) assert rec.user_id == "u1" assert rec.filename == "test.mp4" # storage_key 不确定有没有 strip，看源码是直接赋值 assert rec.storage_key == " oss://k " def test_create_id_unique(self): r1 = DuplicationRecord.create("u1", "a.mp4", 100, "k1") r2 = DuplicationRecord.create("u1", "b.mp4", 200, "k2") assert r1.id != r2.id def test_create_default_segments_empty_list(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") assert rec.segments == [] assert isinstance(rec.segments, list) # ============================================================ # DuplicationRecord 状态流转测试 # ============================================================ class TestDuplicationRecordStatusFlow: def test_mark_processing(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") before = rec.updated_at rec.mark_processing() assert rec.status == "processing" assert rec.updated_at >= before def test_mark_completed(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") seg = DuplicateSegment.create(0, 5, "m", "n", 0, 5, 90.0) rec.mark_completed( duplicate_rate=35.5, duplicate_count=1, segments=[seg], ) assert rec.status == "completed" assert rec.duplicate_rate == 35.5 assert rec.duplicate_count == 1 assert len(rec.segments) == 1 assert rec.segments[0].id == seg.id def test_mark_completed_zero_rate(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") rec.mark_completed(0.0, 0, []) assert rec.status == "completed" assert rec.duplicate_rate == 0.0 assert rec.duplicate_count == 0 assert rec.segments == [] def test_mark_completed_full_rate(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") seg = DuplicateSegment.create(0, 10, "m", "n", 0, 10, 100.0) rec.mark_completed(100.0, 1, [seg]) assert rec.duplicate_rate == 100.0 def test_mark_completed_invalid_rate_negative(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") with pytest.raises(ValueError, match="duplicate_rate must be between 0 and 100"): rec.mark_completed(-1.0, 0, []) def test_mark_completed_invalid_rate_over_100(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") with pytest.raises(ValueError, match="duplicate_rate must be between 0 and 100"): rec.mark_completed(100.1, 0, []) def test_mark_failed(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") rec.mark_failed("网络超时") assert rec.status == "failed" assert rec.error_message == "网络超时" def test_mark_failed_from_processing(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") rec.mark_processing() rec.mark_failed("解析失败") assert rec.status == "failed" assert rec.error_message == "解析失败" # ============================================================ # DuplicationRecord 重试机制测试 # ============================================================ class TestDuplicationRecordRetry: def test_can_retry_failed_true(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") rec.mark_failed("err") assert rec.can_retry() is True def test_can_retry_pending_false(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") assert rec.can_retry() is False def test_can_retry_processing_false(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") rec.mark_processing() assert rec.can_retry() is False def test_can_retry_completed_false(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") rec.mark_completed(10.0, 0, []) assert rec.can_retry() is False def test_reset_for_retry_clears_all(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") seg = DuplicateSegment.create(0, 5, "m", "n", 0, 5, 80.0) rec.mark_completed(30.0, 1, [seg]) # 先设为 failed 再 reset rec.status = "failed" rec.reset_for_retry() assert rec.status == "pending" assert rec.duplicate_rate is None assert rec.duplicate_count == 0 assert rec.error_message == "" assert rec.segments == [] assert rec.video_fingerprint is None def test_reset_for_retry_updates_timestamp(self): rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") rec.status = "failed" before = rec.updated_at rec.reset_for_retry() assert rec.updated_at >= before def test_full_retry_flow(self): """完整的 创建→失败→重置→再处理→完成 流程。""" rec = DuplicationRecord.create("u1", "a.mp4", 100, "k") assert rec.status == "pending" rec.mark_processing() assert rec.status == "processing" rec.mark_failed("超时") assert rec.can_retry() rec.reset_for_retry() assert rec.status == "pending" assert rec.error_message == "" rec.mark_processing() rec.mark_completed(0.0, 0, []) assert rec.status == "completed" assert not rec.can_retry() # ============================================================ # QuotaTier 测试 # ============================================================ class TestQuotaTier: def test_get_limit_defined(self): from packages.domain.quota import QUOTA_TIERS, QuotaTier free = QUOTA_TIERS["free"] assert free.get_limit("storage_gb") == 2 assert free.get_limit("videos_per_month") == 5 def test_get_limit_undefined_returns_zero(self): from packages.domain.quota import QuotaTier tier = QuotaTier(name="test", limits={"a": 10}) assert tier.get_limit("nonexistent") == 0 def test_is_unlimited_inf(self): from packages.domain.quota import QUOTA_TIERS premium = QUOTA_TIERS["premium"] assert premium.is_unlimited("max_templates") is True def test_is_unlimited_finite(self): from packages.domain.quota import QUOTA_TIERS free = QUOTA_TIERS["free"] assert free.is_unlimited("storage_gb") is False def test_is_unlimited_undefined_returns_true(self): """未定义的维度 limits.get 默认 inf，is_unlimited 返回 True。""" from packages.domain.quota import QuotaTier tier = QuotaTier(name="test", limits={}) assert tier.is_unlimited("unknown") is True # ============================================================ # QUOTA_TIERS 三档套餐验证 # ============================================================ class TestQuotaTiers: def test_free_tier_storage(self): from packages.domain.quota import QUOTA_TIERS assert QUOTA_TIERS["free"].get_limit("storage_gb") == 2 def test_free_tier_no_ai_voice(self): from packages.domain.quota import QUOTA_TIERS assert QUOTA_TIERS["free"].get_limit("ai_voice_enabled") == 0 def test_basic_tier_storage(self): from packages.domain.quota import QUOTA_TIERS assert QUOTA_TIERS["basic"].get_limit("storage_gb") == 20 def test_basic_tier_has_ai_voice(self): from packages.domain.quota import QUOTA_TIERS assert QUOTA_TIERS["basic"].get_limit("ai_voice_enabled") == 1 def test_basic_tier_ai_voice_credits(self): from packages.domain.quota import QUOTA_TIERS assert QUOTA_TIERS["basic"].get_limit("ai_voice_credits") == 100 def test_premium_tier_storage(self): from packages.domain.quota import QUOTA_TIERS assert QUOTA_TIERS["premium"].get_limit("storage_gb") == 100 def test_premium_tier_unlimited_templates(self): from packages.domain.quota import QUOTA_TIERS assert QUOTA_TIERS["premium"].is_unlimited("max_templates") def test_premium_tier_multi_platform(self): from packages.domain.quota import QUOTA_TIERS assert QUOTA_TIERS["premium"].get_limit("multi_platform_enabled") == 1 def test_basic_no_multi_platform(self): from packages.domain.quota import QUOTA_TIERS assert QUOTA_TIERS["basic"].get_limit("multi_platform_enabled") == 0 def test_all_tiers_exist(self): from packages.domain.quota import QUOTA_TIERS assert set(QUOTA_TIERS.keys()) == {"free", "basic", "premium", "pro"} # ============================================================ # QuotaCheckResult 测试 # ============================================================ class TestQuotaCheckResult: def test_usage_percent_normal(self): from packages.domain.quota import QuotaCheckResult, QuotaWarningLevel r = QuotaCheckResult( allowed=True, dimension="storage_gb", limit=100, used=50, remaining=50, warning_level=QuotaWarningLevel.NORMAL, ) assert r.usage_percent == 50.0 def test_usage_percent_capped_at_100(self): from packages.domain.quota import QuotaCheckResult, QuotaWarningLevel r = QuotaCheckResult( allowed=False, dimension="x", limit=10, used=15, remaining=0, warning_level=QuotaWarningLevel.EXCEEDED, ) assert r.usage_percent == 100.0 def test_usage_percent_unlimited(self): from packages.domain.quota import QuotaCheckResult, QuotaWarningLevel r = QuotaCheckResult( allowed=True, dimension="x", limit=float("inf"), used=999, remaining=float("inf"), warning_level=QuotaWarningLevel.NORMAL, ) assert r.usage_percent == 0.0 def test_usage_percent_zero_limit_with_usage(self): from packages.domain.quota import QuotaCheckResult, QuotaWarningLevel r = QuotaCheckResult( allowed=False, dimension="x", limit=0, used=5, remaining=0, warning_level=QuotaWarningLevel.EXCEEDED, ) assert r.usage_percent == 100.0 def test_usage_percent_zero_limit_no_usage(self): from packages.domain.quota import QuotaCheckResult, QuotaWarningLevel r = QuotaCheckResult( allowed=True, dimension="x", limit=0, used=0, remaining=0, warning_level=QuotaWarningLevel.NORMAL, ) assert r.usage_percent == 0.0 # ============================================================ # QuotaWarningLevel 计算测试 # ============================================================ class TestWarningLevel: def test_normal_below_80(self): from packages.domain.quota import QuotaWarningLevel, get_warning_level assert get_warning_level(50, 100) == QuotaWarningLevel.NORMAL def test_warning_at_80(self): from packages.domain.quota import QuotaWarningLevel, get_warning_level assert get_warning_level(80, 100) == QuotaWarningLevel.WARNING def test_warning_below_95(self): from packages.domain.quota import QuotaWarningLevel, get_warning_level assert get_warning_level(90, 100) == QuotaWarningLevel.WARNING def test_critical_at_95(self): from packages.domain.quota import QuotaWarningLevel, get_warning_level assert get_warning_level(95, 100) == QuotaWarningLevel.CRITICAL def test_critical_below_100(self): from packages.domain.quota import QuotaWarningLevel, get_warning_level assert get_warning_level(99, 100) == QuotaWarningLevel.CRITICAL def test_exceeded_at_100(self): from packages.domain.quota import QuotaWarningLevel, get_warning_level assert get_warning_level(100, 100) == QuotaWarningLevel.EXCEEDED def test_exceeded_over_100(self): from packages.domain.quota import QuotaWarningLevel, get_warning_level assert get_warning_level(150, 100) == QuotaWarningLevel.EXCEEDED def test_unlimited_always_normal(self): from packages.domain.quota import QuotaWarningLevel, get_warning_level assert get_warning_level(99999, float("inf")) == QuotaWarningLevel.NORMAL def test_zero_limit_with_usage_is_exceeded(self): from packages.domain.quota import QuotaWarningLevel, get_warning_level assert get_warning_level(1, 0) == QuotaWarningLevel.EXCEEDED def test_zero_limit_no_usage_is_normal(self): from packages.domain.quota import QuotaWarningLevel, get_warning_level assert get_warning_level(0, 0) == QuotaWarningLevel.NORMAL # ============================================================ # QuotaChecker 测试 # ============================================================ class TestQuotaChecker: def test_check_allowed(self): from packages.domain.quota import QuotaChecker, QuotaWarningLevel checker = QuotaChecker() r = checker.check("free", "storage_gb", 1) assert r.allowed is True assert r.limit == 2 assert r.used == 1 assert r.remaining == 1 assert r.warning_level == QuotaWarningLevel.NORMAL def test_check_exceeded(self): from packages.domain.quota import QuotaChecker, QuotaWarningLevel checker = QuotaChecker() r = checker.check("free", "storage_gb", 5) assert r.allowed is False assert r.remaining == 0 assert r.warning_level == QuotaWarningLevel.EXCEEDED def test_check_at_limit_not_allowed(self): """used == limit 时 allowed 为 False（严格小于才算允许）。""" from packages.domain.quota import QuotaChecker checker = QuotaChecker() r = checker.check("free", "storage_gb", 2) assert r.allowed is False def test_check_unlimited_always_allowed(self): from packages.domain.quota import QuotaChecker, QuotaWarningLevel checker = QuotaChecker() r = checker.check("premium", "max_templates", 9999) assert r.allowed is True assert r.remaining == float("inf") assert r.warning_level == QuotaWarningLevel.NORMAL def test_check_unknown_plan_returns_zero(self): from packages.domain.quota import QuotaChecker, QuotaWarningLevel checker = QuotaChecker() r = checker.check("unknown_plan", "storage_gb", 1) assert r.allowed is False assert r.limit == 0 assert r.warning_level == QuotaWarningLevel.EXCEEDED def test_check_warning_level_80_percent(self): from packages.domain.quota import QuotaChecker, QuotaWarningLevel checker = QuotaChecker() r = checker.check("basic", "storage_gb", 16) # 20 * 0.8 = 16 assert r.warning_level == QuotaWarningLevel.WARNING def test_check_multiple(self): from packages.domain.quota import QuotaChecker checker = QuotaChecker() results = checker.check_multiple( "free", {"storage_gb": 1, "videos_per_month": 10, "max_templates": 2}, ) assert len(results) == 3 dims = {r.dimension: r for r in results} assert dims["storage_gb"].allowed is True assert dims["videos_per_month"].allowed is False assert dims["max_templates"].allowed is True # ============================================================ # QuotaRegistry 测试 # ============================================================ class TestQuotaRegistry: def test_list_dimensions_includes_builtin(self): from packages.domain.quota import QuotaRegistry reg = QuotaRegistry() dims = reg.list_dimensions() assert "storage_gb" in dims assert "videos_per_month" in dims assert "ai_voice_enabled" in dims def test_list_tiers(self): from packages.domain.quota import QuotaRegistry reg = QuotaRegistry() tiers = reg.list_tiers() assert set(tiers) == {"free", "basic", "premium", "pro"} def test_register_new_dimension(self): from packages.domain.quota import QuotaRegistry reg = QuotaRegistry() reg.register_dimension("custom_dim", "自定义维度", {"free": 5, "basic": 20}) dims = reg.list_dimensions() assert "custom_dim" in dims assert dims["custom_dim"] == "自定义维度" assert reg.get_limit("free", "custom_dim") == 5 assert reg.get_limit("basic", "custom_dim") == 20 def test_register_dimension_idempotent(self): from packages.domain.quota import QuotaRegistry reg = QuotaRegistry() reg.register_dimension("custom_dim", "v1", {"free": 5}) reg.register_dimension("custom_dim", "v2", {"free": 10}) # 第二次应该被忽略（幂等），描述和限制都保持第一次 assert reg.list_dimensions()["custom_dim"] == "v1" assert reg.get_limit("free", "custom_dim") == 5 def test_register_without_defaults_defaults_to_zero(self): from packages.domain.quota import QuotaRegistry reg = QuotaRegistry() reg.register_dimension("new_dim", "新维度") assert reg.get_limit("free", "new_dim") == 0 assert reg.get_limit("basic", "new_dim") == 0 assert reg.get_limit("premium", "new_dim") == 0 def test_get_tier_exists(self): from packages.domain.quota import QuotaRegistry reg = QuotaRegistry() tier = reg.get_tier("free") assert tier is not None assert tier.name == "free" def test_get_tier_not_exists(self): from packages.domain.quota import QuotaRegistry reg = QuotaRegistry() assert reg.get_tier("nonexistent") is None def test_global_registry_instance(self): from packages.domain.quota import quota_checker, quota_registry assert quota_registry is not None assert quota_checker is not None assert quota_registry.get_limit("free", "storage_gb") == 2