diff --git a/packages/domain/entities.py b/packages/domain/entities.py index fcf4a2f70..9a76e46bd 100755 --- a/packages/domain/entities.py +++ b/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" diff --git a/tests/unit/test_asset_status_enum.py b/tests/unit/test_asset_status_enum.py new file mode 100755 index 000000000..ef9db7ab1 --- /dev/null +++ b/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)