diff --git a/tests/unit/test_wave78_duplication_domain.py b/tests/unit/test_wave78_duplication_domain.py new file mode 100755 index 000000000..d2511cd2d --- /dev/null +++ b/tests/unit/test_wave78_duplication_domain.py @@ -0,0 +1,742 @@ +"""第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 QuotaTier, QUOTA_TIERS + + 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"} + + +# ============================================================ +# 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 get_warning_level, QuotaWarningLevel + + assert get_warning_level(50, 100) == QuotaWarningLevel.NORMAL + + def test_warning_at_80(self): + from packages.domain.quota import get_warning_level, QuotaWarningLevel + + assert get_warning_level(80, 100) == QuotaWarningLevel.WARNING + + def test_warning_below_95(self): + from packages.domain.quota import get_warning_level, QuotaWarningLevel + + assert get_warning_level(90, 100) == QuotaWarningLevel.WARNING + + def test_critical_at_95(self): + from packages.domain.quota import get_warning_level, QuotaWarningLevel + + assert get_warning_level(95, 100) == QuotaWarningLevel.CRITICAL + + def test_critical_below_100(self): + from packages.domain.quota import get_warning_level, QuotaWarningLevel + + assert get_warning_level(99, 100) == QuotaWarningLevel.CRITICAL + + def test_exceeded_at_100(self): + from packages.domain.quota import get_warning_level, QuotaWarningLevel + + assert get_warning_level(100, 100) == QuotaWarningLevel.EXCEEDED + + def test_exceeded_over_100(self): + from packages.domain.quota import get_warning_level, QuotaWarningLevel + + assert get_warning_level(150, 100) == QuotaWarningLevel.EXCEEDED + + def test_unlimited_always_normal(self): + from packages.domain.quota import get_warning_level, QuotaWarningLevel + + assert get_warning_level(99999, float("inf")) == QuotaWarningLevel.NORMAL + + def test_zero_limit_with_usage_is_exceeded(self): + from packages.domain.quota import get_warning_level, QuotaWarningLevel + + assert get_warning_level(1, 0) == QuotaWarningLevel.EXCEEDED + + def test_zero_limit_no_usage_is_normal(self): + from packages.domain.quota import get_warning_level, QuotaWarningLevel + + 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"} + + 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_registry, quota_checker + + assert quota_registry is not None + assert quota_checker is not None + assert quota_registry.get_limit("free", "storage_gb") == 2