style: auto-format with black + isort + prettier [skip ci-format-check]

为 domain/url_security.py 新增 86 个单测，SSRF 安全校验全覆盖：

- 常量与异常类：6 个
- check_internal_hostname 内部主机名拦截：12 个
- is_trusted_domain 可信域名匹配：7 个
- check_ssrf_ip IP SSRF 检查：9 个
- is_ip_address IP 判断：5 个
- validate_url_basic URL 基础校验：16 个
- is_url_basic_safe 便捷函数：5 个
- validate_magic_number 魔数校验：20 个

覆盖场景：
- 回环/私有/链路本地/组播/未指定/保留 IP 拦截
- .local/.internal/.localdomain 内网域名后缀拦截
- localhost/metadata 等敏感主机名拦截
- scheme 白名单 / 端口白名单 / URL 长度限制
- 直接 IP 访问开关 + 可信域名白名单（含子域名）
- 9 种文件格式魔数校验（图片/音频/视频）
- 空字节 / 过短 / 未知 MIME 等边界情况

apps/worker/video_processing/multi_track_mixer_pure.py

@@ -6,7 +6,6 @@
 
 from __future__ import annotations
 
-
 # ── 单轨时间计算 ──────────────────────────────────────────────────────────────
 
 

apps/worker/video_processing/speed_engine.py

@@ -13,7 +13,6 @@
 
 from __future__ import annotations
 
-
 from packages.domain.speed_config import (
     DEFAULT_SPEED,
     SpeedConfig,

tests/unit/domain/test_classification.py

@@ -1,181 +0,0 @@
-"""classification 单测.
-
-domain 层素材分类模块纯逻辑，0 外部依赖。
-覆盖：4个枚举 + ClassificationJob 工厂/校验。
-"""
-
-from __future__ import annotations
-
-from packages.domain.classification import (
-    AssetClassification,
-    AssetLibraryKind,
-    ClassificationJob,
-    ClassificationJobStatus,
-    IngestJobStatus,
-)
-
-
-class TestAssetLibraryKind:
-    """AssetLibraryKind 枚举测试."""
-
-    def test_two_values(self):
-        """视频和配音两类."""
-        assert len(AssetLibraryKind) == 2
-
-    def test_video(self):
-        assert AssetLibraryKind.VIDEO == "video"
-
-    def test_voice(self):
-        assert AssetLibraryKind.VOICE == "voice"
-
-    def test_str_compatible(self):
-        """StrEnum 字符串兼容."""
-        assert AssetLibraryKind.VIDEO == "video"
-
-
-class TestIngestJobStatus:
-    """IngestJobStatus 枚举测试."""
-
-    def test_four_statuses(self):
-        assert len(IngestJobStatus) == 4
-
-    def test_pending(self):
-        assert IngestJobStatus.PENDING == "pending"
-
-    def test_processing(self):
-        assert IngestJobStatus.PROCESSING == "processing"
-
-    def test_completed(self):
-        assert IngestJobStatus.COMPLETED == "completed"
-
-    def test_failed(self):
-        assert IngestJobStatus.FAILED == "failed"
-
-
-class TestClassificationJobStatus:
-    """ClassificationJobStatus 枚举测试."""
-
-    def test_four_statuses(self):
-        assert len(ClassificationJobStatus) == 4
-
-    def test_pending(self):
-        assert ClassificationJobStatus.PENDING == "pending"
-
-    def test_processing(self):
-        assert ClassificationJobStatus.PROCESSING == "processing"
-
-    def test_completed(self):
-        assert ClassificationJobStatus.COMPLETED == "completed"
-
-    def test_failed(self):
-        assert ClassificationJobStatus.FAILED == "failed"
-
-    def test_same_values_as_ingest(self):
-        """两种任务状态值相同."""
-        assert set(ClassificationJobStatus) == set(IngestJobStatus)
-
-
-class TestAssetClassification:
-    """AssetClassification 枚举测试."""
-
-    def test_nine_categories(self):
-        """9个分类."""
-        assert len(AssetClassification) == 9
-
-    def test_scenic(self):
-        assert AssetClassification.SCENIC == "scenic"
-
-    def test_product(self):
-        assert AssetClassification.PRODUCT == "product"
-
-    def test_person(self):
-        assert AssetClassification.PERSON == "person"
-
-    def test_animal(self):
-        assert AssetClassification.ANIMAL == "animal"
-
-    def test_food(self):
-        assert AssetClassification.FOOD == "food"
-
-    def test_tech(self):
-        assert AssetClassification.TECH == "tech"
-
-    def test_sport(self):
-        assert AssetClassification.SPORT == "sport"
-
-    def test_music(self):
-        assert AssetClassification.MUSIC == "music"
-
-    def test_other(self):
-        assert AssetClassification.OTHER == "other"
-
-    def test_all_values_unique(self):
-        """所有分类值唯一."""
-        values = [c.value for c in AssetClassification]
-        assert len(values) == len(set(values))
-
-
-class TestClassificationJobCreate:
-    """ClassificationJob.create 测试."""
-
-    def test_create_valid(self):
-        """正常创建."""
-        job = ClassificationJob.create(project_id="proj1", asset_id="asset1")
-        assert job.project_id == "proj1"
-        assert job.asset_id == "asset1"
-        assert job.status == ClassificationJobStatus.PENDING
-        assert job.classification == ""
-        assert job.confidence == 0.0
-        assert job.error_message == ""
-        assert isinstance(job.id, str)
-        assert len(job.id) > 0
-
-    def test_create_strips(self):
-        """project_id 和 asset_id 会 strip."""
-        job = ClassificationJob.create(project_id="  proj1  ", asset_id="  asset1  ")
-        assert job.project_id == "proj1"
-        assert job.asset_id == "asset1"
-
-    def test_create_empty_project_id(self):
-        """空 project_id 无效."""
-        try:
-            ClassificationJob.create(project_id="", asset_id="a1")
-            assert False
-        except ValueError as e:
-            assert "project_id" in str(e)
-
-    def test_create_whitespace_project_id(self):
-        """纯空白 project_id 无效."""
-        try:
-            ClassificationJob.create(project_id="   ", asset_id="a1")
-            assert False
-        except ValueError as e:
-            assert "project_id" in str(e)
-
-    def test_create_empty_asset_id(self):
-        """空 asset_id 无效."""
-        try:
-            ClassificationJob.create(project_id="p1", asset_id="")
-            assert False
-        except ValueError as e:
-            assert "asset_id" in str(e)
-
-    def test_create_whitespace_asset_id(self):
-        """纯空白 asset_id 无效."""
-        try:
-            ClassificationJob.create(project_id="p1", asset_id="   ")
-            assert False
-        except ValueError as e:
-            assert "asset_id" in str(e)
-
-    def test_create_unique_id(self):
-        """不同 job id 不同."""
-        j1 = ClassificationJob.create("p", "a")
-        j2 = ClassificationJob.create("p", "a")
-        assert j1.id != j2.id
-
-    def test_create_has_timestamps(self):
-        """有创建和更新时间."""
-        job = ClassificationJob.create("p", "a")
-        assert job.created_at is not None
-        assert job.updated_at is not None

tests/unit/domain/test_duplication.py

@@ -1,376 +0,0 @@
-"""duplication 单测.
-
-domain 层查重记录纯逻辑模块，0 外部依赖。
-覆盖：DuplicateSegment 工厂/校验、DuplicationRecord 创建/状态流转/重试。
-"""
-
-from __future__ import annotations
-
-from packages.domain.duplication import DuplicateSegment, DuplicationRecord
-
-
-class TestDuplicateSegmentCreate:
-    """DuplicateSegment.create 工厂方法测试."""
-
-    def test_create_valid(self):
-        """正常创建."""
-        seg = DuplicateSegment.create(
-            source_start=1.0,
-            source_end=5.0,
-            matched_video_id="vid123",
-            matched_video_name="测试视频",
-            matched_start=10.0,
-            matched_end=14.0,
-            similarity=85.5,
-        )
-        assert seg.source_start == 1.0
-        assert seg.source_end == 5.0
-        assert seg.matched_video_id == "vid123"
-        assert seg.matched_video_name == "测试视频"
-        assert seg.matched_start == 10.0
-        assert seg.matched_end == 14.0
-        assert seg.similarity == 85.5
-        assert isinstance(seg.id, str)
-        assert len(seg.id) > 0
-
-    def test_create_generates_unique_id(self):
-        """每次创建生成不同的 id."""
-        seg1 = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 50.0)
-        seg2 = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 50.0)
-        assert seg1.id != seg2.id
-
-    def test_create_negative_source_start(self):
-        """source_start 为负抛出 ValueError."""
-        try:
-            DuplicateSegment.create(-1, 5, "v", "n", 0, 1, 50.0)
-            assert False, "应该抛出 ValueError"
-        except ValueError as e:
-            assert "source" in str(e).lower()
-
-    def test_create_source_end_equals_start(self):
-        """source_end 等于 source_start 无效."""
-        try:
-            DuplicateSegment.create(5, 5, "v", "n", 0, 1, 50.0)
-            assert False
-        except ValueError as e:
-            assert "source" in str(e).lower()
-
-    def test_create_source_end_less_than_start(self):
-        """source_end 小于 source_start 无效."""
-        try:
-            DuplicateSegment.create(5, 3, "v", "n", 0, 1, 50.0)
-            assert False
-        except ValueError as e:
-            assert "source" in str(e).lower()
-
-    def test_create_negative_matched_start(self):
-        """matched_start 为负无效."""
-        try:
-            DuplicateSegment.create(0, 5, "v", "n", -1, 1, 50.0)
-            assert False
-        except ValueError as e:
-            assert "matched" in str(e).lower()
-
-    def test_create_matched_end_invalid(self):
-        """matched_end <= matched_start 无效."""
-        try:
-            DuplicateSegment.create(0, 5, "v", "n", 5, 5, 50.0)
-            assert False
-        except ValueError as e:
-            assert "matched" in str(e).lower()
-
-    def test_create_similarity_zero(self):
-        """similarity = 0 是合法的."""
-        seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 0.0)
-        assert seg.similarity == 0.0
-
-    def test_create_similarity_100(self):
-        """similarity = 100 是合法的."""
-        seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 100.0)
-        assert seg.similarity == 100.0
-
-    def test_create_similarity_negative(self):
-        """similarity < 0 无效."""
-        try:
-            DuplicateSegment.create(0, 1, "v", "n", 0, 1, -1.0)
-            assert False
-        except ValueError as e:
-            assert "similarity" in str(e).lower()
-
-    def test_create_similarity_over_100(self):
-        """similarity > 100 无效."""
-        try:
-            DuplicateSegment.create(0, 1, "v", "n", 0, 1, 101.0)
-            assert False
-        except ValueError as e:
-            assert "similarity" in str(e).lower()
-
-
-class TestDuplicationRecordCreate:
-    """DuplicationRecord.create 工厂方法测试."""
-
-    def test_create_minimal(self):
-        """最简创建."""
-        rec = DuplicationRecord.create(
-            user_id="user1",
-            filename="test.mp4",
-            file_size=1024,
-            storage_key="oss://bucket/test.mp4",
-        )
-        assert rec.user_id == "user1"
-        assert rec.filename == "test.mp4"
-        assert rec.file_size == 1024
-        assert rec.storage_key == "oss://bucket/test.mp4"
-        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 == ""
-        assert isinstance(rec.id, str)
-        assert len(rec.id) > 0
-
-    def test_create_with_duration(self):
-        """带时长创建."""
-        rec = DuplicationRecord.create(
-            user_id="user1",
-            filename="test.mp4",
-            file_size=1024,
-            storage_key="oss://key",
-            duration_seconds=120.5,
-        )
-        assert rec.duration_seconds == 120.5
-
-    def test_create_strips_whitespace(self):
-        """user_id 和 filename 会 strip."""
-        rec = DuplicationRecord.create(
-            user_id="  user1  ",
-            filename="  test.mp4  ",
-            file_size=1024,
-            storage_key="oss://key",
-        )
-        assert rec.user_id == "user1"
-        assert rec.filename == "test.mp4"
-
-    def test_create_empty_user_id(self):
-        """空 user_id 无效."""
-        try:
-            DuplicationRecord.create("", "test.mp4", 1024, "oss://key")
-            assert False
-        except ValueError as e:
-            assert "user_id" in str(e)
-
-    def test_create_whitespace_user_id(self):
-        """纯空白 user_id 无效."""
-        try:
-            DuplicationRecord.create("   ", "test.mp4", 1024, "oss://key")
-            assert False
-        except ValueError as e:
-            assert "user_id" in str(e)
-
-    def test_create_empty_filename(self):
-        """空 filename 无效."""
-        try:
-            DuplicationRecord.create("user1", "", 1024, "oss://key")
-            assert False
-        except ValueError as e:
-            assert "filename" in str(e)
-
-    def test_create_whitespace_filename(self):
-        """纯空白 filename 无效."""
-        try:
-            DuplicationRecord.create("user1", "   ", 1024, "oss://key")
-            assert False
-        except ValueError as e:
-            assert "filename" in str(e)
-
-    def test_create_zero_file_size(self):
-        """file_size = 0 无效."""
-        try:
-            DuplicationRecord.create("user1", "test.mp4", 0, "oss://key")
-            assert False
-        except ValueError as e:
-            assert "file_size" in str(e)
-
-    def test_create_negative_file_size(self):
-        """file_size 为负无效."""
-        try:
-            DuplicationRecord.create("user1", "test.mp4", -1, "oss://key")
-            assert False
-        except ValueError as e:
-            assert "file_size" in str(e)
-
-    def test_create_unique_id(self):
-        """不同记录 id 不同."""
-        r1 = DuplicationRecord.create("u", "f", 1, "k")
-        r2 = DuplicationRecord.create("u", "f", 1, "k")
-        assert r1.id != r2.id
-
-    def test_create_has_timestamps(self):
-        """有创建和更新时间."""
-        rec = DuplicationRecord.create("u", "f", 1, "k")
-        assert rec.created_at is not None
-        assert rec.updated_at is not None
-        # 两者应该很接近（都是 now）
-        delta = (rec.updated_at - rec.created_at).total_seconds()
-        assert abs(delta) < 1.0
-
-
-class TestDuplicationRecordStatusFlow:
-    """状态流转测试."""
-
-    def _make_record(self):
-        return DuplicationRecord.create("user1", "test.mp4", 1024, "oss://key")
-
-    def test_initial_status_pending(self):
-        """初始状态 pending."""
-        rec = self._make_record()
-        assert rec.status == "pending"
-
-    def test_mark_processing(self):
-        """标记为处理中."""
-        rec = self._make_record()
-        old_updated = rec.updated_at
-        rec.mark_processing()
-        assert rec.status == "processing"
-        assert rec.updated_at >= old_updated
-
-    def test_mark_completed(self):
-        """标记为完成."""
-        rec = self._make_record()
-        seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 80.0)
-        rec.mark_completed(duplicate_rate=45.5, duplicate_count=3, segments=[seg])
-        assert rec.status == "completed"
-        assert rec.duplicate_rate == 45.5
-        assert rec.duplicate_count == 3
-        assert len(rec.segments) == 1
-        assert rec.segments[0].similarity == 80.0
-
-    def test_mark_completed_zero_rate(self):
-        """重复率为 0 合法."""
-        rec = self._make_record()
-        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):
-        """重复率 100 合法."""
-        rec = self._make_record()
-        rec.mark_completed(100.0, 1, [])
-        assert rec.duplicate_rate == 100.0
-
-    def test_mark_completed_negative_rate(self):
-        """重复率为负无效."""
-        rec = self._make_record()
-        try:
-            rec.mark_completed(-1, 0, [])
-            assert False
-        except ValueError as e:
-            assert "duplicate_rate" in str(e)
-
-    def test_mark_completed_over_100(self):
-        """重复率超过 100 无效."""
-        rec = self._make_record()
-        try:
-            rec.mark_completed(101, 0, [])
-            assert False
-        except ValueError as e:
-            assert "duplicate_rate" in str(e)
-
-    def test_mark_failed(self):
-        """标记为失败."""
-        rec = self._make_record()
-        rec.mark_failed("网络超时")
-        assert rec.status == "failed"
-        assert rec.error_message == "网络超时"
-
-    def test_mark_failed_empty_message(self):
-        """失败信息可以为空字符串."""
-        rec = self._make_record()
-        rec.mark_failed("")
-        assert rec.status == "failed"
-        assert rec.error_message == ""
-
-    def test_can_retry_failed(self):
-        """failed 状态可以重试."""
-        rec = self._make_record()
-        rec.mark_failed("error")
-        assert rec.can_retry() is True
-
-    def test_cannot_retry_pending(self):
-        """pending 状态不可重试."""
-        rec = self._make_record()
-        assert rec.can_retry() is False
-
-    def test_cannot_retry_processing(self):
-        """processing 状态不可重试."""
-        rec = self._make_record()
-        rec.mark_processing()
-        assert rec.can_retry() is False
-
-    def test_cannot_retry_completed(self):
-        """completed 状态不可重试."""
-        rec = self._make_record()
-        rec.mark_completed(50, 1, [])
-        assert rec.can_retry() is False
-
-    def test_reset_for_retry(self):
-        """重置回 pending."""
-        rec = self._make_record()
-        rec.mark_failed("error")
-        seg = DuplicateSegment.create(0, 1, "v", "n", 0, 1, 50.0)
-        rec.segments = [seg]
-        rec.video_fingerprint = {"hash": "abc"}
-        rec.duplicate_rate = 50.0
-        rec.duplicate_count = 5
-
-        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_updates_timestamp(self):
-        """重置更新 updated_at."""
-        rec = self._make_record()
-        rec.mark_failed("error")
-        old_updated = rec.updated_at
-        rec.reset_for_retry()
-        assert rec.updated_at >= old_updated
-
-
-class TestDuplicationRecordSegments:
-    """segments 列表相关测试."""
-
-    def _make_record(self):
-        return DuplicationRecord.create("user1", "test.mp4", 1024, "oss://key")
-
-    def test_segments_default_empty(self):
-        """初始 segments 为空列表."""
-        rec = self._make_record()
-        assert rec.segments == []
-
-    def test_segments_independent_list(self):
-        """不同记录的 segments 是独立列表."""
-        r1 = self._make_record()
-        r2 = self._make_record()
-        r1.segments.append("fake")
-        assert len(r2.segments) == 0
-
-    def test_completed_with_multiple_segments(self):
-        """完成时带多个片段."""
-        rec = self._make_record()
-        segs = [
-            DuplicateSegment.create(0, 1, "v1", "n1", 0, 1, 90.0),
-            DuplicateSegment.create(2, 3, "v2", "n2", 5, 6, 70.0),
-            DuplicateSegment.create(4, 5, "v3", "n3", 10, 11, 85.0),
-        ]
-        rec.mark_completed(60.0, 3, segs)
-        assert len(rec.segments) == 3
-        assert rec.segments[0].similarity == 90.0
-        assert rec.segments[1].matched_video_id == "v2"
-        assert rec.segments[2].matched_video_name == "n3"

tests/unit/domain/test_edit_plan_clip.py

@@ -1,362 +0,0 @@
-"""edit_plan_clip 单测.
-
-domain 层剪辑计划片段纯逻辑模块，0 外部依赖。
-覆盖：枚举常量、create工厂/校验、素材分配、状态流转、属性计算。
-"""
-
-from __future__ import annotations
-
-from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
-
-
-class TestEditPlanClipStatus:
-    """EditPlanClipStatus 枚举测试."""
-
-    def test_four_statuses(self):
-        """四种状态."""
-        assert len(EditPlanClipStatus) == 4
-
-    def test_pending(self):
-        """pending 状态."""
-        assert EditPlanClipStatus.PENDING == "pending"
-
-    def test_ready(self):
-        """ready 状态."""
-        assert EditPlanClipStatus.READY == "ready"
-
-    def test_rendered(self):
-        """rendered 状态."""
-        assert EditPlanClipStatus.RENDERED == "rendered"
-
-    def test_failed(self):
-        """failed 状态."""
-        assert EditPlanClipStatus.FAILED == "failed"
-
-    def test_is_string(self):
-        """枚举值是字符串."""
-        for status in EditPlanClipStatus:
-            assert isinstance(status.value, str)
-            assert len(status.value) > 0
-
-    def test_str_compatible(self):
-        """StrEnum 可与字符串比较."""
-        assert EditPlanClipStatus.PENDING == "pending"
-        assert EditPlanClipStatus.READY + "" == "ready"
-
-
-class TestEditPlanClipCreate:
-    """EditPlanClip.create 工厂方法测试."""
-
-    def test_create_minimal(self):
-        """最简创建."""
-        clip = EditPlanClip.create(plan_id="plan1", clip_type="video", order=0)
-        assert clip.plan_id == "plan1"
-        assert clip.clip_type == "video"
-        assert clip.order == 0
-        assert clip.status == EditPlanClipStatus.PENDING
-        assert clip.template_clip_config_id == ""
-        assert clip.asset_id == ""
-        assert clip.text_content == ""
-        assert clip.start_time == 0.0
-        assert clip.duration == 0.0
-        assert clip.transition_effect == "cut"
-        assert clip.config == {}
-        assert isinstance(clip.id, str)
-        assert len(clip.id) > 0
-
-    def test_create_with_all_fields(self):
-        """带全部字段创建."""
-        clip = EditPlanClip.create(
-            plan_id="plan1",
-            clip_type="video",
-            order=2,
-            template_clip_config_id="tpl1",
-            asset_id="asset1",
-            text_content="  你好世界  ",
-            start_time=10.5,
-            duration=5.0,
-            transition_effect="fade",
-            config={"key": "value"},
-        )
-        assert clip.order == 2
-        assert clip.template_clip_config_id == "tpl1"
-        assert clip.asset_id == "asset1"
-        assert clip.text_content == "你好世界"  # strip了
-        assert clip.start_time == 10.5
-        assert clip.duration == 5.0
-        assert clip.transition_effect == "fade"
-        assert clip.config == {"key": "value"}
-
-    def test_create_strips_ids(self):
-        """plan_id 和 clip_type 会 strip."""
-        clip = EditPlanClip.create(plan_id="  plan1  ", clip_type="  video  ", order=0)
-        assert clip.plan_id == "plan1"
-        assert clip.clip_type == "video"
-
-    def test_create_empty_plan_id(self):
-        """空 plan_id 无效."""
-        try:
-            EditPlanClip.create(plan_id="", clip_type="video", order=0)
-            assert False
-        except ValueError as e:
-            assert "plan_id" in str(e)
-
-    def test_create_whitespace_plan_id(self):
-        """纯空白 plan_id 无效."""
-        try:
-            EditPlanClip.create(plan_id="   ", clip_type="video", order=0)
-            assert False
-        except ValueError as e:
-            assert "plan_id" in str(e)
-
-    def test_create_empty_clip_type(self):
-        """空 clip_type 无效."""
-        try:
-            EditPlanClip.create(plan_id="p1", clip_type="", order=0)
-            assert False
-        except ValueError as e:
-            assert "clip_type" in str(e)
-
-    def test_create_whitespace_clip_type(self):
-        """纯空白 clip_type 无效."""
-        try:
-            EditPlanClip.create(plan_id="p1", clip_type="   ", order=0)
-            assert False
-        except ValueError as e:
-            assert "clip_type" in str(e)
-
-    def test_create_negative_start_time(self):
-        """start_time 为负无效."""
-        try:
-            EditPlanClip.create(plan_id="p1", clip_type="v", order=0, start_time=-1.0)
-            assert False
-        except ValueError as e:
-            assert "start_time" in str(e)
-
-    def test_create_negative_duration(self):
-        """duration 为负无效."""
-        try:
-            EditPlanClip.create(plan_id="p1", clip_type="v", order=0, duration=-1.0)
-            assert False
-        except ValueError as e:
-            assert "duration" in str(e)
-
-    def test_create_zero_duration_valid(self):
-        """duration 为 0 合法."""
-        clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, duration=0.0)
-        assert clip.duration == 0.0
-
-    def test_create_empty_transition_defaults_to_cut(self):
-        """空 transition_effect 默认 cut."""
-        clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, transition_effect="")
-        assert clip.transition_effect == "cut"
-
-    def test_create_whitespace_transition_defaults_to_cut(self):
-        """空白 transition_effect 默认 cut."""
-        clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, transition_effect="   ")
-        assert clip.transition_effect == "cut"
-
-    def test_create_config_none_defaults_empty_dict(self):
-        """config=None 默认为空 dict."""
-        clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, config=None)
-        assert clip.config == {}
-
-    def test_create_unique_id(self):
-        """不同 clip id 不同."""
-        c1 = EditPlanClip.create("p1", "v", 0)
-        c2 = EditPlanClip.create("p1", "v", 0)
-        assert c1.id != c2.id
-
-    def test_create_has_timestamps(self):
-        """有创建和更新时间."""
-        clip = EditPlanClip.create("p1", "v", 0)
-        assert clip.created_at is not None
-        assert clip.updated_at is not None
-
-    def test_create_negative_order_valid(self):
-        """order 可以为负（表示排序位置）."""
-        clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=-1)
-        assert clip.order == -1
-
-
-class TestEditPlanClipAssignAsset:
-    """素材分配测试."""
-
-    def _make_clip(self):
-        return EditPlanClip.create(plan_id="p1", clip_type="video", order=0)
-
-    def test_assign_asset(self):
-        """正常分配素材."""
-        clip = self._make_clip()
-        old_updated = clip.updated_at
-        clip.assign_asset("asset123")
-        assert clip.asset_id == "asset123"
-        assert clip.updated_at >= old_updated
-
-    def test_assign_asset_strips(self):
-        """asset_id 会 strip."""
-        clip = self._make_clip()
-        clip.assign_asset("  asset123  ")
-        assert clip.asset_id == "asset123"
-
-    def test_assign_asset_empty(self):
-        """空 asset_id 无效."""
-        clip = self._make_clip()
-        try:
-            clip.assign_asset("")
-            assert False
-        except ValueError as e:
-            assert "asset_id" in str(e)
-
-    def test_assign_asset_whitespace(self):
-        """纯空白 asset_id 无效."""
-        clip = self._make_clip()
-        try:
-            clip.assign_asset("   ")
-            assert False
-        except ValueError as e:
-            assert "asset_id" in str(e)
-
-    def test_has_asset_false_initially(self):
-        """初始无素材."""
-        clip = self._make_clip()
-        assert clip.has_asset is False
-
-    def test_has_asset_true_after_assign(self):
-        """分配后有素材."""
-        clip = self._make_clip()
-        clip.assign_asset("a1")
-        assert clip.has_asset is True
-
-
-class TestEditPlanClipStatusFlow:
-    """状态流转测试."""
-
-    def _make_clip(self):
-        return EditPlanClip.create(plan_id="p1", clip_type="video", order=0)
-
-    def test_initial_status_pending(self):
-        """初始状态 pending."""
-        clip = self._make_clip()
-        assert clip.status == EditPlanClipStatus.PENDING
-
-    def test_pending_to_ready(self):
-        """pending -> ready."""
-        clip = self._make_clip()
-        clip.mark_ready()
-        assert clip.status == EditPlanClipStatus.READY
-
-    def test_ready_to_rendered(self):
-        """ready -> rendered."""
-        clip = self._make_clip()
-        clip.mark_ready()
-        clip.mark_rendered()
-        assert clip.status == EditPlanClipStatus.RENDERED
-
-    def test_ready_to_failed(self):
-        """ready -> failed."""
-        clip = self._make_clip()
-        clip.mark_ready()
-        clip.mark_failed()
-        assert clip.status == EditPlanClipStatus.FAILED
-
-    def test_cannot_ready_from_rendered(self):
-        """rendered 状态不能再 mark_ready."""
-        clip = self._make_clip()
-        clip.mark_ready()
-        clip.mark_rendered()
-        try:
-            clip.mark_ready()
-            assert False
-        except ValueError as e:
-            assert "pending" in str(e).lower()
-
-    def test_cannot_ready_from_failed(self):
-        """failed 状态不能 mark_ready."""
-        clip = self._make_clip()
-        clip.mark_ready()
-        clip.mark_failed()
-        try:
-            clip.mark_ready()
-            assert False
-        except ValueError as e:
-            assert "pending" in str(e).lower()
-
-    def test_cannot_render_from_pending(self):
-        """pending 不能直接 mark_rendered."""
-        clip = self._make_clip()
-        try:
-            clip.mark_rendered()
-            assert False
-        except ValueError as e:
-            assert "ready" in str(e).lower()
-
-    def test_cannot_failed_from_pending(self):
-        """pending 不能直接 mark_failed."""
-        clip = self._make_clip()
-        try:
-            clip.mark_failed()
-            assert False
-        except ValueError as e:
-            assert "ready" in str(e).lower()
-
-    def test_status_change_updates_timestamp(self):
-        """状态变更更新 updated_at."""
-        clip = self._make_clip()
-        old_updated = clip.updated_at
-        clip.mark_ready()
-        assert clip.updated_at >= old_updated
-
-
-class TestEditPlanClipProperties:
-    """属性计算测试."""
-
-    def test_end_time(self):
-        """end_time = start_time + duration."""
-        clip = EditPlanClip.create(
-            plan_id="p1",
-            clip_type="v",
-            order=0,
-            start_time=10.0,
-            duration=5.5,
-        )
-        assert clip.end_time == 15.5
-
-    def test_end_time_zero_duration(self):
-        """零时长 end_time = start_time."""
-        clip = EditPlanClip.create(
-            plan_id="p1",
-            clip_type="v",
-            order=0,
-            start_time=10.0,
-            duration=0.0,
-        )
-        assert clip.end_time == 10.0
-
-    def test_end_time_zero_start(self):
-        """零起点 end_time = duration."""
-        clip = EditPlanClip.create(
-            plan_id="p1",
-            clip_type="v",
-            order=0,
-            start_time=0.0,
-            duration=7.0,
-        )
-        assert clip.end_time == 7.0
-
-    def test_has_asset_empty_string(self):
-        """空字符串无素材."""
-        clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, asset_id="")
-        assert clip.has_asset is False
-
-    def test_has_asset_with_value(self):
-        """有值则有素材."""
-        clip = EditPlanClip.create(plan_id="p1", clip_type="v", order=0, asset_id="a1")
-        assert clip.has_asset is True
-
-    def test_config_independent_between_clips(self):
-        """不同 clip 的 config 独立."""
-        c1 = EditPlanClip.create(plan_id="p1", clip_type="v", order=0)
-        c2 = EditPlanClip.create(plan_id="p1", clip_type="v", order=1)
-        c1.config["key"] = "value"
-        assert "key" not in c2.config

tests/unit/domain/test_quota.py

@@ -1,575 +0,0 @@
-"""quota 单测.
-
-domain 层配额系统纯逻辑模块，0 外部依赖。
-覆盖：枚举常量、QuotaTier、QUOTA_TIERS常量、QuotaWarningLevel、
-QuotaCheckResult、QuotaRegistry注册/查询、QuotaChecker检查/告警级别。
-"""
-
-from __future__ import annotations
-
-import math
-
-from packages.domain.quota import (
-    QUOTA_TIERS,
-    QuotaCheckResult,
-    QuotaChecker,
-    QuotaDimension,
-    QuotaRegistry,
-    QuotaTier,
-    QuotaWarningLevel,
-    get_warning_level,
-    quota_checker,
-    quota_registry,
-)
-
-
-class TestQuotaDimension:
-    """QuotaDimension 枚举测试."""
-
-    def test_member_count(self):
-        """内置维度数量."""
-        assert len(QuotaDimension) == 11
-
-    def test_storage_gb(self):
-        """存储维度值."""
-        assert QuotaDimension.STORAGE_GB.value == "storage_gb"
-
-    def test_videos_per_month(self):
-        """每月视频数维度."""
-        assert QuotaDimension.VIDEOS_PER_MONTH.value == "videos_per_month"
-
-    def test_max_concurrent(self):
-        """并发数维度."""
-        assert QuotaDimension.MAX_CONCURRENT.value == "max_concurrent"
-
-    def test_max_templates(self):
-        """模板数维度."""
-        assert QuotaDimension.MAX_TEMPLATES.value == "max_templates"
-
-    def test_max_titles(self):
-        """标题库维度."""
-        assert QuotaDimension.MAX_TITLES.value == "max_titles"
-
-    def test_max_voiceovers(self):
-        """配音库维度."""
-        assert QuotaDimension.MAX_VOICEOVERS.value == "max_voiceovers"
-
-    def test_ai_voice_enabled(self):
-        """AI配音开关维度."""
-        assert QuotaDimension.AI_VOICE_ENABLED.value == "ai_voice_enabled"
-
-    def test_ai_voice_credits(self):
-        """AI配音积分维度."""
-        assert QuotaDimension.AI_VOICE_CREDITS.value == "ai_voice_credits"
-
-    def test_all_values_are_strings(self):
-        """所有枚举值都是字符串."""
-        for dim in QuotaDimension:
-            assert isinstance(dim.value, str)
-            assert len(dim.value) > 0
-
-
-class TestQuotaTier:
-    """QuotaTier 数据类测试."""
-
-    def test_create_empty(self):
-        """创建空配额等级."""
-        tier = QuotaTier(name="test")
-        assert tier.name == "test"
-        assert tier.limits == {}
-
-    def test_create_with_limits(self):
-        """创建带限制的配额等级."""
-        tier = QuotaTier(name="pro", limits={"storage": 100, "videos": 50})
-        assert tier.name == "pro"
-        assert tier.get_limit("storage") == 100
-        assert tier.get_limit("videos") == 50
-
-    def test_get_limit_undefined_returns_zero(self):
-        """未定义的维度返回 0."""
-        tier = QuotaTier(name="test")
-        assert tier.get_limit("nonexistent") == 0
-
-    def test_is_unlimited_inf(self):
-        """inf 视为不限量."""
-        tier = QuotaTier(name="test", limits={"templates": float("inf")})
-        assert tier.is_unlimited("templates") is True
-
-    def test_is_unlimited_finite(self):
-        """有限值不是不限量."""
-        tier = QuotaTier(name="test", limits={"storage": 100})
-        assert tier.is_unlimited("storage") is False
-
-    def test_is_unlimited_undefined(self):
-        """未定义的维度默认不限量."""
-        tier = QuotaTier(name="test")
-        # 未定义的 key 取默认值 inf，因此 is_unlimited 应该返回 True
-        assert tier.is_unlimited("unknown") is True
-
-
-class TestQuotaTiers:
-    """QUOTA_TIERS 常量测试."""
-
-    def test_three_tiers_exist(self):
-        """三个套餐等级都存在."""
-        assert "free" in QUOTA_TIERS
-        assert "basic" in QUOTA_TIERS
-        assert "premium" in QUOTA_TIERS
-
-    def test_free_storage(self):
-        """免费版 2GB 存储."""
-        assert QUOTA_TIERS["free"].get_limit(QuotaDimension.STORAGE_GB) == 2
-
-    def test_free_videos_per_month(self):
-        """免费版 5个视频/月."""
-        assert QUOTA_TIERS["free"].get_limit(QuotaDimension.VIDEOS_PER_MONTH) == 5
-
-    def test_free_ai_voice_disabled(self):
-        """免费版无AI配音."""
-        assert QUOTA_TIERS["free"].get_limit(QuotaDimension.AI_VOICE_ENABLED) == 0
-
-    def test_basic_storage(self):
-        """基础版 20GB 存储."""
-        assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.STORAGE_GB) == 20
-
-    def test_basic_videos(self):
-        """基础版 30视频/月."""
-        assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.VIDEOS_PER_MONTH) == 30
-
-    def test_basic_ai_voice_enabled(self):
-        """基础版有AI配音."""
-        assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.AI_VOICE_ENABLED) == 1
-
-    def test_basic_ai_voice_credits(self):
-        """基础版 100 AI配音积分."""
-        assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.AI_VOICE_CREDITS) == 100
-
-    def test_premium_storage(self):
-        """高级版 100GB 存储."""
-        assert QUOTA_TIERS["premium"].get_limit(QuotaDimension.STORAGE_GB) == 100
-
-    def test_premium_videos(self):
-        """高级版 100视频/月."""
-        assert QUOTA_TIERS["premium"].get_limit(QuotaDimension.VIDEOS_PER_MONTH) == 100
-
-    def test_premium_templates_unlimited(self):
-        """高级版模板不限量."""
-        assert QUOTA_TIERS["premium"].is_unlimited(QuotaDimension.MAX_TEMPLATES)
-
-    def test_premium_multi_platform_enabled(self):
-        """高级版多平台发布."""
-        assert QUOTA_TIERS["premium"].get_limit(QuotaDimension.MULTI_PLATFORM_ENABLED) == 1
-
-    def test_storage_monotonic(self):
-        """存储量随套餐升级单调递增."""
-        free = QUOTA_TIERS["free"].get_limit(QuotaDimension.STORAGE_GB)
-        basic = QUOTA_TIERS["basic"].get_limit(QuotaDimension.STORAGE_GB)
-        premium = QUOTA_TIERS["premium"].get_limit(QuotaDimension.STORAGE_GB)
-        assert free < basic < premium
-
-    def test_videos_monotonic(self):
-        """视频数随套餐升级单调递增."""
-        free = QUOTA_TIERS["free"].get_limit(QuotaDimension.VIDEOS_PER_MONTH)
-        basic = QUOTA_TIERS["basic"].get_limit(QuotaDimension.VIDEOS_PER_MONTH)
-        premium = QUOTA_TIERS["premium"].get_limit(QuotaDimension.VIDEOS_PER_MONTH)
-        assert free < basic < premium
-
-
-class TestQuotaWarningLevel:
-    """告警级别常量测试."""
-
-    def test_levels_defined(self):
-        """四个级别都有定义."""
-        assert QuotaWarningLevel.NORMAL == "normal"
-        assert QuotaWarningLevel.WARNING == "warning"
-        assert QuotaWarningLevel.CRITICAL == "critical"
-        assert QuotaWarningLevel.EXCEEDED == "exceeded"
-
-    def test_four_distinct_levels(self):
-        """四个级别各不相同."""
-        levels = {
-            QuotaWarningLevel.NORMAL,
-            QuotaWarningLevel.WARNING,
-            QuotaWarningLevel.CRITICAL,
-            QuotaWarningLevel.EXCEEDED,
-        }
-        assert len(levels) == 4
-
-
-class TestQuotaCheckResult:
-    """QuotaCheckResult 测试."""
-
-    def test_usage_percent_normal(self):
-        """正常使用百分比."""
-        result = QuotaCheckResult(
-            allowed=True,
-            dimension="storage",
-            limit=100,
-            used=30,
-            remaining=70,
-            warning_level=QuotaWarningLevel.NORMAL,
-        )
-        assert result.usage_percent == 30.0
-
-    def test_usage_percent_zero_used(self):
-        """使用量为 0."""
-        result = QuotaCheckResult(
-            allowed=True,
-            dimension="storage",
-            limit=100,
-            used=0,
-            remaining=100,
-            warning_level=QuotaWarningLevel.NORMAL,
-        )
-        assert result.usage_percent == 0.0
-
-    def test_usage_percent_exactly_100(self):
-        """刚好用完."""
-        result = QuotaCheckResult(
-            allowed=False,
-            dimension="storage",
-            limit=100,
-            used=100,
-            remaining=0,
-            warning_level=QuotaWarningLevel.EXCEEDED,
-        )
-        assert result.usage_percent == 100.0
-
-    def test_usage_percent_over_limit_capped(self):
-        """超出限制时封顶 100%."""
-        result = QuotaCheckResult(
-            allowed=False,
-            dimension="storage",
-            limit=100,
-            used=150,
-            remaining=0,
-            warning_level=QuotaWarningLevel.EXCEEDED,
-        )
-        assert result.usage_percent == 100.0
-
-    def test_usage_percent_unlimited(self):
-        """不限量时使用率为 0."""
-        result = QuotaCheckResult(
-            allowed=True,
-            dimension="templates",
-            limit=float("inf"),
-            used=1000,
-            remaining=float("inf"),
-            warning_level=QuotaWarningLevel.NORMAL,
-        )
-        assert result.usage_percent == 0.0
-
-    def test_usage_percent_zero_limit_with_usage(self):
-        """限制为 0 但有使用量，返回 100%."""
-        result = QuotaCheckResult(
-            allowed=False,
-            dimension="ai_voice",
-            limit=0,
-            used=1,
-            remaining=0,
-            warning_level=QuotaWarningLevel.EXCEEDED,
-        )
-        assert result.usage_percent == 100.0
-
-    def test_usage_percent_zero_limit_no_usage(self):
-        """限制为 0 且无使用量，返回 0%."""
-        result = QuotaCheckResult(
-            allowed=True,
-            dimension="ai_voice",
-            limit=0,
-            used=0,
-            remaining=0,
-            warning_level=QuotaWarningLevel.NORMAL,
-        )
-        assert result.usage_percent == 0.0
-
-
-class TestGetWarningLevel:
-    """get_warning_level 便捷函数测试."""
-
-    def test_zero_usage(self):
-        """0% 使用 - normal."""
-        assert get_warning_level(0, 100) == QuotaWarningLevel.NORMAL
-
-    def test_below_80_percent(self):
-        """低于80% - normal."""
-        assert get_warning_level(50, 100) == QuotaWarningLevel.NORMAL
-        assert get_warning_level(79, 100) == QuotaWarningLevel.NORMAL
-
-    def test_at_80_percent(self):
-        """刚好80% - warning."""
-        assert get_warning_level(80, 100) == QuotaWarningLevel.WARNING
-
-    def test_between_80_and_95(self):
-        """80%-95%之间 - warning."""
-        assert get_warning_level(90, 100) == QuotaWarningLevel.WARNING
-
-    def test_at_95_percent(self):
-        """刚好95% - critical."""
-        assert get_warning_level(95, 100) == QuotaWarningLevel.CRITICAL
-
-    def test_between_95_and_100(self):
-        """95%-100%之间 - critical."""
-        assert get_warning_level(99, 100) == QuotaWarningLevel.CRITICAL
-
-    def test_at_100_percent(self):
-        """刚好100% - exceeded."""
-        assert get_warning_level(100, 100) == QuotaWarningLevel.EXCEEDED
-
-    def test_over_100_percent(self):
-        """超过100% - exceeded."""
-        assert get_warning_level(150, 100) == QuotaWarningLevel.EXCEEDED
-
-    def test_zero_limit_with_usage(self):
-        """限制为0但有使用 - exceeded."""
-        assert get_warning_level(1, 0) == QuotaWarningLevel.EXCEEDED
-
-    def test_zero_limit_no_usage(self):
-        """限制为0且无使用 - normal."""
-        assert get_warning_level(0, 0) == QuotaWarningLevel.NORMAL
-
-    def test_unlimited(self):
-        """不限量 - 始终 normal."""
-        assert get_warning_level(0, float("inf")) == QuotaWarningLevel.NORMAL
-        assert get_warning_level(9999, float("inf")) == QuotaWarningLevel.NORMAL
-
-    def test_negative_usage(self):
-        """负使用量 - normal."""
-        assert get_warning_level(-10, 100) == QuotaWarningLevel.NORMAL
-
-
-class TestQuotaRegistry:
-    """QuotaRegistry 测试."""
-
-    def test_initial_dimensions(self):
-        """初始化后内置维度都在."""
-        reg = QuotaRegistry()
-        dims = reg.list_dimensions()
-        for dim in QuotaDimension:
-            assert dim.value in dims
-
-    def test_initial_dimensions_count(self):
-        """初始维度数量等于枚举数量."""
-        reg = QuotaRegistry()
-        assert len(reg.list_dimensions()) == len(QuotaDimension)
-
-    def test_list_tiers(self):
-        """三个套餐等级."""
-        reg = QuotaRegistry()
-        tiers = reg.list_tiers()
-        assert "free" in tiers
-        assert "basic" in tiers
-        assert "premium" in tiers
-        assert len(tiers) == 3
-
-    def test_get_tier_existing(self):
-        """获取已有的套餐."""
-        reg = QuotaRegistry()
-        tier = reg.get_tier("free")
-        assert tier is not None
-        assert tier.name == "free"
-
-    def test_get_tier_nonexistent(self):
-        """获取不存在的套餐返回 None."""
-        reg = QuotaRegistry()
-        assert reg.get_tier("enterprise") is None
-
-    def test_get_limit_existing(self):
-        """获取已有限制."""
-        reg = QuotaRegistry()
-        assert reg.get_limit("free", QuotaDimension.STORAGE_GB) == 2
-
-    def test_get_limit_nonexistent_plan(self):
-        """不存在的套餐返回 0."""
-        reg = QuotaRegistry()
-        assert reg.get_limit("enterprise", QuotaDimension.STORAGE_GB) == 0
-
-    def test_get_limit_unknown_dimension(self):
-        """未知维度返回 0."""
-        reg = QuotaRegistry()
-        assert reg.get_limit("free", "unknown_dim") == 0
-
-    def test_register_dimension_new(self):
-        """注册新维度."""
-        reg = QuotaRegistry()
-        count_before = len(reg.list_dimensions())
-        reg.register_dimension("custom_dim", "自定义维度")
-        dims = reg.list_dimensions()
-        assert "custom_dim" in dims
-        assert len(dims) == count_before + 1
-
-    def test_register_dimension_with_default_limits(self):
-        """注册带默认限制的维度."""
-        reg = QuotaRegistry()
-        reg.register_dimension(
-            "api_calls",
-            "API调用次数",
-            default_limits={"free": 100, "basic": 1000, "premium": 10000},
-        )
-        assert reg.get_limit("free", "api_calls") == 100
-        assert reg.get_limit("basic", "api_calls") == 1000
-        assert reg.get_limit("premium", "api_calls") == 10000
-
-    def test_register_dimension_default_zero(self):
-        """不带默认限制的维度，各套餐默认为 0."""
-        reg = QuotaRegistry()
-        reg.register_dimension("beta_feature", "测试功能")
-        assert reg.get_limit("free", "beta_feature") == 0
-        assert reg.get_limit("basic", "beta_feature") == 0
-        assert reg.get_limit("premium", "beta_feature") == 0
-
-    def test_register_dimension_idempotent(self):
-        """重复注册幂等."""
-        reg = QuotaRegistry()
-        reg.register_dimension("dup", "重复测试", default_limits={"free": 10})
-        count_before = len(reg.list_dimensions())
-        # 第二次注册不同的限制，应该不生效
-        reg.register_dimension("dup", "重复测试2", default_limits={"free": 999})
-        count_after = len(reg.list_dimensions())
-        assert count_before == count_after
-        assert reg.get_limit("free", "dup") == 10  # 仍然是第一次的值
-
-    def test_list_dimensions_returns_copy(self):
-        """list_dimensions 返回副本，修改不影响内部."""
-        reg = QuotaRegistry()
-        dims = reg.list_dimensions()
-        dims["hacked"] = "hack"
-        assert "hacked" not in reg.list_dimensions()
-
-    def test_register_partial_default_limits(self):
-        """只给部分套餐设置默认限制."""
-        reg = QuotaRegistry()
-        reg.register_dimension(
-            "partial",
-            "部分套餐",
-            default_limits={"premium": 100},
-        )
-        assert reg.get_limit("free", "partial") == 0
-        assert reg.get_limit("basic", "partial") == 0
-        assert reg.get_limit("premium", "partial") == 100
-
-
-class TestQuotaChecker:
-    """QuotaChecker 测试."""
-
-    def test_check_within_limit(self):
-        """在限制内，allowed=True."""
-        checker = QuotaChecker()
-        result = checker.check("free", QuotaDimension.STORAGE_GB, 1)
-        assert result.allowed is True
-        assert result.dimension == QuotaDimension.STORAGE_GB
-        assert result.limit == 2
-        assert result.used == 1
-        assert result.remaining == 1
-        assert result.warning_level == QuotaWarningLevel.NORMAL
-
-    def test_check_exceeded(self):
-        """超出限制，allowed=False."""
-        checker = QuotaChecker()
-        result = checker.check("free", QuotaDimension.STORAGE_GB, 3)
-        assert result.allowed is False
-        assert result.remaining == 0
-        assert result.warning_level == QuotaWarningLevel.EXCEEDED
-
-    def test_check_exactly_at_limit(self):
-        """刚好等于限制视为超出（used < limit 才允许）."""
-        checker = QuotaChecker()
-        result = checker.check("free", QuotaDimension.STORAGE_GB, 2)
-        assert result.allowed is False
-
-    def test_check_unlimited(self):
-        """不限量维度."""
-        checker = QuotaChecker()
-        result = checker.check("premium", QuotaDimension.MAX_TEMPLATES, 999)
-        assert result.allowed is True
-        assert math.isinf(result.remaining)
-        assert result.warning_level == QuotaWarningLevel.NORMAL
-
-    def test_check_warning_level_boundaries(self):
-        """各告警级别的边界值."""
-        checker = QuotaChecker()
-        # 79% - normal
-        assert checker.check("free", QuotaDimension.STORAGE_GB, 1.58).warning_level == QuotaWarningLevel.NORMAL
-        # 80% - warning
-        assert checker.check("free", QuotaDimension.STORAGE_GB, 1.6).warning_level == QuotaWarningLevel.WARNING
-        # 95% - critical
-        assert checker.check("free", QuotaDimension.STORAGE_GB, 1.9).warning_level == QuotaWarningLevel.CRITICAL
-        # 100% - exceeded
-        assert checker.check("free", QuotaDimension.STORAGE_GB, 2).warning_level == QuotaWarningLevel.EXCEEDED
-
-    def test_check_zero_usage(self):
-        """0使用量."""
-        checker = QuotaChecker()
-        result = checker.check("basic", QuotaDimension.VIDEOS_PER_MONTH, 0)
-        assert result.allowed is True
-        assert result.remaining == 30
-        assert result.usage_percent == 0.0
-
-    def test_check_unknown_plan(self):
-        """未知套餐，限制为0."""
-        checker = QuotaChecker()
-        result = checker.check("enterprise", QuotaDimension.STORAGE_GB, 0)
-        assert result.limit == 0
-        # used=0, limit=0 → 0 < 0 is False → allowed=False
-        assert result.allowed is False
-        assert result.warning_level == QuotaWarningLevel.NORMAL
-
-    def test_check_unknown_dimension(self):
-        """未知维度，限制为0."""
-        checker = QuotaChecker()
-        result = checker.check("free", "unknown", 0)
-        assert result.limit == 0
-
-    def test_check_multiple(self):
-        """批量检查多个维度."""
-        checker = QuotaChecker()
-        usage = {
-            QuotaDimension.STORAGE_GB: 1,
-            QuotaDimension.VIDEOS_PER_MONTH: 10,
-        }
-        results = checker.check_multiple("free", usage)
-        assert len(results) == 2
-        dims = {r.dimension for r in results}
-        assert QuotaDimension.STORAGE_GB in dims
-        assert QuotaDimension.VIDEOS_PER_MONTH in dims
-
-    def test_check_multiple_empty(self):
-        """空字典返回空列表."""
-        checker = QuotaChecker()
-        results = checker.check_multiple("free", {})
-        assert results == []
-
-    def test_remaining_never_negative(self):
-        """剩余量不为负."""
-        checker = QuotaChecker()
-        result = checker.check("free", QuotaDimension.STORAGE_GB, 100)
-        assert result.remaining >= 0
-
-    def test_custom_registry(self):
-        """使用自定义 registry."""
-        reg = QuotaRegistry()
-        reg.register_dimension("custom", "自定义", default_limits={"free": 42})
-        checker = QuotaChecker(reg)
-        result = checker.check("free", "custom", 10)
-        assert result.limit == 42
-        assert result.allowed is True
-
-
-class TestGlobalSingletons:
-    """全局单例测试."""
-
-    def test_quota_registry_exists(self):
-        """全局 registry 单例存在."""
-        assert quota_registry is not None
-        assert isinstance(quota_registry, QuotaRegistry)
-
-    def test_quota_checker_exists(self):
-        """全局 checker 单例存在."""
-        assert quota_checker is not None
-        assert isinstance(quota_checker, QuotaChecker)
-
-    def test_global_checker_works(self):
-        """全局 checker 能正常工作."""
-        result = quota_checker.check("free", QuotaDimension.STORAGE_GB, 1)
-        assert result.allowed is True
-        assert result.limit == 2

tests/unit/domain/test_url_security.py

@@ -0,0 +1,567 @@
+"""url_security 单测.
+
+domain 层 URL 安全校验纯逻辑模块，0 网络依赖。
+覆盖 SSRF 防护、主机名校验、IP 检查、魔数校验等。
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from packages.domain.url_security import (
+    ALLOWED_IMAGE_MIME_TYPES,
+    ALLOWED_PORTS,
+    ALLOWED_SCHEMES,
+    ALLOWED_VIDEO_MIME_TYPES,
+    MAGIC_NUMBERS,
+    MAX_URL_LENGTH,
+    UrlSecurityError,
+    check_internal_hostname,
+    check_ssrf_ip,
+    is_ip_address,
+    is_trusted_domain,
+    is_url_basic_safe,
+    validate_magic_number,
+    validate_url_basic,
+)
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# 常量与异常类
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestConstants:
+    """常量测试."""
+
+    def test_allowed_schemes(self):
+        """允许的 scheme 包含 http 和 https."""
+        assert "http" in ALLOWED_SCHEMES
+        assert "https" in ALLOWED_SCHEMES
+
+    def test_allowed_ports(self):
+        """允许的端口：80, 443."""
+        assert 80 in ALLOWED_PORTS
+        assert 443 in ALLOWED_PORTS
+
+    def test_max_url_length(self):
+        """最大 URL 长度 2048."""
+        assert MAX_URL_LENGTH == 2048
+
+    def test_magic_numbers_has_common_formats(self):
+        """魔数表包含常见格式."""
+        assert "image/jpeg" in MAGIC_NUMBERS
+        assert "image/png" in MAGIC_NUMBERS
+        assert "image/gif" in MAGIC_NUMBERS
+        assert "video/mp4" in MAGIC_NUMBERS
+        assert "audio/mpeg" in MAGIC_NUMBERS
+
+
+class TestUrlSecurityError:
+    """异常类测试."""
+
+    def test_is_value_error(self):
+        """UrlSecurityError 继承 ValueError."""
+        assert issubclass(UrlSecurityError, ValueError)
+
+    def test_raise_with_message(self):
+        """抛出时携带错误信息."""
+        with pytest.raises(UrlSecurityError, match="test error"):
+            raise UrlSecurityError("test error")
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# check_internal_hostname
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestCheckInternalHostname:
+    """内部主机名检查测试."""
+
+    def test_normal_domain_passes(self):
+        """普通外部域名通过."""
+        check_internal_hostname("example.com")
+        check_internal_hostname("www.google.com")
+
+    def test_localhost_blocked(self):
+        """localhost 被拦截."""
+        with pytest.raises(UrlSecurityError, match="内部主机名"):
+            check_internal_hostname("localhost")
+
+    def test_localhost_case_insensitive(self):
+        """大小写不敏感."""
+        with pytest.raises(UrlSecurityError):
+            check_internal_hostname("LOCALHOST")
+        with pytest.raises(UrlSecurityError):
+            check_internal_hostname("LocalHost")
+
+    def test_localhost_localdomain_blocked(self):
+        """localhost.localdomain 被拦截."""
+        with pytest.raises(UrlSecurityError):
+            check_internal_hostname("localhost.localdomain")
+
+    def test_metadata_blocked(self):
+        """metadata 被拦截."""
+        with pytest.raises(UrlSecurityError):
+            check_internal_hostname("metadata")
+
+    def test_metadata_google_internal_blocked(self):
+        """GCP 元数据服务被拦截."""
+        with pytest.raises(UrlSecurityError):
+            check_internal_hostname("metadata.google.internal")
+
+    def test_cloud_metadata_ip_blocked(self):
+        """云元数据 IP 169.254.169.254 被拦截."""
+        with pytest.raises(UrlSecurityError):
+            check_internal_hostname("169.254.169.254")
+
+    def test_local_suffix_blocked(self):
+        """.local 后缀域名被拦截."""
+        with pytest.raises(UrlSecurityError, match="内网域名"):
+            check_internal_hostname("myhost.local")
+
+    def test_internal_suffix_blocked(self):
+        """.internal 后缀被拦截."""
+        with pytest.raises(UrlSecurityError):
+            check_internal_hostname("svc.cluster.internal")
+
+    def test_localdomain_suffix_blocked(self):
+        """.localdomain 后缀被拦截."""
+        with pytest.raises(UrlSecurityError):
+            check_internal_hostname("host.localdomain")
+
+    def test_com_domain_not_blocked(self):
+        """.com 域名不被拦截."""
+        check_internal_hostname("example.com")
+        check_internal_hostname("sub.example.com")
+
+    def test_subdomain_of_public_domain_ok(self):
+        """公网域名的子域名正常."""
+        check_internal_hostname("api.example.com")
+        check_internal_hostname("cdn.assets.example.org")
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# is_trusted_domain
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestIsTrustedDomain:
+    """可信域名匹配测试."""
+
+    def test_empty_trusted_domains_allows_all(self):
+        """空集合允许所有域名."""
+        assert is_trusted_domain("anything.com", set()) is True
+        assert is_trusted_domain("anywhere.org", set()) is True
+
+    def test_exact_match(self):
+        """精确匹配."""
+        trusted = {"example.com", "example.org"}
+        assert is_trusted_domain("example.com", trusted) is True
+        assert is_trusted_domain("example.org", trusted) is True
+
+    def test_subdomain_match(self):
+        """子域名匹配."""
+        trusted = {"example.com"}
+        assert is_trusted_domain("api.example.com", trusted) is True
+        assert is_trusted_domain("cdn.assets.example.com", trusted) is True
+
+    def test_no_match(self):
+        """不匹配."""
+        trusted = {"example.com"}
+        assert is_trusted_domain("other.com", trusted) is False
+        assert is_trusted_domain("example.net", trusted) is False
+
+    def test_case_insensitive(self):
+        """大小写不敏感."""
+        trusted = {"Example.COM"}
+        assert is_trusted_domain("example.com", trusted) is True
+        assert is_trusted_domain("API.EXAMPLE.COM", trusted) is True
+
+    def test_partial_match_no(self):
+        """域名部分相同但不是子域名不匹配."""
+        trusted = {"example.com"}
+        # fakeexample.com 不是 example.com 的子域名
+        assert is_trusted_domain("fakeexample.com", trusted) is False
+
+    def test_none_trusted_domains(self):
+        """trusted_domains 为 None 时由调用方处理，空 set 全允许."""
+        # 传空集合时全允许
+        assert is_trusted_domain("a.com", set()) is True
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# check_ssrf_ip
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestCheckSsrIp:
+    """IP SSRF 检查测试."""
+
+    def test_public_ip_passes(self):
+        """公网 IP 通过."""
+        check_ssrf_ip("8.8.8.8")
+        check_ssrf_ip("1.1.1.1")
+        check_ssrf_ip("114.114.114.114")
+
+    def test_loopback_blocked(self):
+        """回环地址被拦截."""
+        with pytest.raises(UrlSecurityError, match="回环"):
+            check_ssrf_ip("127.0.0.1")
+        with pytest.raises(UrlSecurityError):
+            check_ssrf_ip("127.0.0.53")
+
+    def test_private_ip_blocked(self):
+        """私有内网 IP 被拦截."""
+        with pytest.raises(UrlSecurityError, match="内网"):
+            check_ssrf_ip("192.168.1.1")
+        with pytest.raises(UrlSecurityError):
+            check_ssrf_ip("10.0.0.1")
+        with pytest.raises(UrlSecurityError):
+            check_ssrf_ip("172.16.0.1")
+
+    def test_link_local_blocked(self):
+        """链路本地地址被拦截."""
+        with pytest.raises(UrlSecurityError, match="链路本地"):
+            check_ssrf_ip("169.254.169.254")
+        with pytest.raises(UrlSecurityError):
+            check_ssrf_ip("169.254.1.1")
+
+    def test_multicast_blocked(self):
+        """组播地址被拦截."""
+        with pytest.raises(UrlSecurityError, match="组播"):
+            check_ssrf_ip("224.0.0.1")
+        with pytest.raises(UrlSecurityError):
+            check_ssrf_ip("239.255.255.250")
+
+    def test_unspecified_blocked(self):
+        """未指定地址被拦截."""
+        with pytest.raises(UrlSecurityError, match="未指定"):
+            check_ssrf_ip("0.0.0.0")
+
+    def test_ipv6_loopback_blocked(self):
+        """IPv6 回环地址被拦截."""
+        with pytest.raises(UrlSecurityError):
+            check_ssrf_ip("::1")
+
+    def test_ipv6_private_blocked(self):
+        """IPv6 内网地址被拦截."""
+        with pytest.raises(UrlSecurityError):
+            check_ssrf_ip("fc00::1")
+        with pytest.raises(UrlSecurityError):
+            check_ssrf_ip("fe80::1")
+
+    def test_ipv6_public_passes(self):
+        """IPv6 公网地址通过."""
+        check_ssrf_ip("2001:4860:4860::8888")
+
+    def test_invalid_ip_raises_value_error(self):
+        """非法 IP 抛出 ValueError（不是 UrlSecurityError）."""
+        with pytest.raises(ValueError):
+            check_ssrf_ip("not-an-ip")
+        with pytest.raises(ValueError):
+            check_ssrf_ip("999.999.999.999")
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# is_ip_address
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestIsIpAddress:
+    """IP 地址判断测试."""
+
+    def test_ipv4_true(self):
+        """IPv4 地址返回 True."""
+        assert is_ip_address("127.0.0.1") is True
+        assert is_ip_address("8.8.8.8") is True
+        assert is_ip_address("0.0.0.0") is True
+
+    def test_ipv6_true(self):
+        """IPv6 地址返回 True."""
+        assert is_ip_address("::1") is True
+        assert is_ip_address("2001:db8::1") is True
+
+    def test_hostname_false(self):
+        """主机名返回 False."""
+        assert is_ip_address("example.com") is False
+        assert is_ip_address("localhost") is False
+        assert is_ip_address("sub.domain.org") is False
+
+    def test_empty_string_false(self):
+        """空字符串返回 False."""
+        assert is_ip_address("") is False
+
+    def test_invalid_ip_false(self):
+        """非法 IP 返回 False."""
+        assert is_ip_address("999.999.999.999") is False
+        assert is_ip_address("1234") is False
+        assert is_ip_address("abc.def") is False
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# validate_url_basic
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestValidateUrlBasic:
+    """URL 基础校验测试."""
+
+    def test_normal_https_url_passes(self):
+        """正常 HTTPS URL 通过."""
+        result = validate_url_basic("https://example.com/path")
+        assert result == "https://example.com/path"
+
+    def test_normal_http_url_passes(self):
+        """正常 HTTP URL 通过."""
+        result = validate_url_basic("http://example.com/path")
+        assert result == "http://example.com/path"
+
+    def test_empty_url_rejected(self):
+        """空 URL 被拒."""
+        with pytest.raises(UrlSecurityError, match="为空"):
+            validate_url_basic("")
+
+    def test_none_url_not_passed_as_str(self):
+        """None 作为 URL（这里只测空字符串）."""
+        # 空字符串被拒
+        with pytest.raises(UrlSecurityError):
+            validate_url_basic("")
+
+    def test_too_long_url_rejected(self):
+        """超长 URL 被拒."""
+        long_url = "https://example.com/" + "a" * 3000
+        with pytest.raises(UrlSecurityError, match="过长"):
+            validate_url_basic(long_url)
+
+    def test_invalid_scheme_rejected(self):
+        """非法 scheme 被拒."""
+        with pytest.raises(UrlSecurityError, match="scheme"):
+            validate_url_basic("ftp://example.com/file")
+        with pytest.raises(UrlSecurityError):
+            validate_url_basic("file:///etc/passwd")
+        with pytest.raises(UrlSecurityError):
+            validate_url_basic("javascript:alert(1)")
+
+    def test_missing_hostname_rejected(self):
+        """缺少主机名被拒."""
+        with pytest.raises(UrlSecurityError, match="主机名"):
+            validate_url_basic("https:///path")
+
+    def test_localhost_rejected(self):
+        """localhost 被拒."""
+        with pytest.raises(UrlSecurityError):
+            validate_url_basic("https://localhost/api")
+
+    def test_internal_domain_rejected(self):
+        """内网域名被拒."""
+        with pytest.raises(UrlSecurityError):
+            validate_url_basic("http://server.local/api")
+
+    def test_non_standard_port_rejected(self):
+        """非标准端口被拒."""
+        with pytest.raises(UrlSecurityError, match="端口"):
+            validate_url_basic("https://example.com:8080/")
+        with pytest.raises(UrlSecurityError):
+            validate_url_basic("http://example.com:3000/")
+
+    def test_port_80_ok(self):
+        """80 端口允许."""
+        validate_url_basic("http://example.com:80/path")
+
+    def test_port_443_ok(self):
+        """443 端口允许."""
+        validate_url_basic("https://example.com:443/path")
+
+    def test_no_port_ok(self):
+        """无端口默认允许."""
+        validate_url_basic("https://example.com/path")
+
+    def test_direct_ip_rejected_by_default(self):
+        """默认禁止直接 IP 访问."""
+        with pytest.raises(UrlSecurityError, match="直接 IP"):
+            validate_url_basic("https://8.8.8.8/path")
+
+    def test_direct_ip_allowed_when_enabled(self):
+        """allow_direct_ip=True 时允许公网 IP."""
+        validate_url_basic("https://8.8.8.8/path", allow_direct_ip=True)
+
+    def test_direct_ip_private_still_blocked(self):
+        """即使 allow_direct_ip，内网 IP 仍被拒."""
+        with pytest.raises(UrlSecurityError, match="内网"):
+            validate_url_basic("https://192.168.1.1/", allow_direct_ip=True)
+
+    def test_direct_ip_loopback_still_blocked(self):
+        """回环 IP 即使开启 direct_ip 也被拒."""
+        with pytest.raises(UrlSecurityError):
+            validate_url_basic("https://127.0.0.1/", allow_direct_ip=True)
+
+    def test_trusted_domains_pass(self):
+        """可信域名列表内的域名通过."""
+        trusted = {"example.com", "cdn.com"}
+        validate_url_basic("https://api.example.com/path", trusted_domains=trusted)
+        validate_url_basic("https://cdn.com/asset.jpg", trusted_domains=trusted)
+
+    def test_untrusted_domain_rejected(self):
+        """不在可信域名列表中的域名被拒."""
+        trusted = {"example.com"}
+        with pytest.raises(UrlSecurityError, match="白名单"):
+            validate_url_basic("https://evil.com/malware", trusted_domains=trusted)
+
+    def test_trusted_domain_subdomain_pass(self):
+        """可信域名的子域名通过."""
+        trusted = {"example.com"}
+        validate_url_basic("https://sub.example.com/a", trusted_domains=trusted)
+        validate_url_basic("https://a.b.example.com/b", trusted_domains=trusted)
+
+    def test_return_value_is_original_url(self):
+        """返回原始 URL 字符串."""
+        url = "https://example.com/path?query=value#frag"
+        assert validate_url_basic(url) == url
+
+    def test_metadata_ip_rejected(self):
+        """云元数据 IP 被内部主机名检查拦截."""
+        with pytest.raises(UrlSecurityError):
+            validate_url_basic("http://169.254.169.254/latest/meta-data/")
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# is_url_basic_safe
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestIsUrlBasicSafe:
+    """便捷函数 is_url_basic_safe 测试."""
+
+    def test_safe_url_returns_true(self):
+        """安全 URL 返回 True."""
+        assert is_url_basic_safe("https://example.com/") is True
+        assert is_url_basic_safe("http://example.org/path") is True
+
+    def test_unsafe_url_returns_false(self):
+        """不安全 URL 返回 False."""
+        assert is_url_basic_safe("https://localhost/") is False
+        assert is_url_basic_safe("ftp://example.com/") is False
+        assert is_url_basic_safe("") is False
+
+    def test_trusted_domains_param(self):
+        """支持 trusted_domains 参数."""
+        trusted = {"example.com"}
+        assert is_url_basic_safe("https://other.com/", trusted_domains=trusted) is False
+        assert is_url_basic_safe("https://example.com/", trusted_domains=trusted) is True
+
+    def test_allow_direct_ip_param(self):
+        """支持 allow_direct_ip 参数."""
+        assert is_url_basic_safe("https://8.8.8.8/") is False
+        assert is_url_basic_safe("https://8.8.8.8/", allow_direct_ip=True) is True
+
+    def test_no_exceptions_raised(self):
+        """不抛出异常，只返回 bool."""
+        # 各种边界情况都不抛异常
+        try:
+            is_url_basic_safe("")
+            is_url_basic_safe("not a url")
+            is_url_basic_safe("http://" + "a" * 3000)
+        except UrlSecurityError:
+            pytest.fail("is_url_basic_safe should not raise UrlSecurityError")
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# validate_magic_number
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestValidateMagicNumber:
+    """魔数校验测试."""
+
+    def test_jpeg_valid(self):
+        """JPEG 文件通过."""
+        # JPEG 文件头: FF D8 FF
+        jpeg_header = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00"
+        validate_magic_number(jpeg_header, {"image/jpeg"})
+
+    def test_png_valid(self):
+        """PNG 文件通过."""
+        png_header = b"\x89PNG\r\n\x1a\n\x00\x00\x00"
+        validate_magic_number(png_header, {"image/png"})
+
+    def test_gif_valid(self):
+        """GIF 文件通过（GIF89a 和 GIF87a）."""
+        validate_magic_number(b"GIF89a...", {"image/gif"})
+        validate_magic_number(b"GIF87a...", {"image/gif"})
+
+    def test_wav_valid(self):
+        """WAV 文件通过（RIFF + WAVE）."""
+        wav_header = b"RIFF\x00\x00\x00\x00WAVEfmt "
+        validate_magic_number(wav_header, {"audio/wav"})
+
+    def test_mp3_id3_valid(self):
+        """带 ID3 标签的 MP3 通过."""
+        mp3_header = b"ID3\x03\x00\x00\x00\x00\x0f\x76"
+        validate_magic_number(mp3_header, {"audio/mpeg"})
+
+    def test_mp3_sync_valid(self):
+        """不带 ID3 的 MP3（帧同步字）通过."""
+        mp3_header = b"\xff\xfb\x90\x00" + b"\x00" * 32
+        validate_magic_number(mp3_header, {"audio/mpeg"})
+
+    def test_ogg_valid(self):
+        """OGG 文件通过."""
+        validate_magic_number(b"OggS\x00\x00...", {"audio/ogg"})
+
+    def test_flac_valid(self):
+        """FLAC 文件通过."""
+        validate_magic_number(b"fLaC\x00\x00...", {"audio/flac"})
+
+    def test_webp_valid(self):
+        """WebP 文件通过（RIFF + WEBP）."""
+        webp_header = b"RIFF\x00\x00\x00\x00WEBPVP8 "
+        validate_magic_number(webp_header, {"image/webp"})
+
+    def test_bmp_valid(self):
+        """BMP 文件通过."""
+        validate_magic_number(b"BM\x00\x00\x00\x00...", {"image/bmp"})
+
+    def test_mp4_valid(self):
+        """MP4 文件通过（ftyp 在偏移 4）."""
+        mp4_header = b"\x00\x00\x00\x20ftypisom\x00\x00\x02\x00"
+        validate_magic_number(mp4_header, {"video/mp4"})
+
+    def test_invalid_format_rejected(self):
+        """不匹配的格式被拒."""
+        with pytest.raises(UrlSecurityError, match="魔数"):
+            validate_magic_number(b"hello world", {"image/jpeg"})
+
+    def test_empty_bytes_rejected(self):
+        """空字节被拒."""
+        with pytest.raises(UrlSecurityError, match="为空"):
+            validate_magic_number(b"", {"image/jpeg"})
+
+    def test_too_short_bytes_rejected(self):
+        """字节太短不匹配魔数时被拒."""
+        with pytest.raises(UrlSecurityError):
+            validate_magic_number(b"\xff\xd8", {"image/jpeg"})  # 只2字节，不够JPEG魔数
+
+    def test_multiple_allowed_types(self):
+        """允许多种格式时任一匹配即通过."""
+        jpeg_header = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00"
+        validate_magic_number(jpeg_header, {"image/jpeg", "image/png", "image/gif"})
+
+    def test_wrong_type_rejected(self):
+        """用 PNG 魔数校验 JPEG 类型失败."""
+        jpeg_header = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00"
+        with pytest.raises(UrlSecurityError):
+            validate_magic_number(jpeg_header, {"image/png"})
+
+    def test_unknown_mime_skipped(self):
+        """未知 MIME 类型（无对应魔数）不阻断."""
+        # application/octet-stream 没有魔数定义，直接通过
+        validate_magic_number(b"random bytes here", {"application/octet-stream"})
+
+    def test_allowed_image_mime_types_has_common(self):
+        """图片 MIME 白名单包含常见类型."""
+        assert "image/jpeg" in ALLOWED_IMAGE_MIME_TYPES
+        assert "image/png" in ALLOWED_IMAGE_MIME_TYPES
+
+    def test_allowed_video_mime_types_has_common(self):
+        """视频 MIME 白名单包含常见类型."""
+        assert "video/mp4" in ALLOWED_VIDEO_MIME_TYPES

tests/unit/domain/test_video_filter_builder.py

@@ -6,7 +6,6 @@ domain 层纯逻辑模块，0 FFmpeg 依赖，快速轻量。
 from __future__ import annotations
 
 import dataclasses
-
 from unittest.mock import MagicMock
 
 import pytest

tests/unit/test_transition_presets.py

@@ -4,7 +4,6 @@ from __future__ import annotations
 
 import dataclasses
 
-
 import pytest
 
 from packages.domain.transition_presets import (