Repository: dbt-labs/dbt-core (python)

Issue:
[Bug] Unit tests and generic data tests not utilizing custom `ref` override

### Is this a new bug in dbt-core?

- [x] I believe this is a new bug in dbt-core
- [x] I have searched the existing issues, and I could not find an existing issue for this bug

### Current Behavior

The Fusion engine uses any custom `ref` overrides within generic data tests and unit tests, but the Core engine raises errors.

### Expected Behavior

Expect parity between Fusion and Core. In this case, for Core to also utilize custom overrides of `ref` in generic data tests and unit tests.

### Steps To Reproduce

I know the following example is a bit complicated. If someone has a more minimal example, please share all the file details! 🙏 

Create these files:

`macros/ref.sql`

Which adds a `revision` argument (to be another alias for `version` / `v`).

```sql
{% macro ref() %}

-- extract user-provided positional and keyword arguments
{% set version = 2 %}
{% set version = kwargs.get('revision') or kwargs.get('version') or kwargs.get('v') %}
{% set packagename = none %}
{%- if (varargs | length) == 1 -%}
    {% set modelname = varargs[0] %}
{%- else -%}
    {% set packagename = varargs[0] %}
    {% set modelname = varargs[1] %}
{% endif %}

-- call builtins.ref based on provided positional arguments
{% set rel = None %}
{% if packagename is not none %}
    {% set rel = builtins.ref(packagename, modelname, version=version) %}
{% else %}
    {% set rel = builtins.ref(modelname, version=version) %}
{% endif %}

{% do return(rel) %}

{% endmacro %}
```

`models/stg_customers_v1.sql`

```sql
select 1 as customer_id
```

`models/stg_customers_v2.sql`

```sql
select 2 as customer_id
```

`models/dim_customers.sql`

```sql
select * from {{ ref('stg_customers', revision=2) }}
```

`models/fct_customer_orders.sql`

```sql
select 1 as customer_id, 2 as order_id
```

`models/_properties.yml`

```yaml
models:
  - name: stg_customers
    latest_version: 2
    versions:
      - v: 1
      - v: 2

  - name: fct_customer_orders
    columns:

      - name: customer_id
        data_tests:
          - relationships:
              arguments:
                to: ref('stg_customers', revision=1)
                field: customer_id

unit_tests:
  - name: my_unit_test
    model: dim_customers
    given:
      - input: ref('stg_customers', revision=2)
        rows:
          - {customer_id: 3}
    expect:
      rows:
        - {customer_id: 3}
```

Run these commands with the dbt Fusion CLI:

```shell
rm -rf target && dbtf run && dbtf test
```

Get this output:

```
dbt-fusion 2.0.0-preview.57
   Loading profiles.yml
 Succeeded [  1.14s] model dbt_dbeatty.stg_customers_v2 (view)
 Succeeded [  0.65s] model dbt_dbeatty.dim_customers (view)
 Succeeded [  1.84s] model dbt_dbeatty.stg_customers_v1 (view)
 Succeeded [  1.97s] model dbt_dbeatty.fct_customer_orders (view)
New version available 2.0.0-preview.60 (run `dbt system update`)

======================================================== Execution Summary ========================================================
Finished 'run' successfully for target 'dev' [4.2s]
Processed: 4 models
Summary: 4 total | 4 success
```

```
dbt-fusion 2.0.0-preview.57
   Loading profiles.yml
    Passed [  1.64s] test  dbt_dbeatty_dbt_test__audit.relationships_fct_customer_ord_bc274c168d57dd10454e9e976fa58626
    Passed [  1.49s] unit_test dbt_dbeatty_dbt_test__audit.my_unit_test
New version available 2.0.0-preview.60 (run `dbt system update`)

======================================================== Execution Summary ========================================================
Finished 'test' successfully for target 'dev' [2.1s]
Processed: 1 test | 1 unit test
Summary: 3 total | 3 success
```

Run the same commands with the dbt Core CLI:

```shell
rm -rf target && dbt run && dbt test
```

Get this output:

```
21:47:17  Running with dbt=1.10.9
21:47:18  Registered adapter: snowflake=1.10.0
21:47:18  Unable to do partial parsing because saved manifest not found. Starting full parse.
21:47:19  Found 4 models, 1 test, 480 macros, 1 unit test
21:47:19  
21:47:19  Concurrency: 10 threads (target='dev')
21:47:19  
21:47:23  1 of 4 START sql view model dbt_dbeatty.fct_customer_orders .................... [RUN]
21:47:23  2 of 4 START sql view model dbt_dbeatty.stg_customers_v1 ....................... [RUN]
21:47:23  3 of 4 START sql view model dbt_dbeatty.stg_customers_v2 ....................... [RUN]
21:47:23  1 of 4 OK created sql view model dbt_dbeatty.fct_customer_orders ............... [SUCCESS 1 in 0.34s]
21:47:23  2 of 4 OK created sql view model dbt_dbeatty.stg_customers_v1 .................. [SUCCESS 1 in 0.71s]
21:47:23  3 of 4 OK created sql view model dbt_dbeatty.stg_customers_v2 .................. [SUCCESS 1 in 0.75s]
21:47:23  4 of 4 START sql view model dbt_dbeatty.dim_customers .......................... [RUN]
21:47:24  4 of 4 OK created sql view model dbt_dbeatty.dim_customers ..................... [SUCCESS 1 in 0.70s]
21:47:25  
21:47:25  Finished running 4 view models in 0 hours 0 minutes and 6.27 seconds (6.27s).
21:47:25  
21:47:25  Completed successfully
21:47:25  
21:47:25  Done. PASS=4 WARN=0 ERROR=0 SKIP=0 NO-OP=0 TOTAL=4
```

```
21:47:27  Running with dbt=1.10.9
21:47:28  Registered adapter: snowflake=1.10.0
21:47:28  Found 4 models, 1 test, 480 macros, 1 unit test
21:47:28  
21:47:28  Concurrency: 10 threads (target='dev')
21:47:28  
21:47:28  1 of 2 START test relationships_fct_customer_orders_customer_id__customer_id__ref_stg_customers_revision_1_  [RUN]
21:47:28  2 of 2 START unit_test dim_customers::my_unit_test ............................. [RUN]
21:47:28  2 of 2 ERROR dim_customers::my_unit_test ....................................... [ERROR in 0.01s]
21:47:29  1 of 2 ERROR relationships_fct_customer_orders_customer_id__customer_id__ref_stg_customers_revision_1_  [ERROR in 0.02s]
21:47:29  
21:47:29  Finished running 1 test, 1 unit test in 0 hours 0 minutes and 0.94 seconds (0.94s).
21:47:29  
21:47:29  Completed with 2 errors, 0 partial successes, and 0 warnings:
21:47:29  
21:47:29  Failure in unit_test my_unit_test (models/_properties.yml)
21:47:29    Parsing Error in unit_test my_unit_test (models/_properties.yml)
  Unit test given inputs must be either a 'ref', 'source' or 'this' call. Got: 'ref('stg_customers', revision=2)'.
21:47:29  
21:47:29  Failure in test relationships_fct_customer_orders_customer_id__customer_id__ref_stg_customers_revision_1_ (models/_properties.yml)
21:47:29    Compilation Error in test relationships_fct_customer_orders_customer_id__customer_id__ref_stg_customers_revision_1_ (models/_properties.yml)
  dbt was unable to infer all dependencies for the model "relationships_fct_customer_orders_customer_id__customer_id__ref_stg_customers_revision_1_".
  This typically happens when ref() is placed within a conditional block.
  
  To fix this, add the following hint to the top of the model "relationships_fct_customer_orders_customer_id__customer_id__ref_stg_customers_revision_1_":
  
  -- depends_on: {{ ref('stg_customers',version='1') }}
  
  > in macro ref (macros/ref.sql)
  > called by test relationships_fct_customer_orders_customer_id__customer_id__ref_stg_customers_revision_1_ (models/_properties.yml)
21:47:29  
21:47:29  Done. PASS=0 WARN=0 ERROR=2 SKIP=0 NO-OP=0 TOTAL=2
```

### Relevant log output

Note the failure logs for the unit test:

```shell
21:47:29  Failure in unit_test my_unit_test (models/_properties.yml)
21:47:29    Parsing Error in unit_test my_unit_test (models/_properties.yml)
  Unit test given inputs must be either a 'ref', 'source' or 'this' call. Got: 'ref('stg_customers', revision=2)'.
```

Note the failure logs for the generic data test:

```
21:47:29  Failure in test relationships_fct_customer_orders_customer_id__customer_id__ref_stg_customers_revision_1_ (models/_properties.yml)
21:47:29    Compilation Error in test relationships_fct_customer_orders_customer_id__customer_id__ref_stg_customers_revision_1_ (models/_properties.yml)
  dbt was unable to infer all dependencies for the model "relationships_fct_customer_orders_customer_id__customer_id__ref_stg_customers_revision_1_".
  This typically happens when ref() is placed within a conditional block.
  
  To fix this, add the following hint to the top of the model "relationships_fct_customer_orders_customer_id__customer_id__ref_stg_customers_revision_1_":
  
  -- depends_on: {{ ref('stg_customers',version='1') }}
  
  > in macro ref (macros/ref.sql)
  > called by test relationships_fct_customer_orders_customer_id__customer_id__ref_stg_customers_revision_1_ (models/_properties.yml)
21:47:29  
```

### Environment

```markdown
- OS:
- Python:
- dbt:
```

### Which database adapter are you using with dbt?

_No response_

### Additional Context

_No response_


Patch:
diff --git a/.changes/unreleased/Fixes-20260318-091800.yaml b/.changes/unreleased/Fixes-20260318-091800.yaml
new file mode 100644
index 00000000000..b5c186fe790
--- /dev/null
+++ b/.changes/unreleased/Fixes-20260318-091800.yaml
@@ -0,0 +1,6 @@
+kind: Fixes
+body: Support custom ref kwargs in unit tests and generic data tests, behind behavior flag `support_custom_ref_kwargs`
+time: 2026-03-18T09:18:00.000000-05:00
+custom:
+    Author: aahel
+    Issue: 12148
diff --git a/core/dbt/context/providers.py b/core/dbt/context/providers.py
index d0f3c55851f..4426d09a0ce 100644
--- a/core/dbt/context/providers.py
+++ b/core/dbt/context/providers.py
@@ -2246,6 +2246,18 @@ def _build_test_namespace(self):
             depends_on_macros.append(get_where_subquery.unique_id)
         if self.model.depends_on and self.model.depends_on.macros:
             depends_on_macros.extend(self.model.depends_on.macros)
+
+        # When support_custom_ref_kwargs is enabled, include custom ref/source
+        # macro overrides in the test namespace so that test arguments like
+        # ref('model', custom_kwarg=value) are properly resolved.
+        if getattr(get_flags(), "SUPPORT_CUSTOM_REF_KWARGS", False):
+            for macro_name in ("ref", "source"):
+                macro = self.macro_resolver.macros_by_name.get(macro_name)
+                if macro and macro.unique_id not in depends_on_macros:
+                    # Only include user-defined overrides, not built-in macros
+                    if macro.package_name != "dbt":
+                        depends_on_macros.append(macro.unique_id)
+
         lookup_macros = depends_on_macros.copy()
         for macro_unique_id in lookup_macros:
             lookup_macro = self.macro_resolver.macros.get(macro_unique_id)
diff --git a/core/dbt/contracts/project.py b/core/dbt/contracts/project.py
index e8a3d11f650..6c53ce39a94 100644
--- a/core/dbt/contracts/project.py
+++ b/core/dbt/contracts/project.py
@@ -370,6 +370,7 @@ class ProjectFlags(ExtensibleDbtClassMixin):
     require_ref_searches_node_package_before_root: bool = False
     require_valid_schema_from_generate_schema_name: bool = False
     require_sql_header_in_test_configs: bool = False
+    support_custom_ref_kwargs: bool = False
 
     @property
     def project_only_flags(self) -> Dict[str, Any]:
@@ -390,6 +391,7 @@ def project_only_flags(self) -> Dict[str, Any]:
             "require_ref_searches_node_package_before_root": self.require_ref_searches_node_package_before_root,
             "require_valid_schema_from_generate_schema_name": self.require_valid_schema_from_generate_schema_name,
             "require_sql_header_in_test_configs": self.require_sql_header_in_test_configs,
+            "support_custom_ref_kwargs": self.support_custom_ref_kwargs,
         }
 
 


Test patch:
diff --git a/core/dbt/parser/unit_tests.py b/core/dbt/parser/unit_tests.py
index 113a9866c82..bed54509a6b 100644
--- a/core/dbt/parser/unit_tests.py
+++ b/core/dbt/parser/unit_tests.py
@@ -23,6 +23,7 @@
 )
 from dbt.contracts.graph.unparsed import UnparsedUnitTest
 from dbt.exceptions import InvalidUnitTestGivenInput, ParsingError
+from dbt.flags import get_flags
 from dbt.graph import UniqueId
 from dbt.node_types import NodeType
 from dbt.parser.schemas import (
@@ -123,7 +124,7 @@ def parse_unit_test_case(self, test_case: UnitTestDefinition):
         for given in test_case.given:
             # extract the original_input_node from the ref in the "input" key of the given list
             original_input_node = self._get_original_input_node(
-                given.input, tested_node, test_case.name
+                given.input, tested_node, test_case.name, ctx, unit_test_node
             )
             input_name = original_input_node.name
             common_fields = {
@@ -201,7 +202,14 @@ def _build_fixture_raw_code(self, rows, column_name_to_data_types, fixture_forma
                 rows=rows, column_name_to_data_types=column_name_to_data_types
             )
 
-    def _get_original_input_node(self, input: str, tested_node: ModelNode, test_case_name: str):
+    def _get_original_input_node(
+        self,
+        input: str,
+        tested_node: ModelNode,
+        test_case_name: str,
+        ctx: Optional[Dict[str, Any]] = None,
+        unit_test_node: Optional[UnitTestNode] = None,
+    ):
         """
         Returns the original input node as defined in the project given an input reference
         and the node being tested.
@@ -219,7 +227,16 @@ def _get_original_input_node(self, input: str, tested_node: ModelNode, test_case
             try:
                 statically_parsed = py_extract_from_source(f"{{{{ {input} }}}}")
             except ExtractionError:
-                raise InvalidUnitTestGivenInput(input=input)
+                if (
+                    getattr(get_flags(), "SUPPORT_CUSTOM_REF_KWARGS", False)
+                    and ctx is not None
+                    and unit_test_node is not None
+                ):
+                    statically_parsed = self._extract_ref_or_source_via_jinja(
+                        input, ctx, unit_test_node
+                    )
+                else:
+                    raise InvalidUnitTestGivenInput(input=input)
 
             if statically_parsed["refs"]:
                 ref = list(statically_parsed["refs"])[0]
@@ -247,6 +264,49 @@ def _get_original_input_node(self, input: str, tested_node: ModelNode, test_case
 
         return original_input_node
 
+    def _extract_ref_or_source_via_jinja(
+        self,
+        input_str: str,
+        ctx: Dict[str, Any],
+        unit_test_node: UnitTestNode,
+    ) -> Dict[str, Any]:
+        """Fall back to Jinja rendering to resolve ref() calls with custom kwargs.
+
+        When py_extract_from_source cannot handle custom keyword arguments
+        (e.g., ref('model', revision=2) via a custom ref macro), this method
+        renders the input through Jinja so the custom macro can map custom
+        kwargs to standard ones (like version).
+        """
+        # Save current refs/sources to restore after rendering
+        original_refs = list(unit_test_node.refs)
+        original_sources = list(unit_test_node.sources)
+
+        try:
+            get_rendered(f"{{{{ {input_str} }}}}", ctx, unit_test_node, capture_macros=True)
+        except Exception:
+            raise InvalidUnitTestGivenInput(input=input_str)
+
+        # Extract newly captured refs/sources
+        new_refs = unit_test_node.refs[len(original_refs) :]
+        new_sources = unit_test_node.sources[len(original_sources) :]
+
+        # Restore original refs/sources
+        unit_test_node.refs = original_refs
+        unit_test_node.sources = original_sources
+
+        result: Dict[str, Any] = {"refs": [], "sources": []}
+        for ref_arg in new_refs:
+            result["refs"].append(
+                {"name": ref_arg.name, "package": ref_arg.package, "version": ref_arg.version}
+            )
+        for source in new_sources:
+            result["sources"].append(source)
+
+        if not result["refs"] and not result["sources"]:
+            raise InvalidUnitTestGivenInput(input=input_str)
+
+        return result
+
 
 class UnitTestParser(YamlReader):
     def __init__(self, schema_parser: SchemaParser, yaml: YamlBlock) -> None:
diff --git a/tests/functional/ref_override/test_custom_ref_kwargs.py b/tests/functional/ref_override/test_custom_ref_kwargs.py
new file mode 100644
index 00000000000..e28eb9d8966
--- /dev/null
+++ b/tests/functional/ref_override/test_custom_ref_kwargs.py
@@ -0,0 +1,167 @@
+import pytest
+
+from dbt.tests.util import run_dbt
+
+# Custom ref macro that accepts and ignores a 'label' kwarg
+macros__custom_ref_sql = """
+{% macro ref() %}
+{% set label = kwargs.get('label') %}
+{% set version = kwargs.get('version') or kwargs.get('v') %}
+{% set packagename = none %}
+{%- if (varargs | length) == 1 -%}
+    {% set modelname = varargs[0] %}
+{%- else -%}
+    {% set packagename = varargs[0] %}
+    {% set modelname = varargs[1] %}
+{% endif %}
+
+{% set rel = None %}
+{% if packagename is not none %}
+    {% set rel = builtins.ref(packagename, modelname, version=version) %}
+{% else %}
+    {% set rel = builtins.ref(modelname, version=version) %}
+{% endif %}
+
+{% do return(rel) %}
+{% endmacro %}
+"""
+
+models__model_a_sql = """
+select 1 as id, 'alice' as name
+"""
+
+models__model_b_sql = """
+select * from {{ ref('model_a', label='staging') }}
+"""
+
+models__schema_yml = """
+models:
+  - name: model_a
+  - name: model_b
+    columns:
+      - name: id
+        data_tests:
+          - relationships:
+              to: ref('model_a', label='staging')
+              field: id
+
+unit_tests:
+  - name: my_unit_test
+    model: model_b
+    given:
+      - input: ref('model_a', label='staging')
+        rows:
+          - {id: 1, name: 'alice'}
+    expect:
+      rows:
+        - {id: 1, name: 'alice'}
+"""
+
+# Separate schema with only unit test for the "without flag" test
+models__schema_unit_only_yml = """
+models:
+  - name: model_a
+  - name: model_b
+
+unit_tests:
+  - name: my_unit_test
+    model: model_b
+    given:
+      - input: ref('model_a', label='staging')
+        rows:
+          - {id: 1, name: 'alice'}
+    expect:
+      rows:
+        - {id: 1, name: 'alice'}
+"""
+
+
+class TestCustomRefKwargsUnitTest:
+    """Test that unit tests work with custom ref kwargs when flag is enabled."""
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {
+                "support_custom_ref_kwargs": True,
+            },
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "model_a.sql": models__model_a_sql,
+            "model_b.sql": models__model_b_sql,
+            "schema.yml": models__schema_yml,
+        }
+
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {"ref.sql": macros__custom_ref_sql}
+
+    def test_unit_test_with_custom_ref_kwargs(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 2
+
+        results = run_dbt(["test", "--select", "test_type:unit"])
+        assert len(results) == 1
+        assert results[0].status == "pass"
+
+
+class TestCustomRefKwargsGenericTest:
+    """Test that generic data tests work with custom ref kwargs when flag is enabled."""
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {
+                "support_custom_ref_kwargs": True,
+            },
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "model_a.sql": models__model_a_sql,
+            "model_b.sql": models__model_b_sql,
+            "schema.yml": models__schema_yml,
+        }
+
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {"ref.sql": macros__custom_ref_sql}
+
+    def test_generic_test_with_custom_ref_kwargs(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 2
+
+        results = run_dbt(["test", "--select", "test_type:data"])
+        assert len(results) == 1
+        assert results[0].status == "pass"
+
+
+class TestCustomRefKwargsWithoutFlag:
+    """Test that custom ref kwargs in unit tests fail without the behavior flag."""
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "model_a.sql": models__model_a_sql,
+            "model_b.sql": models__model_b_sql,
+            "schema.yml": models__schema_unit_only_yml,
+        }
+
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {"ref.sql": macros__custom_ref_sql}
+
+    def test_unit_test_fails_without_flag(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 2
+
+        results = run_dbt(["test", "--select", "test_type:unit"], expect_pass=False)
+        assert len(results) == 1
+        assert (
+            "Unit test given inputs must be either a 'ref', 'source' or 'this' call"
+            in results.results[0].message
+        )