allow for bool/str input to backup/autorefresh when configuring mater…

…ialized views

dbt/adapters/redshift/relation_configs/materialized_view.py

@@ -23,6 +23,7 @@
     RedshiftSortConfig,
     RedshiftSortConfigChange,
 )
+from dbt.adapters.redshift.utility import evaluate_bool
 
 
 @dataclass(frozen=True, eq=True, unsafe_hash=True)
@@ -122,25 +123,16 @@ def parse_model_node(cls, model_node: ModelNode) -> dict:
             "mv_name": model_node.identifier,
             "schema_name": model_node.schema,
             "database_name": model_node.database,
-            "backup": model_node.config.extra.get("backup"),
         }
 
+        # backup/autorefresh can be bools or strings
+        backup_value = model_node.config.extra.get("backup")
+        if backup_value is not None:
+            config_dict["backup"] = evaluate_bool(backup_value)
+
         autorefresh_value = model_node.config.extra.get("auto_refresh")
         if autorefresh_value is not None:
-            if isinstance(autorefresh_value, bool):
-                config_dict["autorefresh"] = autorefresh_value
-            elif isinstance(autorefresh_value, str):
-                lower_autorefresh = autorefresh_value.lower()
-                if lower_autorefresh == "true":
-                    config_dict["autorefresh"] = True
-                elif lower_autorefresh == "false":
-                    config_dict["autorefresh"] = False
-                else:
-                    raise ValueError(
-                        "Invalid autorefresh representation. Please use accepted value ex.( True, 'true', 'True')"
-                    )
-            else:
-                raise TypeError("Invalid autorefresh value: expecting boolean or str.")
+            config_dict["autorefresh"] = evaluate_bool(autorefresh_value)
 
         if query := model_node.compiled_code:
             config_dict.update({"query": query.strip()})

dbt/adapters/redshift/utility.py

@@ -0,0 +1,25 @@
+from typing import Union
+
+
+def evaluate_bool_str(value: str) -> bool:
+    value = value.strip().lower()
+    if value == "true":
+        return True
+    elif value == "false":
+        return False
+    else:
+        raise ValueError(f"Invalid boolean string value: {value}")
+
+
+def evaluate_bool(value: Union[str, bool]) -> bool:
+    if not value:
+        return False
+    if isinstance(value, bool):
+        return value
+    elif isinstance(value, str):
+        return evaluate_bool_str(value)
+    else:
+        raise TypeError(
+            f"Invalid type for boolean evaluation, "
+            f"expecting boolean or str, recieved: {type(value)}"
+        )

tests/functional/adapter/materialized_view_tests/test_materialized_views.py

@@ -11,8 +11,13 @@
     MaterializedViewChangesContinueMixin,
     MaterializedViewChangesFailMixin,
 )
-from dbt.tests.adapter.materialized_view.files import MY_TABLE, MY_VIEW
-from dbt.tests.util import assert_message_in_logs, get_model_file, set_model_file
+from dbt.tests.adapter.materialized_view.files import MY_TABLE, MY_VIEW, MY_SEED
+from dbt.tests.util import (
+    assert_message_in_logs,
+    get_model_file,
+    set_model_file,
+    run_dbt,
+)
 
 from tests.functional.adapter.materialized_view_tests.utils import (
     query_autorefresh,
@@ -22,7 +27,6 @@
     run_dbt_and_capture_with_retries_redshift_mv,
 )
 
-
 MY_MATERIALIZED_VIEW = """
 {{ config(
     materialized='materialized_view',
@@ -233,3 +237,28 @@ class TestRedshiftMaterializedViewChangesFail(
 ):
     # Note: using retries doesn't work when we expect `dbt_run` to fail
     pass
+
+
+NO_BACKUP_MATERIALIZED_VIEW = """
+{{ config(
+    materialized='materialized_view',
+    backup=False
+) }}
+select * from {{ ref('my_seed') }}
+"""
+
+
+class TestRedshiftMaterializedViewWithBackupConfig:
+    @pytest.fixture(scope="class", autouse=True)
+    def models(self):
+        yield {
+            "my_materialized_view.sql": NO_BACKUP_MATERIALIZED_VIEW,
+        }
+
+    @pytest.fixture(scope="class", autouse=True)
+    def seeds(self):
+        return {"my_seed.csv": MY_SEED}
+
+    def test_running_mv_with_backup_false_succeeds(self, project):
+        run_dbt(["seed"])
+        run_dbt(["run"])

tests/unit/relation_configs/test_materialized_view.py

@@ -0,0 +1,55 @@
+from unittest.mock import Mock
+
+import pytest
+
+from dbt.adapters.redshift.relation_configs import RedshiftMaterializedViewConfig
+
+
+@pytest.mark.parametrize("bool_value", [True, False, "True", "False", "true", "false"])
+def test_redshift_materialized_view_config_handles_all_valid_bools(bool_value):
+    config = RedshiftMaterializedViewConfig(
+        database_name="somedb",
+        schema_name="public",
+        mv_name="someview",
+        query="select * from sometable",
+    )
+    model_node = Mock()
+    model_node.config.extra.get = (
+        lambda x, y=None: bool_value if x in ["auto_refresh", "backup"] else "someDistValue"
+    )
+    config_dict = config.parse_model_node(model_node)
+    assert isinstance(config_dict["autorefresh"], bool)
+    assert isinstance(config_dict["backup"], bool)
+
+
+@pytest.mark.parametrize("bool_value", [1])
+def test_redshift_materialized_view_config_throws_expected_exception_with_invalid_types(
+    bool_value,
+):
+    config = RedshiftMaterializedViewConfig(
+        database_name="somedb",
+        schema_name="public",
+        mv_name="someview",
+        query="select * from sometable",
+    )
+    model_node = Mock()
+    model_node.config.extra.get = (
+        lambda x, y=None: bool_value if x in ["auto_refresh", "backup"] else "someDistValue"
+    )
+    with pytest.raises(TypeError):
+        config.parse_model_node(model_node)
+
+
+def test_redshift_materialized_view_config_throws_expected_exception_with_invalid_str():
+    config = RedshiftMaterializedViewConfig(
+        database_name="somedb",
+        schema_name="public",
+        mv_name="someview",
+        query="select * from sometable",
+    )
+    model_node = Mock()
+    model_node.config.extra.get = (
+        lambda x, y=None: "notABool" if x in ["auto_refresh", "backup"] else "someDistValue"
+    )
+    with pytest.raises(ValueError):
+        config.parse_model_node(model_node)