Failed rows: fix warn/fail thresholds for fail condition (#2084)

Author: m1n0

soda/core/soda/execution/check/user_defined_failed_rows_expression_check.py

@@ -44,13 +44,22 @@ def __init__(
     def evaluate(self, metrics: Dict[str, Metric], historic_values: Dict[str, object]):
         self.check_value: int = metrics.get(KEY_FAILED_ROWS_COUNT).value
 
-        self.outcome = CheckOutcome.FAIL
-
-        if self.check_value > 0:
+        # Thresholds path
+        if self.check_cfg.fail_threshold_cfg or self.check_cfg.warn_threshold_cfg:
             if self.check_cfg.fail_threshold_cfg and self.check_cfg.fail_threshold_cfg.is_bad(self.check_value):
                 self.outcome = CheckOutcome.FAIL
             elif self.check_cfg.warn_threshold_cfg and self.check_cfg.warn_threshold_cfg.is_bad(self.check_value):
                 self.outcome = CheckOutcome.WARN
+            else:
+                self.outcome = CheckOutcome.PASS
+        else:
+            # Original non-threshold path
+            if self.check_value > 0:
+                self.outcome = CheckOutcome.FAIL
+            else:
+                self.outcome = CheckOutcome.PASS
+
+        if self.check_value > 0:
             # Collect failed rows
             failed_rows_sql = self.get_failed_rows_sql()
             failed_rows_query = UserDefinedFailedRowsExpressionQuery(
@@ -64,8 +73,6 @@ def evaluate(self, metrics: Dict[str, Metric], historic_values: Dict[str, object
             failed_rows_query.execute()
             if failed_rows_query.sample_ref:
                 self.failed_rows_sample_ref = failed_rows_query.sample_ref
-        else:
-            self.outcome = CheckOutcome.PASS
 
     def get_failed_rows_sql(self) -> str:
         columns = ", ".join(

soda/core/tests/data_source/test_user_defined_failed_rows_checks.py

@@ -290,3 +290,39 @@ def test_failed_rows_query_warn_threshold_pass(data_source_fixture: DataSourceFi
     )
     scan.execute()
     scan.assert_check_pass()
+
+
+def test_failed_rows_condition_fail_threshold_pass(data_source_fixture: DataSourceFixture):
+    table_name = data_source_fixture.ensure_test_table(customers_test_table)
+
+    scan = data_source_fixture.create_test_scan()
+    scan.enable_mock_sampler()
+    scan.add_sodacl_yaml_str(
+        f"""
+          checks for {table_name}:
+            - failed rows:
+                name: Customers must have cst_size
+                fail condition: cst_size < 0
+                fail: when > 3
+        """
+    )
+    scan.execute()
+    scan.assert_check_pass()
+
+
+def test_failed_rows_condition_warn_threshold_pass(data_source_fixture: DataSourceFixture):
+    table_name = data_source_fixture.ensure_test_table(customers_test_table)
+
+    scan = data_source_fixture.create_test_scan()
+    scan.enable_mock_sampler()
+    scan.add_sodacl_yaml_str(
+        f"""
+          checks for {table_name}:
+            - failed rows:
+                name: Customers must have cst_size
+                fail condition: cst_size < 0
+                warn: when > 3
+        """
+    )
+    scan.execute()
+    scan.assert_check_pass()