[ALT] Ensure unit test models can depend on external nodes

core/dbt/context/providers.py

@@ -577,7 +577,7 @@
         target_package: Optional[str] = None,
         target_version: Optional[NodeVersion] = None,
     ) -> RelationProxy:
-        return super().resolve(target_name, target_package, target_version)
+        return super().resolve(target_name, self.model.package_name, target_version)
 
 
 # `source` implementations

core/dbt/parser/unit_tests.py

@@ -139,7 +139,7 @@ def parse_unit_test_case(self, test_case: UnitTestDefinition):
                     **common_fields,
                     unique_id=f"model.{test_case.package_name}.{input_name}",
                     name=input_name,
-                    path=original_input_node.path,
+                    path=original_input_node.path or f"{input_name}.sql",
                 )
             elif original_input_node.resource_type == NodeType.Source:
                 # We are reusing the database/schema/identifier from the original source,

tests/functional/unit_testing/fixtures.py

@@ -649,3 +649,33 @@
       rows:
         - {c: 3}
 """
+
+top_level_domains_sql = """
+SELECT 'example.com' AS tld
+UNION ALL
+SELECT 'gmail.com' AS tld
+"""
+
+valid_emails_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,11 +1,13 @@
 import pytest
+from unittest import mock
 from dbt.tests.util import (
     run_dbt,
     write_file,
     get_manifest,
 )
 from dbt.contracts.results import NodeStatus
 from dbt.exceptions import DuplicateResourceNameError, ParsingError
+from dbt.plugins.manifest import PluginNodes, ModelNodeArgs
 from fixtures import (
     my_model_sql,
     my_model_vars_sql,
@@ -17,6 +19,8 @@
     event_sql,
     test_my_model_incremental_yml,
     test_my_model_yml_invalid,
+    valid_emails_sql,
+    top_level_domains_sql,
 )
 
 
@@ -256,3 +260,73 @@ def test_invalid_input_configuration(self, project):
         assert len(results) == 3
 
         run_dbt(["test"], expect_pass=False)
+
+
+unit_test_ext_node_yml = """
+unit_tests:
+  - name: unit_test_ext_node
+    model: valid_emails
+    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: true}
+        - {user_id: 4, is_valid_email_address: true}
+"""
+
+external_node_seed_csv = """user_id,email,email_top_level_domain
+1,"example@example.com","example.com"
+"""
+
+
+class TestUnitTestExternalNode:
+    @pytest.fixture(scope="class")
+    def external_model_node(self, unique_schema):
+        return ModelNodeArgs(
+            name="external_model",
+            package_name="external_package",
+            identifier="external_node_seed",
+            schema=unique_schema,
+        )
+
+    @pytest.fixture(scope="class")
+    def seeds(self):
+        return {"external_node_seed.csv": external_node_seed_csv}
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "top_level_domains.sql": top_level_domains_sql,
+            "valid_emails.sql": valid_emails_sql,
+            "unit_test_ext_node.yml": unit_test_ext_node_yml,
+        }
+
+    @mock.patch("dbt.plugins.get_plugin_manager")
+    def test_unit_test_ext_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
+
+        # `seed` need so a table exists for `external_model` to point to
+        run_dbt(["seed"], expect_pass=True)
+        # `run` needed to ensure `top_level_domains` exists in database for column getting step
+        run_dbt(["run"], expect_pass=True)
+        results = run_dbt(["test", "--select", "valid_emails"], expect_pass=True)
+        assert len(results) == 1