Add generic test for exposure schema validation

integration_tests/tests/test_exposure_schema_validity.py

@@ -0,0 +1,67 @@
+from datetime import date, timedelta
+from typing import Any, Dict, List
+
+from data_generator import DATE_FORMAT, generate_dates
+from dbt_project import DbtProject
+
+DBT_TEST_NAME = "elementary.exposure_schema_validity"
+
+
+def test_expose_schema_validity_with_no_exposures(test_id: str, dbt_project: DbtProject):
+    test_result = dbt_project.test(test_id, DBT_TEST_NAME, as_model=True)
+    assert test_result["status"] == "pass"
+
+def test_expose_schema_validity_with_missing_columns(test_id: str, dbt_project: DbtProject):
+    DBT_TEST_ARGS = {
+        "exposures": {
+            "ZOMG": {
+                "meta": """{
+                    "columns": [{
+                        "name": "w00t",
+                        "data_type": "string"
+                    }]
+                }""",
+                "depends_on": {
+                    "nodes": []
+                }
+            }
+        }
+    }
+    test_result = dbt_project.test(test_id, DBT_TEST_NAME, DBT_TEST_ARGS, as_model=True)
+    assert test_result["status"] == "fail"
+
+def test_expose_schema_validity_with_correct_columns_and_no_data_type(test_id: str, dbt_project: DbtProject):
+    DBT_TEST_ARGS = {
+        "exposures": {
+            "ZOMG": {
+                "meta": """{
+                    "columns": [{
+                        "name": "w00t",
+                    }]
+                }""",
+                "depends_on": {
+                    "nodes": []
+                }
+            }
+        }
+    }
+    test_result = dbt_project.test(test_id, DBT_TEST_NAME, DBT_TEST_ARGS, as_model=True)
+    assert test_result["status"] == "fail"
+
+def test_expose_schema_validity_with_correct_columns_and_incompatible_data_type(test_id: str, dbt_project: DbtProject):
+    DBT_TEST_ARGS = {
+        "exposures": {
+            "ZOMG": {
+                "meta": """{
+                    "columns": [{
+                        "name": "w00t",
+                    }]
+                }""",
+                "depends_on": {
+                    "nodes": []
+                }
+            }
+        }
+    }
+    test_result = dbt_project.test(test_id, DBT_TEST_NAME, DBT_TEST_ARGS, as_model=True)
+    assert test_result["status"] == "fail"

macros/edr/tests/test_exposure_schema_validity.sql

@@ -0,0 +1,62 @@
+{% test exposure_schema_validity(model, exposures=graph.exposures) %}
+  {%- if not execute -%}
+    {%- do return(none) -%}
+  {%- endif -%}
+  {%- set model_relation = elementary.get_model_relation_for_test(model, context["model"]) -%}
+  {%- set full_table_name = elementary.relation_to_full_name(model_relation) -%}
+  {{- elementary.test_log('start', full_table_name, 'exposure validation') -}}
+  {%- set matching_exposures = [] -%}
+  {%- for exposure in exposures.values() -%}
+      {%- if context['model']['attached_node'] in exposure.depends_on.nodes and exposure['meta'] | default(none) is not none -%}
+          {%- do matching_exposures.append(exposure) -%}
+      {%- endif -%}
+  {%- endfor -%}
+  {%- set matching_exposures_len = matching_exposures | length -%}
+  {%- if matching_exposures_len > 0 -%}
+      {%- set columns_from_model = adapter.get_columns_in_relation(model) -%}
+      {%- if not columns_from_model -%}
+          {%- set table_name = elementary.relation_to_full_name(model) -%}
+          {{- elementary.edr_log('Could not extract columns for table - ' ~ table_name ~ ' (might be a permissions issue)') -}}
+          {{ return(none) }}
+      {%- endif -%}
+      {%- set columns_dict = {} -%}
+      {%- if columns_from_model and columns_from_model is iterable -%}
+          {%- for column in columns_from_model -%}
+              {%- do columns_dict.update({ column['name']: elementary.normalize_data_type(column['dtype']) }) -%}
+          {%- endfor -%}
+      {%- endif -%}
+      {%- set invalid_exposures = [] -%}
+      {%- for exposure in matching_exposures -%}
+          {# Depend on meta since column level info is not available on exposures #}
+          {%- set meta = exposure['meta'] | default(none) -%}
+          {%- if meta != none and meta['columns'] | default(none) is iterable -%}
+              {%- for column in meta['columns'] -%}
+                  {%- if column['name'] | upper not in columns_dict.keys() -%}
+                      {%- do invalid_exposures.append({
+                          'exposure': exposure.name,
+                          'url': exposure.url,
+                          'error': column['name'] ~ ' column missing in the model'
+                          })
+                       -%}
+                  {%- elif column['data_type'] | default('') != '' and elementary.normalize_data_type(column['data_type']) != columns_dict[column['name'] | upper] -%}
+                      {%- do invalid_exposures.append({
+                           'exposure': exposure.name,
+                           'url': exposure.url,
+                           'error': 'different data type for the column ' ~ column['name'] ~ ' ' ~ column['data_type'] ~ ' vs ' ~ columns_dict[column['name']]
+                           })
+                       -%}
+                  {%- endif -%}
+              {%- endfor -%}
+          {%- endif -%}
+      {%- endfor -%}
+      {%- for invalid_exposure in invalid_exposures %}
+        {{ 'UNION ALL ' if not loop.first }}SELECT '{{ invalid_exposure['exposure'] }}' as exposure, '{{ invalid_exposure['url'] }}' as url, '{{ invalid_exposure['error'] }}' as error
+      {%- endfor -%}
+      {{ elementary.test_log('end', full_table_name, 'exposure validation') }}
+      {% if invalid_exposures | length == 0 %}
+          {{ elementary.no_results_query() }}
+      {% endif %}
+    {%- else -%}
+      {{ elementary.no_results_query() }}
+    {%- endif -%}
+{% endtest %}

macros/edr/tests/test_utils/get_test_type.sql

@@ -23,12 +23,15 @@
     {%- set schema_changes_tests = [
         'schema_changes',
         'schema_changes_from_baseline',
-        'json_schema'
+        'json_schema',
     ] %}
     {%- set python_tests = [
         'python',
         'json_schema'
-    ]   %}
+    ] %}
+    {%- set dbt_tests = [
+        'exposure_schema_validity'
+    ] %}
 
   {% if flattened_test.test_namespace == "elementary" %}
     {% if flattened_test.short_name | lower in anomaly_detection_tests %}
@@ -37,6 +40,8 @@
       {% do return("schema_change") %}
     {% elif flattened_test.short_name | lower in python_tests %}
         {% do return("python_test") %}
+    {% elif flattened_test.short_name | lower in dbt_tests %}
+        {% do return("dbt_test") %}
     {% endif %}
   {% endif %}
-{% endmacro %}
+{% endmacro %}