Fix breaking changes in recent dbt-fusion version

.github/workflows/test-warehouse.yml

@@ -116,8 +116,7 @@ jobs:
       - name: Install dbt-fusion
         if: inputs.dbt-version == 'fusion'
         run: |
-          FUSION_VERSION=$(cat dbt-data-reliability/.dbt-fusion-version)
-          curl -fsSL https://public.cdn.getdbt.com/fs/install/install.sh | sh -s -- --version "$FUSION_VERSION"
+          curl -fsSL https://public.cdn.getdbt.com/fs/install/install.sh | sh -s --
 
       - name: Install Elementary
         run: pip install "./elementary[${{ (inputs.warehouse-type == 'databricks_catalog' && 'databricks') || inputs.warehouse-type }}]"

integration_tests/tests/test_collect_metrics.py

@@ -150,7 +150,6 @@ def test_collect_group_by_metrics(test_id: str, dbt_project: DbtProject):
 
 # Anomalies currently not supported on ClickHouse
 @pytest.mark.skip_targets(["clickhouse"])
-@pytest.mark.skip_for_dbt_fusion
 def test_collect_metrics_unique_metric_name(test_id: str, dbt_project: DbtProject):
     args = DBT_TEST_ARGS.copy()
     args["metrics"].append(args["metrics"][0])

integration_tests/tests/test_exposure_schema_validity.py

@@ -1,15 +1,31 @@
+import json
+
 import pytest
 from dbt_project import DbtProject
 
 DBT_TEST_NAME = "elementary.exposure_schema_validity"
 
 
+INVALID_EXPOSURES_QUERY = """
+    with latest_elementary_test_result as (
+        select id
+        from {{{{ ref("elementary_test_results") }}}}
+        where lower(table_name) = lower('{test_id}')
+        order by created_at desc
+        limit 1
+    )
+
+    select result_row
+    from {{{{ ref("test_result_rows") }}}}
+    where elementary_test_results_id in (select * from latest_elementary_test_result)
+"""
+
+
 def seed(dbt_project: DbtProject):
     seed_result = dbt_project.dbt_runner.seed(full_refresh=True)
     assert seed_result is True
 
 
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_existing_exposure_yml_invalid(
     test_id: str, dbt_project: DbtProject
 ):
@@ -28,7 +44,6 @@ def test_exposure_schema_validity_existing_exposure_yml_invalid(
     assert test_result.success is False
 
 
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_existing_exposure_yml_valid(
     test_id: str, dbt_project: DbtProject
 ):
@@ -47,15 +62,13 @@ def test_exposure_schema_validity_existing_exposure_yml_valid(
 
 
 @pytest.mark.skip_targets(["spark"])
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_no_exposures(test_id: str, dbt_project: DbtProject):
     test_result = dbt_project.test(test_id, DBT_TEST_NAME)
     assert test_result["status"] == "pass"
 
 
 # Schema validity currently not supported on ClickHouse
 @pytest.mark.skip_targets(["spark", "clickhouse"])
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_correct_columns_and_types(
     test_id: str, dbt_project: DbtProject
 ):
@@ -88,7 +101,6 @@ def test_exposure_schema_validity_correct_columns_and_types(
 
 
 @pytest.mark.skip_targets(["spark"])
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_correct_columns_and_invalid_type(
     test_id: str, dbt_project: DbtProject
 ):
@@ -111,17 +123,25 @@ def test_exposure_schema_validity_correct_columns_and_invalid_type(
     test_result = dbt_project.test(
         test_id, DBT_TEST_NAME, DBT_TEST_ARGS, columns=[dict(name="bla")], as_model=True
     )
+    assert test_result["status"] == "fail"
 
+    invalid_exposures = [
+        json.loads(row["result_row"])
+        for row in dbt_project.run_query(
+            INVALID_EXPOSURES_QUERY.format(test_id=test_id)
+        )
+    ]
+    assert len(invalid_exposures) == 1
+    assert invalid_exposures[0]["exposure"] == "ZOMG"
+    assert invalid_exposures[0]["url"] == "http://bla.com"
     assert (
-        "different data type for the column order_id string vs"
-        in test_result["test_results_query"]
+        invalid_exposures[0]["error"]
+        == "different data type for the column order_id string vs numeric"
     )
-    assert test_result["status"] == "fail"
 
 
 # Schema validity currently not supported on ClickHouse
 @pytest.mark.skip_targets(["spark", "clickhouse"])
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_invalid_type_name_present_in_error(
     test_id: str, dbt_project: DbtProject
 ):
@@ -155,16 +175,24 @@ def test_exposure_schema_validity_invalid_type_name_present_in_error(
     test_result = dbt_project.test(
         test_id, DBT_TEST_NAME, DBT_TEST_ARGS, columns=[dict(name="bla")], as_model=True
     )
+    assert test_result["status"] == "fail"
 
+    invalid_exposures = [
+        json.loads(row["result_row"])
+        for row in dbt_project.run_query(
+            INVALID_EXPOSURES_QUERY.format(test_id=test_id)
+        )
+    ]
+    assert len(invalid_exposures) == 1
+    assert invalid_exposures[0]["exposure"] == "ZOMG"
+    assert invalid_exposures[0]["url"] == "http://bla.com"
     assert (
-        "different data type for the column order_id string vs numeric"
-        in test_result["test_results_query"]
+        invalid_exposures[0]["error"]
+        == "different data type for the column order_id string vs numeric"
     )
-    assert test_result["status"] == "fail"
 
 
 @pytest.mark.skip_targets(["spark"])
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_correct_columns_and_missing_type(
     test_id: str, dbt_project: DbtProject
 ):
@@ -188,7 +216,6 @@ def test_exposure_schema_validity_correct_columns_and_missing_type(
 
 
 @pytest.mark.skip_targets(["spark"])
-@pytest.mark.skip_for_dbt_fusion
 def test_exposure_schema_validity_missing_columns(
     test_id: str, dbt_project: DbtProject
 ):
@@ -211,6 +238,15 @@ def test_exposure_schema_validity_missing_columns(
     test_result = dbt_project.test(
         test_id, DBT_TEST_NAME, DBT_TEST_ARGS, columns=[dict(name="bla")], as_model=True
     )
-
-    assert "order_id column missing in the model" in test_result["test_results_query"]
     assert test_result["status"] == "fail"
+
+    invalid_exposures = [
+        json.loads(row["result_row"])
+        for row in dbt_project.run_query(
+            INVALID_EXPOSURES_QUERY.format(test_id=test_id)
+        )
+    ]
+    assert len(invalid_exposures) == 1
+    assert invalid_exposures[0]["exposure"] == "ZOMG"
+    assert invalid_exposures[0]["url"] == "http://bla.com"
+    assert invalid_exposures[0]["error"] == "order_id column missing in the model"

integration_tests/tests/test_failed_row_count.py

@@ -6,7 +6,6 @@
 
 # Failed row count currently not supported on ClickHouse
 @pytest.mark.skip_targets(["clickhouse"])
-@pytest.mark.skip_for_dbt_fusion
 def test_count_failed_row_count(test_id: str, dbt_project: DbtProject):
     null_count = 50
     data = [{COLUMN_NAME: None} for _ in range(null_count)]
@@ -24,7 +23,6 @@ def test_count_failed_row_count(test_id: str, dbt_project: DbtProject):
     )  # when the failed_row_count_calc is count(*), these should be equal
 
 
-@pytest.mark.skip_for_dbt_fusion
 def test_sum_failed_row_count(test_id: str, dbt_project: DbtProject):
     non_unique_count = 50
     data = [{COLUMN_NAME: 5} for _ in range(non_unique_count)]
@@ -44,7 +42,6 @@ def test_sum_failed_row_count(test_id: str, dbt_project: DbtProject):
 
 # Failed row count currently not supported on ClickHouse
 @pytest.mark.skip_targets(["clickhouse"])
-@pytest.mark.skip_for_dbt_fusion
 def test_custom_failed_row_count(test_id: str, dbt_project: DbtProject):
     null_count = 50
     overwrite_failed_row_count = 5
@@ -64,7 +61,6 @@ def test_custom_failed_row_count(test_id: str, dbt_project: DbtProject):
     assert test_result["failed_row_count"] == overwrite_failed_row_count
 
 
-@pytest.mark.skip_for_dbt_fusion
 def test_warn_if_0(test_id: str, dbt_project: DbtProject):
     # Edge case that we want to verify
 

integration_tests/tests/test_override_samples_config.py

@@ -35,7 +35,7 @@ def test_sample_count_unlimited(test_id: str, dbt_project: DbtProject):
             "enable_elementary_test_materialization": True,
             "test_sample_row_count": 5,
         },
-        test_config={"meta": {"test_sample_row_count": None}},
+        test_config={"meta": {"test_sample_row_count": -1}},
     )
     assert test_result["status"] == "fail"
 

integration_tests/tests/test_schema_changes.py

@@ -42,7 +42,9 @@ def assert_test_results(test_results: List[dict]):
 
 
 # Schema changes currently not supported on targets
+# dbt-fusion caches column information and doesn't refresh when tables are recreated
 @pytest.mark.skip_targets(["databricks", "spark", "athena", "trino", "clickhouse"])
+@pytest.mark.skip_for_dbt_fusion
 def test_schema_changes(test_id: str, dbt_project: DbtProject):
     dbt_test_name = "elementary.schema_changes"
     test_result = dbt_project.test(test_id, dbt_test_name, data=DATASET1)

macros/edr/dbt_artifacts/upload_dbt_invocation.sql

@@ -87,12 +87,8 @@
 
 {%- macro get_invocation_select_filter() -%}
     {% set config = elementary.get_runtime_config() %}
-    {%- if invocation_args_dict -%}
-        {%- if invocation_args_dict.select -%}
-            {{- return(invocation_args_dict.select) -}}
-        {%- elif invocation_args_dict.SELECT -%}
-            {{- return(invocation_args_dict.SELECT) -}}
-        {%- endif -%}
+    {%- if invocation_args_dict and invocation_args_dict.select -%}
+        {{- return(invocation_args_dict.select) -}}
     {%- elif config.args and config.args.select -%}
         {{- return(config.args.select) -}}
     {%- else -%}
@@ -106,10 +102,10 @@
         {% do return(invocation_args_dict.selector) %}
     {% elif invocation_args_dict and invocation_args_dict.selector_name %}
         {% do return(invocation_args_dict.selector_name) %}
-    {% elif invocation_args_dict and invocation_args_dict.INVOCATION_COMMAND %}
+    {% elif invocation_args_dict and invocation_args_dict.invocation_command %}
         {% set match = modules.re.search(
             "--selector(?:\s+|=)(\S+)",
-            invocation_args_dict.INVOCATION_COMMAND
+            invocation_args_dict.invocation_command
         ) %}
         {% if match %}
             {% do return(match.group(1)) %}
@@ -132,8 +128,6 @@
         {% else %}
             {% set invocation_vars = fromyaml(invocation_args_dict.vars) %}
         {% endif %}
-    {% elif invocation_args_dict and invocation_args_dict.VARS %}
-        {% set invocation_vars = invocation_args_dict.VARS %}
     {% elif config.cli_vars %}
         {% set invocation_vars = config.cli_vars %}
     {% endif %}

macros/edr/dbt_artifacts/upload_dbt_tests.sql

@@ -55,25 +55,20 @@
 
     {% set default_description = elementary.get_default_description(test_original_name, test_namespace) %}
 
-    {% set config_meta_dict = elementary.safe_get_with_default(config_dict, 'meta', {}) %}
-    {% set meta_dict = elementary.safe_get_with_default(node_dict, 'meta', {}) %}
-
-    {% set unified_meta = {} %}
-    {% do unified_meta.update(config_meta_dict) %}
-    {% do unified_meta.update(meta_dict) %}
+    {% set meta = elementary.get_node_meta(node_dict) %}
 
     {% set description = none %}
     {% if dbt_version >= '1.9.0' and node_dict.get('description') %}
         {% set description = node_dict.get('description') %}
-    {% elif unified_meta.get('description') %}
-        {% set description = unified_meta.pop('description') %}
+    {% elif meta.get('description') %}
+        {% set description = meta.pop('description') %}
     {% elif default_description %}
         {% set description = default_description %}
     {% endif %}
 
     {% set config_tags = elementary.safe_get_with_default(config_dict, 'tags', []) %}
     {% set global_tags = elementary.safe_get_with_default(node_dict, 'tags', []) %}
-    {% set meta_tags = elementary.safe_get_with_default(unified_meta, 'tags', []) %}
+    {% set meta_tags = elementary.safe_get_with_default(meta, 'tags', []) %}
     {% set tags = elementary.union_lists(config_tags, global_tags) %}
     {% set tags = elementary.union_lists(tags, meta_tags) %}
 
@@ -167,7 +162,7 @@
         'tags': elementary.filter_none_and_sort(tags),
         'model_tags': elementary.filter_none_and_sort(test_models_tags),
         'model_owners': elementary.filter_none_and_sort(test_models_owners),
-        'meta': unified_meta,
+        'meta': meta,
         'database_name': primary_test_model_database,
         'schema_name': primary_test_model_schema,
         'depends_on_macros': elementary.filter_none_and_sort(depends_on_dict.get('macros', [])),
@@ -181,7 +176,7 @@
         'compiled_code': elementary.get_compiled_code(node_dict),
         'path': node_dict.get('path'),
         'generated_at': elementary.datetime_now_utc_as_string(),
-        'quality_dimension': unified_meta.get('quality_dimension') or elementary.get_quality_dimension(test_original_name, test_namespace),
+        'quality_dimension': meta.get('quality_dimension') or elementary.get_quality_dimension(test_original_name, test_namespace),
         'group_name': group_name,
     }%}
     {% do flatten_test_metadata_dict.update({"metadata_hash": elementary.get_artifact_metadata_hash(flatten_test_metadata_dict)}) %}

macros/edr/materializations/test/test.sql

@@ -124,6 +124,12 @@
   {% if sample_limit == 0 %} {# performance: no need to run a sql query that we know returns an empty list #}
     {% do return([]) %}
   {% endif %}
+
+  {# Allow setting -1 for unlimited, as none values are stripped from meta in dbt-fusion #}
+  {% if sample_limit == -1 %}
+    {% set sample_limit = none %}
+  {% endif %}
+
   {% if ignore_passed_tests and elementary.did_test_pass() %}
     {% do elementary.debug_log("Skipping sample query because the test passed.") %}
     {% do return([]) %}

macros/edr/tests/on_run_end/handle_tests_results.sql

@@ -37,16 +37,6 @@
     {% set status = "pass" %}
   {% endif %}
 
-  {% if elementary.is_dbt_fusion() %}
-    {% if status == 'error' %}
-      {# dbt-fusion currently doesn't distinguish between failure and error #}
-      {% set status = "fail" %}
-    {% elif status == 'success' %}
-      {# dbt-fusion seems to sometime return 'pass' and sometimes 'success', so we normalize to 'pass' #}
-      {% set status = "pass" %}
-    {% endif %}
-  {% endif %}
-
   {% do return(status) %}
 {% endmacro %}
 

macros/edr/tests/test_configuration/get_anomalies_test_configuration.sql

@@ -50,7 +50,10 @@
     {%- set detection_delay = elementary.get_detection_delay(detection_delay, model_graph_node) %}
 
     {%- set ignore_small_changes = elementary.get_test_argument('ignore_small_changes', ignore_small_changes, model_graph_node) %}
-    {# Validate ignore_small_changes #}
+    {# Normalize ignore_small_changes to ensure it's always a dict with expected keys #}
+    {%- if ignore_small_changes is none %}
+        {%- set ignore_small_changes = {"spike_failure_percent_threshold": none, "drop_failure_percent_threshold": none} %}
+    {%- endif %}
 
     {% set anomaly_exclude_metrics = elementary.get_test_argument('anomaly_exclude_metrics', anomaly_exclude_metrics, model_graph_node) %}
     {% set exclude_final_results = elementary.get_exclude_final_results(exclude_final_results) %}
@@ -109,6 +112,9 @@
 {% endmacro %}
 
 {% macro validate_ignore_small_changes(test_configuration) %}
+    {% if test_configuration.ignore_small_changes is none %}
+        {% do return(none) %}
+    {% endif %}
     {% for key, value in test_configuration.ignore_small_changes.items() %}
         {% if key not in ['spike_failure_percent_threshold', 'drop_failure_percent_threshold'] %}
           {% do exceptions.raise_compiler_error('Illegal configuration key: {}'.format(key)) %}