test(unit): 第78波 - duplication查重 + quota配额系统 (+80)

- DuplicateSegment: create参数校验（时间范围/相似度），12个
- DuplicationRecord: create校验/状态流转/重试机制，23个
- QuotaTier: get_limit/is_unlimited，5个
- QUOTA_TIERS三档套餐验证：10个
- QuotaCheckResult: usage_percent边界，5个
- 警告级别计算：11个（80%/95%/100%/unlimited/零配额）
- QuotaChecker: check/check_multiple/边界情况，7个
- QuotaRegistry: 注册/幂等/查询，7个
- 全部本地全绿，black格式化通过

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