ADAP-850: Support test results as a view

.changes/unreleased/Features-20230915-174428.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support persisting test results as views
+time: 2023-09-15T17:44:28.833877-04:00
+custom:
+  Author: mikealfare
+  Issue: "6914"

core/dbt/contracts/graph/model_config.py

@@ -216,7 +216,6 @@ class Hook(dbtClassMixin, Replaceable):
 
 @dataclass
 class BaseConfig(AdditionalPropertiesAllowed, Replaceable):
-
     # enable syntax like: config['key']
     def __getitem__(self, key):
         return self.get(key)
@@ -560,6 +559,7 @@ class TestConfig(NodeAndTestConfig):
     fail_calc: str = "count(*)"
     warn_if: str = "!= 0"
     error_if: str = "!= 0"
+    strategy: Optional[str] = None
 
     @classmethod
     def same_contents(cls, unrendered: Dict[str, Any], other: Dict[str, Any]) -> bool:
@@ -572,6 +572,7 @@ def same_contents(cls, unrendered: Dict[str, Any], other: Dict[str, Any]) -> boo
             "warn_if",
             "error_if",
             "store_failures",
+            "strategy",
         ]
 
         seen = set()

core/dbt/contracts/graph/nodes.py

@@ -779,7 +779,6 @@
             or enforced_column_constraint_removed
             or materialization_changed
         ):
-
             breaking_changes = []
             if contract_enforced_disabled:
                 breaking_changes.append(
@@ -984,7 +983,9 @@
 class TestShouldStoreFailures:
     @property
     def should_store_failures(self):
-        if self.config.store_failures:
+        if self.config.strategy in ["view", "table"]:
+            return True
+        elif self.config.store_failures:
             return self.config.store_failures
         return get_flags().STORE_FAILURES
 

core/dbt/include/global_project/macros/materializations/configs.sql

@@ -14,7 +14,10 @@
 
 {% macro should_store_failures() %}
   {% set config_store_failures = config.get('store_failures') %}
-  {% if config_store_failures is none %}
+  {% set config_store_failures_strategy = config.get('strategy') %}
+  {% if config_store_failures_strategy in ['view', 'table'] %}
+    {% set config_store_failures = True %}
+  {% elif config_store_failures is none %}
     {% set config_store_failures = flags.STORE_FAILURES %}
   {% endif %}
   {% do return(config_store_failures) %}

core/dbt/include/global_project/macros/materializations/tests/test.sql

@@ -1,8 +1,33 @@
 {%- materialization test, default -%}
 
   {% set relations = [] %}
+    {% set relation_type = config.get('strategy') %}
 
-  {% if should_store_failures() %}
+    {% if relation_type in ['view', 'table'] %}
+
+        {% set identifier = model['alias'] %}
+        {% set existing_relation = adapter.get_relation(database=database, schema=schema, identifier=identifier) %}
+        {% set target_relation = api.Relation.create(
+            database=database, schema=schema, identifier=identifier, type=relation_type
+        ) %}
+
+        {% call statement(auto_begin=True) %}
+            {% if existing_relation %}
+                {{ get_replace_sql(existing_relation, target_relation, sql) }}
+            {% else %}
+                {{ get_create_sql(target_relation, sql) }}
+            {% endif %}
+        {% endcall %}
+
+        {% do relations.append(target_relation) %}
+
+        {% set main_sql %}
+            select * from {{ target_relation }}
+        {% endset %}
+
+        {{ adapter.commit() }}
+
+  {% elif should_store_failures() %}
 
     {% set identifier = model['alias'] %}
     {% set old_relation = adapter.get_relation(database=database, schema=schema, identifier=identifier) %}

core/dbt/include/global_project/macros/relations/create.sql

@@ -6,7 +6,13 @@
 
 {%- macro default__get_create_sql(relation, sql) -%}
 
-    {%- if relation.is_materialized_view -%}
+    {%- if relation.is_view -%}
+        {{ get_create_view_as_sql(relation, sql) }}
+
+    {%- elif relation.is_table -%}
+        {{ get_create_table_as_sql(False, relation, sql) }}
+
+    {%- elif relation.is_materialized_view -%}
         {{ get_create_materialized_view_as_sql(relation, sql) }}
 
     {%- else -%}

tests/adapter/dbt/tests/adapter/persist_test_results/_files.py

@@ -0,0 +1,44 @@
+SEED__CHIPMUNKS = """
+name,shirt
+alvin,red
+simon,blue
+theodore,green
+""".strip()
+
+
+MODEL__CHIPMUNKS = """
+{{ config(materialized='table') }}
+select *
+from {{ ref('chipmunks_stage') }}
+"""
+
+TEST__FAIL_WITH_VIEW_STRATEGY = """
+{{ config(strategy="view") }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__PASS_WITH_VIEW_STRATEGY = """
+{{ config(strategy="view") }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'grape'
+"""
+
+
+TEST__FAIL_WITH_TABLE_STRATEGY = """
+{{ config(strategy="table") }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__PASS_WITH_TABLE_STRATEGY = """
+{{ config(strategy="table") }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'purple'
+"""

tests/adapter/dbt/tests/adapter/persist_test_results/basic.py

@@ -0,0 +1,154 @@
+from collections import namedtuple
+from typing import Dict
+
+import pytest
+
+from dbt.contracts.results import TestStatus
+from dbt.tests.util import run_dbt, check_relation_types
+
+from dbt.tests.adapter.persist_test_results._files import (
+    SEED__CHIPMUNKS,
+    MODEL__CHIPMUNKS,
+    TEST__FAIL_WITH_VIEW_STRATEGY,
+    TEST__PASS_WITH_VIEW_STRATEGY,
+    TEST__FAIL_WITH_TABLE_STRATEGY,
+    TEST__PASS_WITH_TABLE_STRATEGY,
+)
+
+
+class PersistTestResults:
+    seed_table: str = "chipmunks_stage"
+    model_table: str = "chipmunks"
+    audit_schema_suffix: str = "dbt_test__audit"
+
+    audit_schema: str
+
+    @pytest.fixture(scope="class", autouse=True)
+    def setup_class(self, project):
+        # the seed doesn't get touched, load it once
+        run_dbt(["seed"])
+        yield
+
+    @pytest.fixture(scope="function", autouse=True)
+    def setup_method(self, project, setup_class):
+        # make sure the model is always right
+        run_dbt(["run"])
+
+        # the name of the audit schema doesn't change in a class, but this doesn't run at the class level
+        self.audit_schema = f"{project.test_schema}_{self.audit_schema_suffix}"
+        yield
+
+    @pytest.fixture(scope="function", autouse=True)
+    def teardown_method(self, project):
+        yield
+
+        # clear out the audit schema after each test case
+        with project.adapter.connection_named("__test"):
+            audit_schema = project.adapter.Relation.create(
+                database=project.database, schema=self.audit_schema
+            )
+            project.adapter.drop_schema(audit_schema)
+
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {f"{self.seed_table}.csv": SEED__CHIPMUNKS}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {f"{self.model_table}.sql": MODEL__CHIPMUNKS}
+
+    @pytest.fixture(scope="class")
+    def tests(self):
+        return {
+            "fail_with_view_strategy.sql": TEST__FAIL_WITH_VIEW_STRATEGY,
+            "pass_with_view_strategy.sql": TEST__PASS_WITH_VIEW_STRATEGY,
+            "fail_with_table_strategy.sql": TEST__FAIL_WITH_TABLE_STRATEGY,
+            "pass_with_table_strategy.sql": TEST__PASS_WITH_TABLE_STRATEGY,
+        }
+
+    def row_count(self, project, relation_name: str) -> int:
+        """
+        Return the row count for the relation.
+
+        Args:
+            project: the project fixture
+            relation_name: the name of the relation
+
+        Returns:
+            the row count as an integer
+        """
+        sql = f"select count(*) from {self.audit_schema}.{relation_name}"
+        return project.run_sql(sql, fetch="one")[0]
+
+    def insert_record(self, project, record: Dict[str, str]):
+        field_names, field_values = [], []
+        for field_name, field_value in record.items():
+            field_names.append(field_name)
+            field_values.append(f"'{field_value}'")
+        field_name_clause = ", ".join(field_names)
+        field_value_clause = ", ".join(field_values)
+
+        sql = f"""
+        insert into {project.test_schema}.{self.model_table} ({field_name_clause})
+        values ({field_value_clause})
+        """
+        project.run_sql(sql)
+
+    def delete_record(self, project, record: Dict[str, str]):
+        where_clause = " and ".join(
+            [f"{field_name} = '{field_value}'" for field_name, field_value in record.items()]
+        )
+        sql = f"""
+        delete from {project.test_schema}.{self.model_table}
+        where {where_clause}
+        """
+        project.run_sql(sql)
+
+    def test_tests_run_successfully_and_are_persisted_correctly(self, project):
+        # set up the expected results
+        TestResult = namedtuple("TestResult", ["name", "status", "type", "row_count"])
+        expected_results = {
+            TestResult("pass_with_view_strategy", TestStatus.Pass, "view", 0),
+            TestResult("fail_with_view_strategy", TestStatus.Fail, "view", 1),
+            TestResult("pass_with_table_strategy", TestStatus.Pass, "table", 0),
+            TestResult("fail_with_table_strategy", TestStatus.Fail, "table", 1),
+        }
+
+        # run the tests once
+        results = run_dbt(["test"], expect_pass=False)
+
+        # show that the statuses are what we expect
+        actual = {(result.node.name, result.status) for result in results}
+        expected = {(result.name, result.status) for result in expected_results}
+        assert actual == expected
+
+        # show that the results are persisted in the correct database objects
+        check_relation_types(
+            project.adapter, {result.name: result.type for result in expected_results}
+        )
+
+        # show that only the failed records show up
+        actual = {
+            (result.name, self.row_count(project, result.name)) for result in expected_results
+        }
+        expected = {(result.name, result.row_count) for result in expected_results}
+        assert actual == expected
+
+        # insert a new record in the model that fails the "pass" tests
+        # show that the view updates, but not the table
+        self.insert_record(project, {"name": "dave", "shirt": "grape"})
+        expected_results.remove(TestResult("pass_with_view_strategy", TestStatus.Pass, "view", 0))
+        expected_results.add(TestResult("pass_with_view_strategy", TestStatus.Pass, "view", 1))
+
+        # delete the original record from the model that failed the "fail" tests
+        # show that the view updates, but not the table
+        self.delete_record(project, {"name": "theodore", "shirt": "green"})
+        expected_results.remove(TestResult("fail_with_view_strategy", TestStatus.Fail, "view", 1))
+        expected_results.add(TestResult("fail_with_view_strategy", TestStatus.Fail, "view", 0))
+
+        # show that the views update without needing to run dbt, but the tables do not update
+        actual = {
+            (result.name, self.row_count(project, result.name)) for result in expected_results
+        }
+        expected = {(result.name, result.row_count) for result in expected_results}
+        assert actual == expected

tests/functional/artifacts/expected_manifest.py

@@ -132,6 +132,7 @@ def get_rendered_tst_config(**updates):
         "tags": [],
         "severity": "ERROR",
         "store_failures": None,
+        "strategy": None,
         "warn_if": "!= 0",
         "error_if": "!= 0",
         "fail_calc": "count(*)",
@@ -201,7 +202,6 @@ def __str__(self):
 
 
 def expected_seeded_manifest(project, model_database=None, quote_model=False):
-
     model_sql_path = os.path.join("models", "model.sql")
     second_model_sql_path = os.path.join("models", "second_model.sql")
     model_schema_yml_path = os.path.join("models", "schema.yml")

tests/functional/list/test_list.py

@@ -443,6 +443,7 @@ def expect_test_output(self):
                         "materialized": "test",
                         "severity": "ERROR",
                         "store_failures": None,
+                        "strategy": None,
                         "warn_if": "!= 0",
                         "error_if": "!= 0",
                         "fail_calc": "count(*)",
@@ -470,6 +471,7 @@ def expect_test_output(self):
                         "materialized": "test",
                         "severity": "ERROR",
                         "store_failures": None,
+                        "strategy": None,
                         "warn_if": "!= 0",
                         "error_if": "!= 0",
                         "fail_calc": "count(*)",
@@ -500,6 +502,7 @@ def expect_test_output(self):
                         "materialized": "test",
                         "severity": "ERROR",
                         "store_failures": None,
+                        "strategy": None,
                         "warn_if": "!= 0",
                         "error_if": "!= 0",
                         "fail_calc": "count(*)",

tests/functional/persist_test_results/test_persist_test_results.py

@@ -0,0 +1,11 @@
+import pytest
+
+from dbt.tests.adapter.persist_test_results.basic import PersistTestResults
+
+
+class TestPersistTestResults(PersistTestResults):
+    @pytest.fixture(scope="function", autouse=True)
+    def setup_audit_schema(self, project, setup_method):
+        # postgres only supports schema names of 63 characters
+        # a schema with a longer name still gets created, but the name gets truncated
+        self.audit_schema = self.audit_schema[:63]