Add store_failures_as config for generic tests

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

@@ -0,0 +1,6 @@
+kind: Features
+body: Support storing test failures 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 @@ T = TypeVar("T", bound="BaseConfig")
 
 @dataclass
 class BaseConfig(AdditionalPropertiesAllowed, Replaceable):
-
     # enable syntax like: config['key']
     def __getitem__(self, key):
         return self.get(key)
@@ -555,6 +554,7 @@ class TestConfig(NodeAndTestConfig):
     # Annotated is used by mashumaro for jsonschema generation
     severity: Annotated[Severity, Pattern(SEVERITY_PATTERN)] = Severity("ERROR")
     store_failures: Optional[bool] = None
+    store_failures_as: Optional[str] = "table"
     where: Optional[str] = None
     limit: Optional[int] = None
     fail_calc: str = "count(*)"
@@ -572,6 +572,7 @@ class TestConfig(NodeAndTestConfig):
             "warn_if",
             "error_if",
             "store_failures",
+            "store_failures_as",
         ]
 
         seen = set()

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

@@ -6,15 +6,17 @@
 
     {% set identifier = model['alias'] %}
     {% set old_relation = adapter.get_relation(database=database, schema=schema, identifier=identifier) %}
+
+    {% set store_failures_as = config.get('store_failures_as') %}
     {% set target_relation = api.Relation.create(
-        identifier=identifier, schema=schema, database=database, type='table') -%} %}
+        identifier=identifier, schema=schema, database=database, type=store_failures_as) -%} %}
 
     {% if old_relation %}
         {% do adapter.drop_relation(old_relation) %}
     {% endif %}
 
     {% call statement(auto_begin=True) %}
-        {{ create_table_as(False, target_relation, sql) }}
+        {{ get_create_sql(target_relation, sql) }}
     {% endcall %}
 
     {% do relations.append(target_relation) %}

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 -%}

core/dbt/parser/generic_test_builders.py

@@ -101,6 +101,7 @@ class TestBuilder(Generic[Testable]):
         "error_if",
         "fail_calc",
         "store_failures",
+        "store_failures_as",
         "meta",
         "database",
         "schema",
@@ -242,6 +243,10 @@ class TestBuilder(Generic[Testable]):
     def store_failures(self) -> Optional[bool]:
         return self.config.get("store_failures")
 
+    @property
+    def store_failures_as(self) -> Optional[bool]:
+        return self.config.get("store_failures_as")
+
     @property
     def where(self) -> Optional[str]:
         return self.config.get("where")
@@ -294,6 +299,8 @@ class TestBuilder(Generic[Testable]):
             config["fail_calc"] = self.fail_calc
         if self.store_failures is not None:
             config["store_failures"] = self.store_failures
+        if self.store_failures_as is not None:
+            config["store_failures_as"] = self.store_failures_as
         if self.meta is not None:
             config["meta"] = self.meta
         if self.database is not None:

tests/adapter/dbt/tests/adapter/store_test_failures_tests/_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_AS_VIEW = """
+{{ config(store_failures_as="view") }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__PASS_AS_VIEW = """
+{{ config(store_failures_as="view") }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'grape'
+"""
+
+
+TEST__FAIL_AS_TABLE = """
+{{ config(store_failures_as="table") }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'green'
+"""
+
+
+TEST__PASS_AS_TABLE = """
+{{ config(store_failures_as="table") }}
+select *
+from {{ ref('chipmunks') }}
+where shirt = 'purple'
+"""

tests/adapter/dbt/tests/adapter/store_test_failures_tests/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.store_test_failures_tests._files import (
+    SEED__CHIPMUNKS,
+    MODEL__CHIPMUNKS,
+    TEST__FAIL_AS_VIEW,
+    TEST__PASS_AS_VIEW,
+    TEST__FAIL_AS_TABLE,
+    TEST__PASS_AS_TABLE,
+)
+
+
+class StoreTestFailures:
+    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_as_view.sql": TEST__FAIL_AS_VIEW,
+            "pass_as_view.sql": TEST__PASS_AS_VIEW,
+            "fail_as_table.sql": TEST__FAIL_AS_TABLE,
+            "pass_as_table.sql": TEST__PASS_AS_TABLE,
+        }
+
+    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_stored_as_expected(self, project):
+        # set up the expected results
+        TestResult = namedtuple("TestResult", ["name", "status", "type", "row_count"])
+        expected_results = {
+            TestResult("pass_as_view", TestStatus.Pass, "view", 0),
+            TestResult("fail_as_view", TestStatus.Fail, "view", 1),
+            TestResult("pass_as_table", TestStatus.Pass, "table", 0),
+            TestResult("fail_as_table", TestStatus.Fail, "table", 1),
+        }
+
+        # run the tests once
+        results = run_dbt(["test", "--store-failures"], 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_as_view", TestStatus.Pass, "view", 0))
+        expected_results.add(TestResult("pass_as_view", 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_as_view", TestStatus.Fail, "view", 1))
+        expected_results.add(TestResult("fail_as_view", 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,
+        "store_failures_as": "table",
         "warn_if": "!= 0",
         "error_if": "!= 0",
         "fail_calc": "count(*)",

tests/functional/list/test_list.py

@@ -493,6 +493,7 @@ class TestList:
                         "materialized": "test",
                         "severity": "ERROR",
                         "store_failures": None,
+                        "store_failures_as": "table",
                         "warn_if": "!= 0",
                         "error_if": "!= 0",
                         "fail_calc": "count(*)",
@@ -520,6 +521,7 @@ class TestList:
                         "materialized": "test",
                         "severity": "ERROR",
                         "store_failures": None,
+                        "store_failures_as": "table",
                         "warn_if": "!= 0",
                         "error_if": "!= 0",
                         "fail_calc": "count(*)",
@@ -550,6 +552,7 @@ class TestList:
                         "materialized": "test",
                         "severity": "ERROR",
                         "store_failures": None,
+                        "store_failures_as": "table",
                         "warn_if": "!= 0",
                         "error_if": "!= 0",
                         "fail_calc": "count(*)",

tests/functional/store_test_failures/test_store_test_failures.py

@@ -0,0 +1,11 @@
+import pytest
+
+from dbt.tests.adapter.store_test_failures_tests.basic import StoreTestFailures
+
+
+class TestStoreTestFailures(StoreTestFailures):
+    @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]

tests/unit/test_contracts_graph_compiled.py

@@ -544,6 +544,7 @@ def basic_compiled_schema_test_dict():
         "config": {
             "enabled": True,
             "materialized": "test",
+            "store_failures_as": "table",
             "tags": [],
             "severity": "warn",
             "schema": "dbt_test__audit",

tests/unit/test_contracts_graph_parsed.py

@@ -1038,6 +1038,7 @@ def basic_parsed_schema_test_dict():
         "config": {
             "enabled": True,
             "materialized": "test",
+            "store_failures_as": "table",
             "tags": [],
             "severity": "ERROR",
             "warn_if": "!= 0",
@@ -1117,6 +1118,7 @@ def complex_parsed_schema_test_dict():
         "config": {
             "enabled": True,
             "materialized": "table",
+            "store_failures_as": "table",
             "tags": [],
             "severity": "WARN",
             "warn_if": "!= 0",