diff --git a/packages/domain/entities.py b/packages/domain/entities.py old mode 100644 new mode 100755 index 72e2f97fd..fcf4a2f70 --- a/packages/domain/entities.py +++ b/packages/domain/entities.py @@ -141,6 +141,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: diff --git a/tests/unit/test_classification_status_enum.py b/tests/unit/test_classification_status_enum.py new file mode 100755 index 000000000..1249ab241 --- /dev/null +++ b/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)