fix: validate on_low_score trust-gate policies

CHANGELOG.md

@@ -36,6 +36,8 @@ adheres to [Semantic Versioning](https://semver.org/).
   that difference (#201).
 
 ### Fixed
+- Trust-gate integrations now validate `on_low_score` policies at configuration
+  boundaries, rejecting typos instead of silently skipping failure handling (#345).
 - The minimum supported numpy is now 1.22. The numpy 1.21.6 wheel bundles an
   OpenBLAS that segfaults on BLAS-backed matrix multiplies on current Apple
   Silicon Macs regardless of `OPENBLAS_NUM_THREADS`, so installs at the old

src/freshdata/integrations/_core.py

@@ -17,7 +17,7 @@
 import logging
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING, Any, Literal
+from typing import TYPE_CHECKING, Any, Literal, cast
 
 import freshdata as fd
 from freshdata.enterprise import compute_trust_score
@@ -31,6 +31,19 @@
 
 #: How an adapter should react when the gate does not pass.
 OnLowScore = Literal["warn", "fail", "skip"]
+_VALID_ON_LOW_SCORE = frozenset(("warn", "fail", "skip"))
+
+
+def validate_on_low_score(value: str) -> OnLowScore:
+    """Validate and return an ``on_low_score`` policy.
+
+    Type hints cannot protect configuration loaded from YAML, environment variables,
+    or orchestration frameworks, so validate at the boundary before a gate runs.
+    """
+    if value not in _VALID_ON_LOW_SCORE:
+        allowed = ", ".join(sorted(_VALID_ON_LOW_SCORE))
+        raise ValueError(f"on_low_score must be one of: {allowed}; got {value!r}")
+    return cast(OnLowScore, value)
 
 
 class TrustGateError(RuntimeError):
@@ -185,6 +198,7 @@ def evaluate_trust_gate(
     tuple[pandas.DataFrame, TrustGateResult]
         The cleaned DataFrame and the gate result.
     """
+    on_low_score = validate_on_low_score(on_low_score)
     row_count_in = len(df)
     cleaned, report = fd.clean(df, config=clean_config, report=True)
     trust = compute_trust_score(cleaned)

src/freshdata/integrations/airflow/__init__.py

@@ -18,7 +18,7 @@
 
 from typing import TYPE_CHECKING, Any
 
-from .._core import OnLowScore, evaluate_trust_gate
+from .._core import OnLowScore, evaluate_trust_gate, validate_on_low_score
 
 if TYPE_CHECKING:  # annotations only
     from freshdata import CleanConfig
@@ -67,7 +67,7 @@ def __init__(
             self.output_xcom_key = output_xcom_key
             self.clean_config = clean_config
             self.trust_score_threshold = trust_score_threshold
-            self.on_low_score = on_low_score
+            self.on_low_score = validate_on_low_score(on_low_score)
             self.publish_full_report = publish_full_report
             self.system_actor = system_actor
 

src/freshdata/integrations/dagster/__init__.py

@@ -17,7 +17,7 @@
 import inspect
 from typing import TYPE_CHECKING, Any
 
-from .._core import OnLowScore, TrustGateResult, evaluate_trust_gate
+from .._core import OnLowScore, TrustGateResult, evaluate_trust_gate, validate_on_low_score
 
 if TYPE_CHECKING:  # annotations only
     import pandas as pd
@@ -106,6 +106,7 @@ def freshdata_asset_check(
     blocking:
         Whether a failed check should block downstream materializations.
     """
+    on_low_score = validate_on_low_score(on_low_score)
     dagster = _require_dagster()
     param_name = _asset_param_name(asset)
 
@@ -153,10 +154,11 @@ class FreshDataResource(dagster.ConfigurableResource):  # type: ignore[name-defi
 
         def gate(self, df: pd.DataFrame) -> tuple[pd.DataFrame, TrustGateResult]:
             """Run :func:`evaluate_trust_gate` with this resource's configuration."""
+            on_low_score = validate_on_low_score(self.on_low_score)
             return evaluate_trust_gate(
                 df,
                 trust_score_threshold=self.trust_score_threshold,
-                on_low_score=self.on_low_score,  # type: ignore[arg-type]
+                on_low_score=on_low_score,
                 publish_full_report=self.publish_full_report,
                 system_actor=self.system_actor,
             )

src/freshdata/integrations/dbt/__init__.py

@@ -23,7 +23,13 @@
 from pathlib import Path, PurePath
 from typing import TYPE_CHECKING, Any
 
-from .._core import OnLowScore, TrustGateError, TrustGateResult, evaluate_trust_gate
+from .._core import (
+    OnLowScore,
+    TrustGateError,
+    TrustGateResult,
+    evaluate_trust_gate,
+    validate_on_low_score,
+)
 from .tests_exporter import export_dbt_tests
 
 if TYPE_CHECKING:  # annotations only
@@ -94,6 +100,10 @@ class FreshDataDbtTransform:
     system_actor: str = "freshdata"
     fail_on_low_score: bool = False
 
+    def __post_init__(self) -> None:
+        """Reject invalid gate policies when the transform is configured."""
+        self.on_low_score = validate_on_low_score(self.on_low_score)
+
     def _split_table(self) -> tuple[str | None, str]:
         if self.schema:
             return self.schema, self.model_name
@@ -153,6 +163,7 @@ def gate_manifest(
     is missing) is recorded with an ``"error"`` and counted as failed, so one bad
     model never aborts the whole run.
     """
+    on_low_score = validate_on_low_score(on_low_score)
     manifest = json.loads(Path(manifest_path).read_text())
     nodes = manifest.get("nodes", {})
     models = [n for n in nodes.values() if n.get("resource_type") == "model"]

tests/test_integrations/test_airflow.py

@@ -64,3 +64,10 @@ def test_operator_warn_does_not_raise(fake_airflow, make_ti, sample_df):
     )
     out = op.execute({"ti": make_ti(sample_df)})
     assert out is not None  # warn returns the cleaned frame
+
+
+def test_operator_rejects_invalid_policy(fake_airflow):
+    from freshdata.integrations.airflow import FreshDataCleanOperator
+
+    with pytest.raises(ValueError, match="on_low_score must be one of"):
+        FreshDataCleanOperator(task_id="gate", input_task_id="extract", on_low_score="erro")

tests/test_integrations/test_core.py

@@ -43,6 +43,12 @@ def test_warn_does_not_raise(sample_df):
     assert result.should_skip is False
 
 
+@pytest.mark.parametrize("invalid_policy", ["Fail", "error", ""])
+def test_invalid_low_score_policy_raises_before_cleaning(sample_df, invalid_policy):
+    with pytest.raises(ValueError, match="on_low_score must be one of"):
+        evaluate_trust_gate(sample_df, on_low_score=invalid_policy)
+
+
 def test_as_metadata_keys(sample_df):
     _, result = evaluate_trust_gate(sample_df, trust_score_threshold=0.0)
     meta = result.as_metadata()

tests/test_integrations/test_dagster.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import pytest
+
 from freshdata.integrations.dagster import freshdata_asset_check
 
 
@@ -46,3 +48,8 @@ def test_resource_gate(fake_dagster, sample_df):
     df, result = resource.gate(sample_df)
     assert result.passed is True
     assert len(df) == result.row_count_out
+
+
+def test_asset_check_rejects_invalid_policy(fake_dagster, fake_asset):
+    with pytest.raises(ValueError, match="on_low_score must be one of"):
+        freshdata_asset_check(asset=fake_asset, on_low_score="erro")

tests/test_integrations/test_dbt.py

@@ -82,6 +82,11 @@ def test_transform_no_connection_raises(monkeypatch):
         FreshDataDbtTransform(model_name="orders").run()
 
 
+def test_transform_rejects_invalid_policy():
+    with pytest.raises(ValueError, match="on_low_score must be one of"):
+        FreshDataDbtTransform(model_name="orders", on_low_score="erro")
+
+
 def test_transform_uses_env_conn(warehouse, monkeypatch):
     monkeypatch.setenv("FRESHDATA_WAREHOUSE_CONN", warehouse)
     result = FreshDataDbtTransform(model_name="orders", trust_score_threshold=0.0).run()