implement persist test results tests, leave querying relation type as…

… open issue

.changes/unreleased/Features-20230919-171047.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support persisting tests results as views
+time: 2023-09-19T17:10:47.516098-04:00
+custom:
+  Author: mikealfare
+  Issue: "6914"

dev-requirements.txt

@@ -1,7 +1,7 @@
 # install latest changes in dbt-core
 # TODO: how to automate switching from develop to version branches?
-git+https://github.com/dbt-labs/dbt-core.git#egg=dbt-core&subdirectory=core
-git+https://github.com/dbt-labs/dbt-core.git#egg=dbt-tests-adapter&subdirectory=tests/adapter
+git+https://github.com/dbt-labs/dbt-core.git@feature/materialized-tests/adap-850#egg=dbt-core&subdirectory=core
+git+https://github.com/dbt-labs/dbt-core.git@feature/materialized-tests/adap-850#egg=dbt-tests-adapter&subdirectory=tests/adapter
 
 # if version 1.x or greater -> pin to major version
 # if version 0.x -> pin to minor

tests/functional/persist_test_results/test_persist_test_results.py

@@ -0,0 +1,20 @@
+from typing import Dict, Set, Tuple
+
+from dbt.tests.adapter.persist_test_results.basic import PersistTestResults
+
+from tests.functional.persist_test_results.utils import (
+    delete_record,
+    get_relation_summary_in_schema,
+    insert_record,
+)
+
+
+class TestPersistTestResults(PersistTestResults):
+    def get_audit_relation_summary(self, project) -> Set[Tuple]:
+        return get_relation_summary_in_schema(project, self.audit_schema)
+
+    def insert_record(self, project, record: Dict[str, str]):
+        insert_record(project, project.test_schema, self.model_table, record)
+
+    def delete_record(self, project, record: Dict[str, str]):
+        delete_record(project, project.test_schema, self.model_table, record)

tests/functional/persist_test_results/utils.py

@@ -0,0 +1,55 @@
+from typing import Dict, Set, Tuple
+
+
+def get_relation_summary_in_schema(project, schema: str) -> Set[Tuple]:
+    """
+    Returns a summary like this:
+        {
+            ("my_table", "table", 0),
+            ("my_view", "view", 1),
+        }
+    """
+    sql = """
+    We can't get the relation type in tests right now because it requires a multi-statement sql execution.
+    This needs to be solved prior to automating these tests. This will also resolve the same issue for DT tests.
+    """
+    relation_types = project.run_sql(sql, fetch="all")
+
+    results = set()
+    for relation_name, relation_type in relation_types:
+        row_count_sql = f"select count(*) from {schema}.{relation_name}"
+        row_count = project.run_sql(row_count_sql, fetch="one")[0]
+        summary = (relation_name, relation_type, row_count)
+        results.add(summary)
+
+    return results
+
+
+def insert_record(project, schema: str, table_name: str, record: Dict[str, str]):
+    # postgres only supports schema names of 63 characters
+    # a schema with a longer name still gets created, but the name gets truncated
+    schema_name = schema[:63]
+    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 {schema_name}.{table_name} ({field_name_clause})
+    values ({field_value_clause})
+    """
+    project.run_sql(sql)
+
+
+def delete_record(project, schema: str, table_name: str, record: Dict[str, str]):
+    schema_name = schema[:63]
+    where_clause = " and ".join(
+        [f"{field_name} = '{field_value}'" for field_name, field_value in record.items()]
+    )
+    sql = f"""
+    delete from {schema_name}.{table_name}
+    where {where_clause}
+    """
+    project.run_sql(sql)