include PR feedback

Author: Marthe Van Den Hende

airflow_dbt_python/utils/configs.py

@@ -619,6 +619,11 @@ def __post_init__(self):
         """Call superclass __post_init__."""
         super().__post_init__()
         if isinstance(self.full_refresh, str):
+            if self.full_refresh.lower() not in ("true", "false"):
+                raise ValueError(
+                    f"Invalid value for full_refresh: '{self.full_refresh}'. "
+                    "Expected 'true' or 'false'."
+                )
             self.full_refresh = self.full_refresh.lower() == "true"
 
 

tests/operators/test_dbt_build.py

@@ -242,7 +242,12 @@ def test_full_refresh_in_template_fields():
 
 
 def test_full_refresh_templated():
-    """Test that full_refresh can be templated with Jinja."""
+    """Test that full_refresh can be templated with Jinja.
+
+    Airflow renders template fields as strings, so the operator receives a
+    string after rendering. The string-to-bool coercion happens in
+    TableMutabilityConfig and is tested in test_configs.py.
+    """
     import pendulum
     from airflow.models.dag import DAG
 

tests/operators/test_dbt_compile.py

@@ -214,7 +214,12 @@ def test_full_refresh_in_template_fields():
 
 
 def test_full_refresh_templated():
-    """Test that full_refresh can be templated with Jinja."""
+    """Test that full_refresh can be templated with Jinja.
+
+    Airflow renders template fields as strings, so the operator receives a
+    string after rendering. The string-to-bool coercion happens in
+    TableMutabilityConfig and is tested in test_configs.py.
+    """
     import pendulum
     from airflow.models.dag import DAG
 

tests/operators/test_dbt_run.py

@@ -436,7 +436,12 @@ def test_full_refresh_in_template_fields():
 
 
 def test_full_refresh_templated():
-    """Test that full_refresh can be templated with Jinja."""
+    """Test that full_refresh can be templated with Jinja.
+
+    Airflow renders template fields as strings, so the operator receives a
+    string after rendering. The string-to-bool coercion happens in
+    TableMutabilityConfig and is tested in test_configs.py.
+    """
     from airflow.models.dag import DAG
 
     dag = DAG(dag_id="test_dag", start_date=pendulum.datetime(2025, 1, 1))

tests/operators/test_dbt_seed.py

@@ -335,7 +335,12 @@ def test_full_refresh_in_template_fields():
 
 
 def test_full_refresh_templated():
-    """Test that full_refresh can be templated with Jinja."""
+    """Test that full_refresh can be templated with Jinja.
+
+    Airflow renders template fields as strings, so the operator receives a
+    string after rendering. The string-to-bool coercion happens in
+    TableMutabilityConfig and is tested in test_configs.py.
+    """
     import pendulum
     from airflow.models.dag import DAG
 

tests/utils/test_configs.py

@@ -508,34 +508,37 @@ def test_base_config_does_not_override_when_value_in_environment(
     assert config.require_generic_test_arguments_property is None
 
 
-def test_table_mutability_config_full_refresh_string_coercion(
-    profiles_file, dbt_project_file
+@pytest.mark.parametrize(
+    "input_value,expected",
+    [
+        ("True", True),
+        ("true", True),
+        ("False", False),
+        ("false", False),
+        (True, True),
+        (False, False),
+        (None, None),
+    ],
+)
+def test_table_mutability_config_full_refresh_coercion(
+    profiles_file, dbt_project_file, input_value, expected
 ):
-    """Test that full_refresh string values are coerced to booleans."""
+    """Test that full_refresh values are coerced to booleans."""
     config = RunTaskConfig(
         profiles_dir=profiles_file.parent,
         project_dir=dbt_project_file.parent,
-        full_refresh="True",
+        full_refresh=input_value,
     )
-    assert config.full_refresh is True
+    assert config.full_refresh is expected
 
-    config = RunTaskConfig(
-        profiles_dir=profiles_file.parent,
-        project_dir=dbt_project_file.parent,
-        full_refresh="False",
-    )
-    assert config.full_refresh is False
 
-    config = RunTaskConfig(
-        profiles_dir=profiles_file.parent,
-        project_dir=dbt_project_file.parent,
-        full_refresh=None,
-    )
-    assert config.full_refresh is None
-
-    config = RunTaskConfig(
-        profiles_dir=profiles_file.parent,
-        project_dir=dbt_project_file.parent,
-        full_refresh=True,
-    )
-    assert config.full_refresh is True
+def test_table_mutability_config_full_refresh_invalid_string(
+    profiles_file, dbt_project_file
+):
+    """Test that invalid full_refresh strings raise a ValueError."""
+    with pytest.raises(ValueError, match="Invalid value for full_refresh"):
+        RunTaskConfig(
+            profiles_dir=profiles_file.parent,
+            project_dir=dbt_project_file.parent,
+            full_refresh="treu",
+        )