From bd14b5534c268986af282abbe2258566713c430b Mon Sep 17 00:00:00 2001 From: 1cbyc Date: Mon, 14 Sep 2026 17:12:28 -0400 Subject: [PATCH] fix: validate trust gate policies Reject invalid on_low_score values at integration boundaries so configuration typos cannot silently bypass failure handling. Co-authored-by: insisong --- CHANGELOG.md | 2 ++ src/freshdata/integrations/_core.py | 16 +++++++++++++++- src/freshdata/integrations/airflow/__init__.py | 4 ++-- src/freshdata/integrations/dagster/__init__.py | 6 ++++-- src/freshdata/integrations/dbt/__init__.py | 13 ++++++++++++- tests/test_integrations/test_airflow.py | 7 +++++++ tests/test_integrations/test_core.py | 6 ++++++ tests/test_integrations/test_dagster.py | 7 +++++++ tests/test_integrations/test_dbt.py | 5 +++++ 9 files changed, 60 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8cc50e0d..4769afd0 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/freshdata/integrations/_core.py b/src/freshdata/integrations/_core.py index 6ccdffab..db22e8d9 100644 --- a/src/freshdata/integrations/_core.py +++ b/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) diff --git a/src/freshdata/integrations/airflow/__init__.py b/src/freshdata/integrations/airflow/__init__.py index bf12e862..67d5fea9 100644 --- a/src/freshdata/integrations/airflow/__init__.py +++ b/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 diff --git a/src/freshdata/integrations/dagster/__init__.py b/src/freshdata/integrations/dagster/__init__.py index acf42722..26d8e476 100644 --- a/src/freshdata/integrations/dagster/__init__.py +++ b/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, ) diff --git a/src/freshdata/integrations/dbt/__init__.py b/src/freshdata/integrations/dbt/__init__.py index 1df1c029..f09c2b8b 100644 --- a/src/freshdata/integrations/dbt/__init__.py +++ b/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"] diff --git a/tests/test_integrations/test_airflow.py b/tests/test_integrations/test_airflow.py index fb2ffc35..ba5792ab 100644 --- a/tests/test_integrations/test_airflow.py +++ b/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") diff --git a/tests/test_integrations/test_core.py b/tests/test_integrations/test_core.py index 96d58c92..39146990 100644 --- a/tests/test_integrations/test_core.py +++ b/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() diff --git a/tests/test_integrations/test_dagster.py b/tests/test_integrations/test_dagster.py index 071e2d4d..b09acfbf 100644 --- a/tests/test_integrations/test_dagster.py +++ b/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") diff --git a/tests/test_integrations/test_dbt.py b/tests/test_integrations/test_dbt.py index a7a582b4..f9d0fb77 100644 --- a/tests/test_integrations/test_dbt.py +++ b/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()