fix: skip alerts with unknown Status instead of crashing

elementary/monitor/api/alerts/alert_filters.py

@@ -37,6 +37,18 @@ def _get_alert_node_name(alert: PendingAlertSchema) -> Optional[str]:
     return alert_node_name
 
 
+def _safe_parse_status(raw_status: Optional[str]) -> Optional[Status]:
+    """Parse a raw alert status into a Status enum, or None if it isn't recognized."""
+    try:
+        return Status(raw_status)
+    except ValueError:
+        logger.warning(
+            "Skipping alert with unrecognized status %r (not in Status enum).",
+            raw_status,
+        )
+        return None
+
+
 def apply_filters_schema_on_alert(
     alert: PendingAlertSchema, filters_schema: FiltersSchema
 ) -> bool:
@@ -50,7 +62,9 @@ def apply_filters_schema_on_alert(
         else []
     )
     owners = alert.data.unified_owners or []
-    status = Status(alert.data.status)
+    status = _safe_parse_status(alert.data.status)
+    if status is None:
+        return False
     resource_type = ResourceType(alert.data.resource_type)
     if hasattr(alert.data, "test_unique_id"):
         test_ids = [alert.data.test_unique_id] if alert.data.test_unique_id else []

tests/unit/monitor/api/alerts/test_alert_filters.py

@@ -901,6 +901,57 @@ def test_filter_alerts():
     assert len(filter_model_alerts) == 0
 
 
+def _alert_with_status(alert_id: str, status: str) -> PendingAlertSchema:
+    return PendingAlertSchema(
+        id=alert_id,
+        alert_class_id=alert_id,
+        type=AlertTypes.TEST,
+        detected_at=datetime(2022, 10, 10, 10, 0, 0),
+        created_at=datetime(2022, 10, 10, 10, 0, 0),
+        updated_at=datetime(2022, 10, 10, 10, 0, 0),
+        status=AlertStatus.PENDING,
+        data=TestAlertDataSchema(
+            id=alert_id,
+            alert_class_id=alert_id,
+            model_unique_id="elementary.model_id_1",
+            test_unique_id=alert_id,
+            test_name=alert_id,
+            tags=[],
+            model_meta=dict(owner='["jeff"]'),
+            status=status,
+            elementary_unique_id=f"elementary.model_id_1.{alert_id}.9cf2f5f6ad.None.generic",
+            detected_at=datetime(2022, 10, 10, 10, 0, 0),
+            database_name="test_db",
+            schema_name="test_schema",
+            table_name="table",
+            test_type="dbt_test",
+            test_sub_type="generic",
+            test_results_description="a mock alert",
+            test_results_query="select * from table",
+            test_short_name=alert_id,
+            severity="ERROR",
+            resource_type=ResourceType.TEST,
+        ),
+    )
+
+
+def test_filter_alerts_skips_unknown_status():
+    # 'none' is not a member of the Status enum. The filter must not crash
+    # on it (was raising ValueError) and must drop the alert instead of
+    # letting it bypass the status filter.
+    bad_alert = _alert_with_status("bad_status_alert", "none")
+    good_alert = _alert_with_status("good_alert", "fail")
+
+    # Default filter (FAIL/ERROR/RUNTIME_ERROR/WARN): bad alert dropped, good alert kept.
+    filtered = filter_alerts([bad_alert, good_alert], FiltersSchema())
+    assert [alert.id for alert in filtered] == ["good_alert"]
+
+    # Empty status filter (no status constraint): bad alert still dropped because
+    # we can't classify it; good alert passes.
+    filtered = filter_alerts([bad_alert, good_alert], FiltersSchema(statuses=[]))
+    assert [alert.id for alert in filtered] == ["good_alert"]
+
+
 def test_multi_filters():
     test_alerts, _, _ = initial_alerts()