First pass at functional test for unit tests with external nodes

tests/functional/unit_testing/fixtures.py

@@ -598,3 +598,33 @@
       format: csv
       fixture: test_my_model_basic_fixture
 """
+
+top_level_domains_sql = """
+SELECT 'example.com' AS tld
+UNION ALL
+SELECT 'gmail.com' AS tld
+"""
+
+test_my_model_external_nodes_sql = """
+WITH
+accounts AS (
+  SELECT user_id, email, email_top_level_domain
+  FROM {{ ref('external_package', 'external_model')}}
+),
+top_level_domains AS (
+  SELECT tld FROM {{ ref('top_level_domains')}}
+),
+joined AS (
+  SELECT
+    accounts.user_id as user_id,
+    top_level_domains.tld as tld
+  FROM accounts
+  LEFT OUTER JOIN top_level_domains
+    ON   accounts.email_top_level_domain = top_level_domains.tld
+)
+
+SELECT
+  joined.user_id as user_id,
+  CASE WHEN joined.tld IS NULL THEN 'FALSE' ELSE 'TRUE' END AS is_valid_email_address
+from joined
+"""

tests/functional/unit_testing/test_unit_testing.py

@@ -1,10 +1,12 @@
 import pytest
+from unittest import mock
 from dbt.tests.util import (
     run_dbt,
     write_file,
     get_manifest,
 )
 from dbt.exceptions import DuplicateResourceNameError, ParsingError
+from dbt.plugins.manifest import PluginNodes, ModelNodeArgs
 from fixtures import (
     my_model_vars_sql,
     my_model_a_sql,
@@ -14,6 +16,8 @@
     my_incremental_model_sql,
     event_sql,
     test_my_model_incremental_yml,
+    test_my_model_external_nodes_sql,
+    top_level_domains_sql,
 )
 
 
@@ -230,3 +234,61 @@ def test_nonexistent_seed(self, project):
             ParsingError, match="Unable to find seed 'test.my_second_favorite_seed' for unit tests"
         ):
             run_dbt(["test", "--select", "my_new_model"], expect_pass=False)
+
+
+test_unit_test_with_external_nodes_yml = """
+unit_tests:
+  - name: test_unit_test_with_external_nodes
+    model: test_my_model_external_nodes
+    given:
+      - input: ref('external_package', 'external_model')
+        rows:
+          - {user_id: 1, email: cool@example.com,     email_top_level_domain: example.com}
+          - {user_id: 2, email: cool@unknown.com,     email_top_level_domain: unknown.com}
+          - {user_id: 3, email: badgmail.com,         email_top_level_domain: gmail.com}
+          - {user_id: 4, email: missingdot@gmailcom,  email_top_level_domain: gmail.com}
+      - input: ref('top_level_domains')
+        rows:
+          - {tld: example.com}
+          - {tld: gmail.com}
+    expect:
+      rows:
+        - {user_id: 1, is_valid_email_address: true}
+        - {user_id: 2, is_valid_email_address: false}
+        - {user_id: 3, is_valid_email_address: false}
+        - {user_id: 4, is_valid_email_address: false}
+"""
+
+
+class TestUnitTestExternalNode:
+    @pytest.fixture(scope="class")
+    def external_model_node(self):
+        return ModelNodeArgs(
+            name="external_model",
+            package_name="external_package",
+            identifier="test_identifier",
+            schema="test_schema",
+        )
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "top_level_domains.sql": top_level_domains_sql,
+            "test_my_model_external_nodes.sql": test_my_model_external_nodes_sql,
+            "test_unit_test_with_external_nodes.yml": test_unit_test_with_external_nodes_yml,
+        }
+
+    @mock.patch("dbt.plugins.get_plugin_manager")
+    def test_unit_test_external_nodes(
+        self,
+        get_plugin_manager,
+        project,
+        external_model_node,
+    ):
+        # initial plugin - one external model
+        external_nodes = PluginNodes()
+        external_nodes.add_model(external_model_node)
+        get_plugin_manager.return_value.get_nodes.return_value = external_nodes
+
+        results = run_dbt(["test", "--select", "test_my_model_external_nodes"], expect_pass=True)
+        assert len(results) == 1