Ensure inferred primary_key is a List[str] (#10984)

.changes/unreleased/Fixes-20241106-144656.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: 'Ensure inferred primary_key is a List[str] with no null values '
+time: 2024-11-06T14:46:56.652963-05:00
+custom:
+  Author: michelleark
+  Issue: "10983"

core/dbt/contracts/graph/nodes.py

@@ -541,11 +541,20 @@ class ModelNode(ModelResource, CompiledNode):
         columns_with_disabled_unique_tests = set()
         columns_with_not_null_tests = set()
         for test in data_tests:
-            columns = []
-            if "column_name" in test.test_metadata.kwargs:
+            columns: List[str] = []
+            # extract columns from test kwargs, ensuring columns is a List[str] given tests can have custom (user or pacakge-defined) kwarg types
+            if "column_name" in test.test_metadata.kwargs and isinstance(
+                test.test_metadata.kwargs["column_name"], str
+            ):
                 columns = [test.test_metadata.kwargs["column_name"]]
-            elif "combination_of_columns" in test.test_metadata.kwargs:
-                columns = test.test_metadata.kwargs["combination_of_columns"]
+            elif "combination_of_columns" in test.test_metadata.kwargs and isinstance(
+                test.test_metadata.kwargs["combination_of_columns"], list
+            ):
+                columns = [
+                    column
+                    for column in test.test_metadata.kwargs["combination_of_columns"]
+                    if isinstance(column, str)
+                ]
 
             for column in columns:
                 if test.test_metadata.name in ["unique", "unique_combination_of_columns"]:

tests/functional/primary_keys/fixtures.py

@@ -11,6 +11,16 @@ models:
           - unique
 """
 
+invalid_model_unique_test = """
+models:
+  - name: simple_model
+    data_tests:
+      - unique:
+          column_name: null
+    columns:
+      - name: id
+"""
+
 simple_model_disabled_unique_test = """
 models:
   - name: simple_model
@@ -40,6 +50,16 @@ models:
           combination_of_columns: [id, color]
 """
 
+invalid_model_unique_combo_of_columns = """
+models:
+  - name: simple_model
+    tests:
+      - dbt_utils.unique_combination_of_columns:
+          combination_of_columns: [null]
+      - dbt_utils.unique_combination_of_columns:
+          combination_of_columns: "test"
+"""
+
 simple_model_constraints = """
 models:
   - name: simple_model

tests/functional/primary_keys/test_primary_keys.py

@@ -2,6 +2,8 @@ import pytest
 
 from dbt.tests.util import get_manifest, run_dbt
 from tests.functional.primary_keys.fixtures import (
+    invalid_model_unique_combo_of_columns,
+    invalid_model_unique_test,
     simple_model_constraints,
     simple_model_disabled_unique_test,
     simple_model_sql,
@@ -155,3 +157,57 @@ class TestSimpleModelCombinationOfColumns:
         manifest = get_manifest(project.project_root)
         node = manifest.nodes["model.test.simple_model"]
         assert node.primary_key == ["color", "id"]
+
+
+class TestInvalidModelCombinationOfColumns:
+    @pytest.fixture(scope="class")
+    def packages(self):
+        return {
+            "packages": [
+                {
+                    "git": "https://github.com/dbt-labs/dbt-utils.git",
+                    "revision": "1.1.0",
+                },
+            ]
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "simple_model.sql": simple_model_sql,
+            "schema.yml": invalid_model_unique_combo_of_columns,
+        }
+
+    def test_invalid_combo_of_columns(self, project):
+        run_dbt(["deps"])
+        run_dbt(["run"])
+        manifest = get_manifest(project.project_root)
+        node = manifest.nodes["model.test.simple_model"]
+        assert node.primary_key == []
+
+
+class TestInvalidModelUniqueTest:
+    @pytest.fixture(scope="class")
+    def packages(self):
+        return {
+            "packages": [
+                {
+                    "git": "https://github.com/dbt-labs/dbt-utils.git",
+                    "revision": "1.1.0",
+                },
+            ]
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "simple_model.sql": simple_model_sql,
+            "schema.yml": invalid_model_unique_test,
+        }
+
+    def test_invalid_combo_of_columns(self, project):
+        run_dbt(["deps"])
+        run_dbt(["run"])
+        manifest = get_manifest(project.project_root)
+        node = manifest.nodes["model.test.simple_model"]
+        assert node.primary_key == []