diff --git a/tests/unit/test_generation_preview.py b/tests/unit/test_generation_preview.py
index 8ff8b2c73..06b7f839a 100755
--- a/tests/unit/test_generation_preview.py
+++ b/tests/unit/test_generation_preview.py
@@ -248,7 +248,6 @@ class TestPreviewGenerationTaskResponse:
 
     def test_pending_state_response(self):
         """pending 状态的响应"""
-        from app.schemas.generation_task import PreviewGenerationTaskResponse
 
         resp = PreviewGenerationTaskResponse(
             task_id="task_123",
@@ -270,8 +269,6 @@ class TestPreviewGenerationTaskResponse:
         """completed 状态的响应"""
         from datetime import datetime, timezone
 
-        from app.schemas.generation_task import PreviewGenerationTaskResponse
-
         now = datetime.now(timezone.utc)
         resp = PreviewGenerationTaskResponse(
             task_id="task_123",
@@ -302,7 +299,6 @@ class TestPreviewGenerationTaskResponse:
 
     def test_failed_state_response(self):
         """failed 状态的响应"""
-        from app.schemas.generation_task import PreviewGenerationTaskResponse
 
         resp = PreviewGenerationTaskResponse(
             task_id="task_123",
@@ -559,3 +555,487 @@ class TestPreviewTaskStatusFlow:
             asset_library_id="lib1",
         )
         assert task.is_preview is False
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# 路由层测试 — 覆盖 generation_preview.py 的 HTTP handler 逻辑
+# ═══════════════════════════════════════════════════════════════════════════════
+
+from dataclasses import dataclass
+from unittest.mock import patch, PropertyMock
+
+# 导入路由模块和被测函数
+from app.api.routes.generation_preview import (
+    _to_preview_response,
+    create_preview_generation_task,
+    get_preview_generation_task,
+    PREVIEW_RESOLUTION,
+)
+from app.schemas.generation_task import (
+    CreatePreviewGenerationTaskRequest,
+    PreviewGenerationTaskResponse,
+)
+from app.core.task_enqueue import GlobalQueueFull, UserPendingLimitExceeded
+
+
+def _make_user(user_id="test_user_001"):
+    """构造 mock AuthenticatedUser"""
+    mock_user = MagicMock()
+    mock_user.id = user_id
+    auth = MagicMock()
+    auth.user = mock_user
+    auth.session_id = "sess_001"
+    auth.token_type = "bearer"
+    return auth
+
+
+def _make_task(
+    task_id="preview_task_001",
+    status=GenerationTaskStatus.PENDING,
+    progress=0.0,
+    is_preview=True,
+    resolution="854x480",
+    error_message="",
+    created_at=None,
+    started_at=None,
+    completed_at=None,
+    extra_meta=None,
+    asset_ids=None,
+):
+    """构造 mock GenerationTask"""
+    task = MagicMock()
+    task.id = task_id
+    task.status = status
+    task.progress = progress
+    task.is_preview = is_preview
+    task.resolution = resolution
+    task.error_message = error_message
+    task.created_at = created_at or datetime.now(timezone.utc)
+    task.started_at = started_at
+    task.completed_at = completed_at
+    task.extra_meta = extra_meta or {}
+    task.asset_ids = asset_ids or []
+    task.created_by_user_id = "test_user_001"
+    return task
+
+
+class TestToPreviewResponse:
+    """_to_preview_response 辅助函数测试"""
+
+    def test_basic_pending_task(self):
+        """基本 pending 任务转换"""
+        task = _make_task(status=GenerationTaskStatus.PENDING)
+        resp = _to_preview_response(task)
+        assert resp.task_id == "preview_task_001"
+        assert resp.status == "pending"
+        assert resp.is_preview is True
+        assert resp.video_url == ""
+        assert resp.duration == 0.0
+        assert resp.file_size == 0
+
+    def test_completed_task_with_videos(self):
+        """已完成任务，带视频结果"""
+        task = _make_task(
+            status=GenerationTaskStatus.COMPLETED,
+            progress=100.0,
+        )
+        video = MagicMock()
+        video.file_url = "https://cdn.example.com/preview.mp4"
+        video.duration = 30.5
+        video.file_size = 1024000
+        resp = _to_preview_response(task, generated_videos=[video])
+        assert resp.video_url == "https://cdn.example.com/preview.mp4"
+        assert resp.duration == 30.5
+        assert resp.file_size == 1024000
+
+    def test_extra_meta_fields(self):
+        """从 extra_meta 提取统计信息"""
+        task = _make_task(
+            extra_meta={
+                "clip_count": 5,
+                "transition_count": 4,
+                "material_usage": {"video": 3, "audio": 2},
+            }
+        )
+        resp = _to_preview_response(task)
+        assert resp.clip_count == 5
+        assert resp.transition_count == 4
+        assert resp.material_usage == {"video": 3, "audio": 2}
+
+    def test_generate_duration_calculation(self):
+        """计算生成耗时"""
+        from datetime import timedelta
+
+        started = datetime(2026, 8, 1, 10, 0, 0, tzinfo=timezone.utc)
+        completed = started + timedelta(seconds=45)
+        task = _make_task(started_at=started, completed_at=completed)
+        resp = _to_preview_response(task)
+        assert resp.generate_duration == 45.0
+
+    def test_no_started_completed_time(self):
+        """无开始/完成时间时 generate_duration=0"""
+        task = _make_task(started_at=None, completed_at=None)
+        resp = _to_preview_response(task)
+        assert resp.generate_duration == 0.0
+
+    def test_empty_generated_videos(self):
+        """空视频列表时 video_url 为空"""
+        task = _make_task()
+        resp = _to_preview_response(task, generated_videos=[])
+        assert resp.video_url == ""
+
+    def test_none_generated_videos(self):
+        """None 视频列表时 video_url 为空"""
+        task = _make_task()
+        resp = _to_preview_response(task, generated_videos=None)
+        assert resp.video_url == ""
+
+
+class TestCreatePreviewRoute:
+    """POST /preview 路由测试"""
+
+    def _make_request(self, **kwargs):
+        defaults = {
+            "template_id": "tpl_001",
+            "asset_ids": ["asset_1", "asset_2"],
+            "title_ids": [],
+            "voice_ids": [],
+            "video_title": "测试预览",
+            "duration": 0.0,
+            "video_ratio": "",
+            "bgm_config": {},
+        }
+        defaults.update(kwargs)
+        return CreatePreviewGenerationTaskRequest(**defaults)
+
+    def test_create_success(self):
+        """正常创建预览任务"""
+        repo = MagicMock()
+        repo.count_pending_by_user.return_value = 0
+        repo.count_pending_total.return_value = 5
+
+        task = _make_task()
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.return_value = task
+            with patch(
+                "app.api.routes.generation_preview.safe_enqueue_generation_task",
+                return_value=True,
+            ):
+                resp = create_preview_generation_task(
+                    self._make_request(),
+                    authenticated_user=_make_user(),
+                    generation_task_repository=repo,
+                )
+        assert resp.task_id == "preview_task_001"
+        assert resp.status == "pending"
+
+    def test_user_pending_limit_exceeded(self):
+        """用户待处理任务超限 → 429"""
+        repo = MagicMock()
+        repo.count_pending_by_user.return_value = 3
+        repo.count_pending_total.return_value = 5
+
+        from fastapi import HTTPException
+
+        with pytest.raises(HTTPException) as exc_info:
+            create_preview_generation_task(
+                self._make_request(),
+                authenticated_user=_make_user(),
+                generation_task_repository=repo,
+            )
+        assert exc_info.value.status_code == 429
+
+    def test_global_queue_full(self):
+        """全局队列满 → 503"""
+        repo = MagicMock()
+        repo.count_pending_by_user.return_value = 0
+        repo.count_pending_total.return_value = 20
+
+        from fastapi import HTTPException
+
+        with pytest.raises(HTTPException) as exc_info:
+            create_preview_generation_task(
+                self._make_request(),
+                authenticated_user=_make_user(),
+                generation_task_repository=repo,
+            )
+        assert exc_info.value.status_code == 503
+
+    def test_create_value_error(self):
+        """UseCase 抛出 ValueError → 400"""
+        repo = MagicMock()
+        repo.count_pending_by_user.return_value = 0
+        repo.count_pending_total.return_value = 0
+
+        from fastapi import HTTPException
+
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.side_effect = ValueError("template_id 不能为空")
+            with pytest.raises(HTTPException) as exc_info:
+                create_preview_generation_task(
+                    self._make_request(),
+                    authenticated_user=_make_user(),
+                    generation_task_repository=repo,
+                )
+            assert exc_info.value.status_code == 400
+
+    def test_create_unexpected_error(self):
+        """UseCase 抛出未知异常 → 500"""
+        repo = MagicMock()
+        repo.count_pending_by_user.return_value = 0
+        repo.count_pending_total.return_value = 0
+
+        from fastapi import HTTPException
+
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.side_effect = RuntimeError("db connection lost")
+            with pytest.raises(HTTPException) as exc_info:
+                create_preview_generation_task(
+                    self._make_request(),
+                    authenticated_user=_make_user(),
+                    generation_task_repository=repo,
+                )
+            assert exc_info.value.status_code == 500
+
+    def test_enqueue_returns_false(self):
+        """safe_enqueue 返回 False → 500"""
+        repo = MagicMock()
+        repo.count_pending_by_user.return_value = 0
+        repo.count_pending_total.return_value = 0
+
+        task = _make_task()
+        from fastapi import HTTPException
+
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.return_value = task
+            with patch(
+                "app.api.routes.generation_preview.safe_enqueue_generation_task",
+                return_value=False,
+            ):
+                with pytest.raises(HTTPException) as exc_info:
+                    create_preview_generation_task(
+                        self._make_request(),
+                        authenticated_user=_make_user(),
+                        generation_task_repository=repo,
+                    )
+                assert exc_info.value.status_code == 500
+
+    def test_enqueue_raises_user_limit(self):
+        """safe_enqueue 抛出 UserPendingLimitExceeded → 429"""
+        repo = MagicMock()
+        repo.count_pending_by_user.return_value = 0
+        repo.count_pending_total.return_value = 0
+
+        task = _make_task()
+        from fastapi import HTTPException
+
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.return_value = task
+            with patch(
+                "app.api.routes.generation_preview.safe_enqueue_generation_task",
+                side_effect=UserPendingLimitExceeded(user_id="u1", pending_count=4, limit=3),
+            ):
+                with pytest.raises(HTTPException) as exc_info:
+                    create_preview_generation_task(
+                        self._make_request(),
+                        authenticated_user=_make_user(),
+                        generation_task_repository=repo,
+                    )
+                assert exc_info.value.status_code == 429
+
+    def test_enqueue_raises_global_queue_full(self):
+        """safe_enqueue 抛出 GlobalQueueFull → 503"""
+        repo = MagicMock()
+        repo.count_pending_by_user.return_value = 0
+        repo.count_pending_total.return_value = 0
+
+        task = _make_task()
+        from fastapi import HTTPException
+
+        with patch("app.api.routes.generation_preview.CreateGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.return_value = task
+            with patch(
+                "app.api.routes.generation_preview.safe_enqueue_generation_task",
+                side_effect=GlobalQueueFull(pending_count=21, limit=20),
+            ):
+                with pytest.raises(HTTPException) as exc_info:
+                    create_preview_generation_task(
+                        self._make_request(),
+                        authenticated_user=_make_user(),
+                        generation_task_repository=repo,
+                    )
+                assert exc_info.value.status_code == 503
+
+
+class TestGetPreviewRoute:
+    """GET /preview/{task_id} 路由测试"""
+
+    def test_task_not_found(self):
+        """任务不存在 → 404"""
+        repo = MagicMock()
+        vid_repo = MagicMock()
+
+        from fastapi import HTTPException
+
+        with patch("app.api.routes.generation_preview.GetGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.return_value = None
+            with pytest.raises(HTTPException) as exc_info:
+                get_preview_generation_task(
+                    task_id="nonexistent",
+                    authenticated_user=_make_user(),
+                    generation_task_repository=repo,
+                    generated_video_repository=vid_repo,
+                )
+            assert exc_info.value.status_code == 404
+
+    def test_permission_denied(self):
+        """任务不属于当前用户 → 403"""
+        repo = MagicMock()
+        vid_repo = MagicMock()
+        task = _make_task()
+        task.created_by_user_id = "other_user_999"
+
+        from fastapi import HTTPException
+
+        with patch("app.api.routes.generation_preview.GetGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.return_value = task
+            with pytest.raises(HTTPException) as exc_info:
+                get_preview_generation_task(
+                    task_id="preview_task_001",
+                    authenticated_user=_make_user("test_user_001"),
+                    generation_task_repository=repo,
+                    generated_video_repository=vid_repo,
+                )
+            assert exc_info.value.status_code == 403
+
+    def test_not_preview_task(self):
+        """查询的不是预览任务 → 404"""
+        repo = MagicMock()
+        vid_repo = MagicMock()
+        task = _make_task(is_preview=False)
+        task.created_by_user_id = "test_user_001"
+
+        from fastapi import HTTPException
+
+        with patch("app.api.routes.generation_preview.GetGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.return_value = task
+            with pytest.raises(HTTPException) as exc_info:
+                get_preview_generation_task(
+                    task_id="preview_task_001",
+                    authenticated_user=_make_user(),
+                    generation_task_repository=repo,
+                    generated_video_repository=vid_repo,
+                )
+            assert exc_info.value.status_code == 404
+
+    def test_get_pending_task(self):
+        """查询 pending 状态的预览任务"""
+        repo = MagicMock()
+        vid_repo = MagicMock()
+        task = _make_task(status=GenerationTaskStatus.PENDING, progress=0.0)
+
+        with patch("app.api.routes.generation_preview.GetGenerationTaskUseCase") as MockUC:
+            MockUC.return_value.execute.return_value = task
+            resp = get_preview_generation_task(
+                task_id="preview_task_001",
+                authenticated_user=_make_user(),
+                generation_task_repository=repo,
+                generated_video_repository=vid_repo,
+            )
+        assert resp.status == "pending"
+        assert resp.is_preview is True
+
+    def test_get_completed_task_with_videos(self):
+        """查询 completed 状态任务，返回视频列表"""
+        repo = MagicMock()
+        vid_repo = MagicMock()
+        task = _make_task(status=GenerationTaskStatus.COMPLETED, progress=100.0)
+
+        video = MagicMock()
+        video.file_url = "https://cdn.example.com/preview_final.mp4"
+        video.duration = 25.0
+        video.file_size = 512000
+
+        with patch("app.api.routes.generation_preview.GetGenerationTaskUseCase") as MockGet:
+            MockGet.return_value.execute.return_value = task
+            with patch("app.api.routes.generation_preview.ListGeneratedVideosByTaskUseCase") as MockList:
+                MockList.return_value.execute.return_value = [video]
+                resp = get_preview_generation_task(
+                    task_id="preview_task_001",
+                    authenticated_user=_make_user(),
+                    generation_task_repository=repo,
+                    generated_video_repository=vid_repo,
+                )
+        assert resp.video_url == "https://cdn.example.com/preview_final.mp4"
+        assert resp.duration == 25.0
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# Worker 预览模式测试 — 覆盖 generation.py 预览分辨率强制逻辑
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestWorkerPreviewResolution:
+    """Worker 层预览模式分辨率强制逻辑测试"""
+
+    def test_preview_mode_forces_480p(self):
+        """预览模式强制 854x480 + 1M 码率"""
+        # 模拟 worker 中 export_cfg 逻辑
+        is_preview = True
+        resolution = "1920x1080"  # 用户指定的分辨率应被忽略
+        plan_cfg = {"export": {"resolution": "1280x720", "bitrate": "5M"}}
+        export_cfg = plan_cfg.get("export", {}) or {}
+
+        if is_preview:
+            export_cfg["resolution"] = "854x480"
+            export_cfg["bitrate"] = "1M"
+        elif resolution:
+            export_cfg["resolution"] = resolution
+
+        assert export_cfg["resolution"] == "854x480"
+        assert export_cfg["bitrate"] == "1M"
+
+    def test_non_preview_uses_user_resolution(self):
+        """非预览模式使用用户指定分辨率"""
+        is_preview = False
+        resolution = "1920x1080"
+        plan_cfg = {"export": {"resolution": "1280x720"}}
+        export_cfg = plan_cfg.get("export", {}) or {}
+
+        if is_preview:
+            export_cfg["resolution"] = "854x480"
+            export_cfg["bitrate"] = "1M"
+        elif resolution:
+            export_cfg["resolution"] = resolution
+
+        assert export_cfg["resolution"] == "1920x1080"
+
+    def test_non_preview_no_resolution_uses_template(self):
+        """非预览模式且用户未指定分辨率，使用模板配置"""
+        is_preview = False
+        resolution = ""
+        plan_cfg = {"export": {"resolution": "1280x720"}}
+        export_cfg = plan_cfg.get("export", {}) or {}
+
+        if is_preview:
+            export_cfg["resolution"] = "854x480"
+            export_cfg["bitrate"] = "1M"
+        elif resolution:
+            export_cfg["resolution"] = resolution
+        elif not export_cfg.get("resolution"):
+            export_cfg["resolution"] = "1280x720"
+
+        assert export_cfg["resolution"] == "1280x720"
+
+    def test_preview_mode_empty_export_cfg(self):
+        """预览模式且模板无 export 配置"""
+        is_preview = True
+        plan_cfg = {}
+        export_cfg = plan_cfg.get("export", {}) or {}
+
+        if is_preview:
+            export_cfg["resolution"] = "854x480"
+            export_cfg["bitrate"] = "1M"
+
+        assert export_cfg["resolution"] == "854x480"
+        assert export_cfg["bitrate"] == "1M"