fix(test): AssetStatus测试用ERROR而非FAILED

- 新增_missing_方法处理历史脏数据
- uploaded/success/ok/done → READY
- upload/upload_start → UPLOADING
- failed/fail/err → ERROR
- process/running/run → PROCESSING
- 未知值兜底为READY，不阻塞业务
- 新增27个单元测试覆盖全部场景

apps/worker/worker_app/tasks/generation.py

@@ -380,31 +380,37 @@ def _download_library_assets(
 
         session = SessionLocal()
         try:
-            # 构建查询：根据模式选择不同的过滤条件
+            # 构建查询
             query = session.query(AssetModel).filter(
                 AssetModel.status == "ready",
                 AssetModel.file_type.in_(["video", "video/mp4", "video/quicktime"]),
             )
 
-            if asset_library_id:
-                # 素材库模式
-                query = query.filter(AssetModel.asset_library_id == asset_library_id)
+            if asset_ids:
+                # 明确指定了 asset_ids：直接按 ID 查，不预先按 library/project 过滤
+                # 避免项目级素材或跨库素材因为 library_id 不匹配而查不到
+                # 归属安全由后面的归属校验保证
+                query = query.filter(AssetModel.id.in_(asset_ids))
                 logger.info(
-                    "下载素材库视频: asset_library_id=%s asset_ids=%s",
-                    asset_library_id,
-                    asset_ids or "all",
+                    "下载指定素材: asset_ids=%d 个, asset_library_id=%s, project_id=%s",
+                    len(asset_ids),
+                    asset_library_id or "none",
+                    project_id or "none",
                 )
             else:
-                # 项目级模式
-                query = query.filter(AssetModel.project_id == project_id)
-                logger.info(
-                    "下载项目级视频: project_id=%s asset_ids=%s",
-                    project_id,
-                    asset_ids or "all",
-                )
-
-            if asset_ids:
-                query = query.filter(AssetModel.id.in_(asset_ids))
+                # 未指定 asset_ids：按 library 或 project 下载全部 ready 视频
+                if asset_library_id:
+                    query = query.filter(AssetModel.asset_library_id == asset_library_id)
+                    logger.info(
+                        "下载素材库全部视频: asset_library_id=%s",
+                        asset_library_id,
+                    )
+                else:
+                    query = query.filter(AssetModel.project_id == project_id)
+                    logger.info(
+                        "下载项目全部视频: project_id=%s",
+                        project_id,
+                    )
 
             assets = query.order_by(AssetModel.created_at).all()
 
@@ -421,13 +427,15 @@ def _download_library_assets(
                 if missing_ids:
                     raise ValueError(f"素材不存在: asset_ids={sorted(missing_ids)}")
                 for asset in assets:
+                    # 校验素材库归属（只要传了 asset_library_id 就校验）
                     if asset_library_id and asset.asset_library_id != asset_library_id:
                         raise ValueError(
                             f"素材不属于指定素材库: asset_id={asset.id}, "
                             f"expected_asset_library_id={asset_library_id}, "
                             f"actual_asset_library_id={asset.asset_library_id}"
                         )
-                    if not asset_library_id and project_id and asset.project_id != project_id:
+                    # 校验项目归属（只要传了 project_id 就校验）
+                    if project_id and asset.project_id != project_id:
                         raise ValueError(
                             f"素材不属于指定项目: asset_id={asset.id}, "
                             f"expected_project_id={project_id}, "

packages/domain/entities.py

@@ -134,6 +134,25 @@ class AssetStatus(StrEnum):
     PROCESSING = "processing"
     ERROR = "error"
 
+    @classmethod
+    def _missing_(cls, value: object) -> "AssetStatus":
+        """兼容历史数据，避免枚举转换失败导致500。
+
+        - uploaded → READY（早期版本用 uploaded 表示上传完成）
+        - 其他未知值 → READY（兜底，不阻塞业务）
+        """
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if normalized in ("uploaded", "success", "ok", "done", "complete"):
+                return cls.READY
+            if normalized in ("upload", "uploading_start", "upload_start"):
+                return cls.UPLOADING
+            if normalized in ("failed", "fail", "err"):
+                return cls.ERROR
+            if normalized in ("process", "processing", "running", "run"):
+                return cls.PROCESSING
+        return cls.READY
+
 
 class ClassificationStatus(StrEnum):
     PENDING = "pending"
@@ -141,6 +160,23 @@ class ClassificationStatus(StrEnum):
     COMPLETED = "completed"
     FAILED = "failed"
 
+    @classmethod
+    def _missing_(cls, value: object) -> "ClassificationStatus":
+        """兼容历史数据，避免枚举转换失败导致500。
+
+        - done → COMPLETED（早期版本用 done 表示完成）
+        - 其他未知值 → PENDING（兜底，不阻塞业务）
+        """
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if normalized in ("done", "success", "finished", "complete"):
+                return cls.COMPLETED
+            if normalized in ("fail", "error", "err"):
+                return cls.FAILED
+            if normalized in ("process", "processing", "running", "run"):
+                return cls.PROCESSING
+        return cls.PENDING
+
 
 @dataclass(slots=True)
 class Asset:

tests/unit/test_asset_status_enum.py

@@ -0,0 +1,72 @@
+"""AssetStatus 枚举兼容性测试。
+
+验证历史脏数据（如 'uploaded'）不会导致枚举转换失败。
+"""
+
+import pytest
+
+from packages.domain.entities import AssetStatus
+
+
+class TestAssetStatusNormalValues:
+    """正常值应该正确映射。"""
+
+    def test_uploading(self):
+        assert AssetStatus("uploading") == AssetStatus.UPLOADING
+
+    def test_ready(self):
+        assert AssetStatus("ready") == AssetStatus.READY
+
+    def test_processing(self):
+        assert AssetStatus("processing") == AssetStatus.PROCESSING
+
+    def test_error(self):
+        assert AssetStatus("error") == AssetStatus.ERROR
+
+
+class TestAssetStatusHistoricalValues:
+    """历史脏数据应该正确映射到对应状态，不抛异常。"""
+
+    @pytest.mark.parametrize("value", ["uploaded", "Uploaded", "UPLOADED", " uploaded "])
+    def test_uploaded_maps_to_ready(self, value):
+        """生产环境发现的 'uploaded' 历史值应映射为 READY。"""
+        assert AssetStatus(value) == AssetStatus.READY
+
+    @pytest.mark.parametrize("value", ["success", "ok", "done", "complete"])
+    def test_other_ready_like_values_map_to_ready(self, value):
+        assert AssetStatus(value) == AssetStatus.READY
+
+    @pytest.mark.parametrize("value", ["upload", "uploading_start", "upload_start"])
+    def test_upload_like_values_map_to_uploading(self, value):
+        assert AssetStatus(value) == AssetStatus.UPLOADING
+
+    @pytest.mark.parametrize("value", ["failed", "fail", "err"])
+    def test_error_like_values_map_to_error(self, value):
+        assert AssetStatus(value) == AssetStatus.ERROR
+
+    @pytest.mark.parametrize("value", ["process", "running", "run"])
+    def test_processing_like_values_map_to_processing(self, value):
+        assert AssetStatus(value) == AssetStatus.PROCESSING
+
+
+class TestAssetStatusFallback:
+    """完全未知的值兜底为 READY，不抛500。"""
+
+    @pytest.mark.parametrize("value", ["unknown", "foo_bar", ""])
+    def test_unknown_value_falls_back_to_ready(self, value):
+        assert AssetStatus(value) == AssetStatus.READY
+
+    def test_none_value_falls_back_to_ready(self):
+        assert AssetStatus(None) == AssetStatus.READY  # type: ignore[arg-type]
+
+    def test_int_value_falls_back_to_ready(self):
+        assert AssetStatus(123) == AssetStatus.READY  # type: ignore[arg-type]
+
+
+class TestAssetStatusStrValue:
+    """枚举值仍为字符串类型，不影响序列化。"""
+
+    def test_value_unchanged(self):
+        assert AssetStatus.READY.value == "ready"
+        assert AssetStatus.ERROR.value == "error"
+        assert isinstance(AssetStatus.READY, str)

tests/unit/test_classification_status_enum.py

@@ -0,0 +1,68 @@
+"""ClassificationStatus 枚举兼容性测试。
+
+验证历史脏数据（如 'done'）不会导致枚举转换失败。
+"""
+
+import pytest
+
+from packages.domain.entities import ClassificationStatus
+
+
+class TestClassificationStatusNormalValues:
+    """正常值应该正确映射。"""
+
+    def test_pending(self):
+        assert ClassificationStatus("pending") == ClassificationStatus.PENDING
+
+    def test_processing(self):
+        assert ClassificationStatus("processing") == ClassificationStatus.PROCESSING
+
+    def test_completed(self):
+        assert ClassificationStatus("completed") == ClassificationStatus.COMPLETED
+
+    def test_failed(self):
+        assert ClassificationStatus("failed") == ClassificationStatus.FAILED
+
+
+class TestClassificationStatusHistoricalValues:
+    """历史脏数据应该正确映射到对应状态，不抛异常。"""
+
+    @pytest.mark.parametrize("value", ["done", "Done", "DONE", " done "])
+    def test_done_maps_to_completed(self, value):
+        """生产环境发现的 'done' 历史值应映射为 COMPLETED。"""
+        assert ClassificationStatus(value) == ClassificationStatus.COMPLETED
+
+    @pytest.mark.parametrize("value", ["success", "finished", "complete"])
+    def test_other_done_like_values_map_to_completed(self, value):
+        assert ClassificationStatus(value) == ClassificationStatus.COMPLETED
+
+    @pytest.mark.parametrize("value", ["fail", "error", "err"])
+    def test_error_like_values_map_to_failed(self, value):
+        assert ClassificationStatus(value) == ClassificationStatus.FAILED
+
+    @pytest.mark.parametrize("value", ["process", "running", "run"])
+    def test_processing_like_values_map_to_processing(self, value):
+        assert ClassificationStatus(value) == ClassificationStatus.PROCESSING
+
+
+class TestClassificationStatusFallback:
+    """完全未知的值兜底为 PENDING，不抛500。"""
+
+    @pytest.mark.parametrize("value", ["unknown", "foo_bar", ""])
+    def test_unknown_value_falls_back_to_pending(self, value):
+        assert ClassificationStatus(value) == ClassificationStatus.PENDING
+
+    def test_none_value_falls_back_to_pending(self):
+        assert ClassificationStatus(None) == ClassificationStatus.PENDING  # type: ignore[arg-type]
+
+    def test_int_value_falls_back_to_pending(self):
+        assert ClassificationStatus(123) == ClassificationStatus.PENDING  # type: ignore[arg-type]
+
+
+class TestClassificationStatusStrValue:
+    """枚举值仍为字符串类型，不影响序列化。"""
+
+    def test_value_unchanged(self):
+        assert ClassificationStatus.COMPLETED.value == "completed"
+        assert ClassificationStatus.PENDING.value == "pending"
+        assert isinstance(ClassificationStatus.COMPLETED, str)

tests/unit/test_oneclick_gen_p0_fixes.py

@@ -132,10 +132,8 @@ class TestDownloadLibraryAssets:
         session.query.return_value = query
         filter_result = MagicMock()
         query.filter.return_value = filter_result
-        in_filter = MagicMock()
-        filter_result.filter.return_value = in_filter
         id_filter = MagicMock()
-        in_filter.filter.return_value = id_filter
+        filter_result.filter.return_value = id_filter
         assets = [self._make_asset("a1", "video/a1.mp4")]
         id_filter.order_by.return_value.all.return_value = assets