Repository: dbt-labs/dbt-core
Pull request: Add initial unit tests for `ModelRunner` class
resolves #9867 


### Problem

We haven't had a way to _easily_ unit test things like the `ModelRunner` class previously. This is because there were A LOT of things to mock / build before getting to "testing". We've been building a fair amount of testing fixtures and utilities over the past few months. We wanted to investigate how close we are to being able to unit test the `ModelRunner` class.

### Solution

Attempt to unit test `ModelRunner` class. We were able to write a unit test the `print_result_line` method, which itself calls many `ModelRunner` methods. However we were unable to successfully test the _primary_ method `execute`.

### Checklist

- [X] I have read [the contributing guide](https://github.com/dbt-labs/dbt-core/blob/main/CONTRIBUTING.md) and understand what's expected of me  
- [X] I have run this code in development and it appears to resolve the stated issue  
- [X] This PR includes tests, or tests are not required/relevant for this PR
- [X] This PR has no interface changes (e.g. macros, cli, logs, json artifacts, config files, adapter interface, etc) or this PR has already received feedback and approval from Product or DX
- [X] This PR includes [type annotations](https://docs.python.org/3/library/typing.html) for new and modified functions


Problem statement:
[SPIKE+] Make a plan for how to write a unittest for ModelRunner
# Description
make a plan for how to write a unit test for the ModelRunner, and implement it if possible.
This may require:
- mocking adapter
- utils/fixtures to generate manifest that we already have
- Mocking logging system to capture logging behavior

This is not amid to have a completely tested ModelRunner. The goal should be to have a good understanding of what utils/fixtures are best for a runner's needs. And have one example to follow.

Hints from issue discussion:
Blocked on
- mocking the adapters: this should be part of the Spike
- [mocking the logger](https://github.com/dbt-labs/dbt-core/issues/9947)

Commit message: Add spoofed macro fixture `materialization_table_default` for `test_execute` test

Previously the `TestModelRunner:test_execute` test was running into a runtime error
do to the macro `materialization_table_default` macro not existing in the project. This
commit adds that macro to the project (though it should ideally get loaded via interactions
between the manifest and adapter). Manually adding it resolved our previous issue, but created
a new one. The macro appears to not be properly loaded into the manifest, and thus isn't
discoverable later on when getting the macros for the jinja context. This leads to an error
that looks like the following:
```
model_runner = <dbt.task.run.ModelRunner object at 0x1080a4f70>

    def test_execute(
        self, table_model: ModelNode, manifest: Manifest, model_runner: ModelRunner
    ) -> None:
>       model_runner.execute(model=table_model, manifest=manifest)

tests/unit/task/test_run.py:129:
----
core/dbt/task/run.py:287: in execute
    result = MacroGenerator(
core/dbt/clients/jinja.py:82: in __call__
    return self.call_macro(*args, **kwargs)
venv/lib/python3.10/site-packages/dbt_common/clients/jinja.py:294: in call_macro
    macro = self.get_macro()
---
self = <dbt.clients.jinja.MacroGenerator object at 0x1080f3130>

    def get_macro(self):
        name = self.get_name()
        template = self.get_template()
        # make the module. previously we set both vars and local, but that's
        # redundant: They both end up in the same place
        # make_module is in jinja2.environment. It returns a TemplateModule
        module = template.make_module(vars=self.context, shared=False)
>       macro = module.__dict__[get_dbt_macro_name(name)]
E       KeyError: 'dbt_macro__materialization_table_default'

venv/lib/python3.10/site-packages/dbt_common/clients/jinja.py:277: KeyError
```

It's becoming apparent that we need to find a better way to either mock or legitimately
load the default and adapter macros. At this point I think I've exausted the time box
I should be using to figure out if testing the `ModelRunner` class is possible currently,
with the result being more work has yet to be done.

Patch to review:
diff --git a/tests/unit/task/test_run.py b/tests/unit/task/test_run.py
index c689e8f41aa..92ccde5f325 100644
--- a/tests/unit/task/test_run.py
+++ b/tests/unit/task/test_run.py
@@ -3,10 +3,19 @@
 
 import pytest
 
+from dbt.adapters.postgres import PostgresAdapter
+from dbt.artifacts.schemas.results import RunStatus
+from dbt.artifacts.schemas.run import RunResult
 from dbt.config.runtime import RuntimeConfig
+from dbt.contracts.graph.manifest import Manifest
+from dbt.contracts.graph.nodes import ModelNode
+from dbt.events.types import LogModelResult
 from dbt.flags import get_flags, set_from_args
-from dbt.task.run import RunTask
+from dbt.task.run import ModelRunner, RunTask
 from dbt.tests.util import safe_set_invocation_context
+from dbt_common.events.base_types import EventLevel
+from dbt_common.events.event_manager_client import get_event_manager
+from tests.utils import EventCatcher
 
 
 @pytest.mark.parametrize(
@@ -50,3 +59,72 @@ def test_run_task_preserve_edges():
         task.get_graph_queue()
         # when we get the graph queue, preserve_edges is True
         mock_node_selector.get_graph_queue.assert_called_with(mock_spec, True)
+
+
+class TestModelRunner:
+    @pytest.fixture
+    def log_model_result_catcher(self) -> EventCatcher:
+        return EventCatcher(event_to_catch=LogModelResult)
+
+    @pytest.fixture
+    def model_runner(
+        self,
+        postgres_adapter: PostgresAdapter,
+        table_model: ModelNode,
+        runtime_config: RuntimeConfig,
+    ) -> ModelRunner:
+        return ModelRunner(
+            config=runtime_config,
+            adapter=postgres_adapter,
+            node=table_model,
+            node_index=1,
+            num_nodes=1,
+        )
+
+    @pytest.fixture
+    def run_result(self, table_model: ModelNode) -> RunResult:
+        return RunResult(
+            status=RunStatus.Success,
+            timing=[],
+            thread_id="an_id",
+            execution_time=0,
+            adapter_response={},
+            message="It did it",
+            failures=None,
+            node=table_model,
+        )
+
+    def test_print_result_line(
+        self,
+        log_model_result_catcher: EventCatcher,
+        model_runner: ModelRunner,
+        run_result: RunResult,
+    ) -> None:
+        # Setup way to catch events
+        event_manager = get_event_manager()
+        event_manager.callbacks.append(log_model_result_catcher.catch)
+
+        # Check `print_result_line` with "successful" RunResult
+        model_runner.print_result_line(run_result)
+        assert len(log_model_result_catcher.caught_events) == 1
+        assert log_model_result_catcher.caught_events[0].info.level == EventLevel.INFO
+        assert log_model_result_catcher.caught_events[0].data.status == run_result.message
+
+        # reset event catcher
+        log_model_result_catcher.flush()
+
+        # Check `print_result_line` with "error" RunResult
+        run_result.status = RunStatus.Error
+        model_runner.print_result_line(run_result)
+        assert len(log_model_result_catcher.caught_events) == 1
+        assert log_model_result_catcher.caught_events[0].info.level == EventLevel.ERROR
+        assert log_model_result_catcher.caught_events[0].data.status == EventLevel.ERROR
+
+    @pytest.mark.skip(
+        reason="Default and adapter macros aren't being appropriately populated, leading to a runtime error"
+    )
+    def test_execute(
+        self, table_model: ModelNode, manifest: Manifest, model_runner: ModelRunner
+    ) -> None:
+        model_runner.execute(model=table_model, manifest=manifest)
+        # TODO: Assert that the model was executed
diff --git a/tests/unit/utils/adapter.py b/tests/unit/utils/adapter.py
index 06555b0e400..c760a27ba4b 100644
--- a/tests/unit/utils/adapter.py
+++ b/tests/unit/utils/adapter.py
@@ -1,9 +1,22 @@
+import sys
 from unittest.mock import MagicMock
 
 import pytest
+from pytest_mock import MockerFixture
 
+from dbt.adapters.factory import get_adapter, register_adapter, reset_adapters
 from dbt.adapters.postgres import PostgresAdapter
 from dbt.adapters.sql import SQLConnectionManager
+from dbt.config.runtime import RuntimeConfig
+from dbt.context.providers import generate_runtime_macro_context
+from dbt.contracts.graph.manifest import ManifestStateCheck
+from dbt.mp_context import get_mp_context
+from dbt.parser.manifest import ManifestLoader
+
+if sys.version_info < (3, 9):
+    from typing import Generator
+else:
+    from collections.abc import Generator
 
 
 @pytest.fixture
@@ -19,3 +32,28 @@ def mock_adapter(mock_connection_manager: MagicMock) -> MagicMock:
     mock_adapter.connections = mock_connection_manager
     mock_adapter.clear_macro_resolver = MagicMock()
     return mock_adapter
+
+
+@pytest.fixture
+def postgres_adapter(
+    mocker: MockerFixture, runtime_config: RuntimeConfig
+) -> Generator[PostgresAdapter, None, None]:
+    register_adapter(runtime_config, get_mp_context())
+    adapter = get_adapter(runtime_config)
+    assert isinstance(adapter, PostgresAdapter)
+
+    mocker.patch(
+        "dbt.parser.manifest.ManifestLoader.build_manifest_state_check"
+    ).return_value = ManifestStateCheck()
+    manifest = ManifestLoader.load_macros(
+        runtime_config,
+        adapter.connections.set_query_header,
+        base_macros_only=True,
+    )
+
+    adapter.set_macro_resolver(manifest)
+    adapter.set_macro_context_generator(generate_runtime_macro_context)
+
+    yield adapter
+    adapter.cleanup_connections()
+    reset_adapters()
diff --git a/tests/unit/utils/manifest.py b/tests/unit/utils/manifest.py
index 98c00007cd5..712995c1e24 100644
--- a/tests/unit/utils/manifest.py
+++ b/tests/unit/utils/manifest.py
@@ -15,6 +15,7 @@
     TestConfig,
     TestMetadata,
 )
+from dbt.artifacts.resources.types import ModelLanguage
 from dbt.artifacts.resources.v1.model import ModelConfig
 from dbt.contracts.files import AnySourceFile, FileHash
 from dbt.contracts.graph.manifest import Manifest, ManifestMetadata
@@ -524,6 +525,13 @@ def macro_test_not_null() -> Macro:
     )
 
 
+@pytest.fixture
+def macro_materialization_table_default() -> Macro:
+    macro = make_macro("dbt", "materialization_table_default", "SELECT 1")
+    macro.supported_languages = [ModelLanguage.sql]
+    return macro
+
+
 @pytest.fixture
 def macro_default_test_not_null() -> Macro:
     return make_macro("dbt", "default__test_not_null", "blabla")
@@ -958,12 +966,14 @@ def macros(
     macro_default_test_unique,
     macro_test_not_null,
     macro_default_test_not_null,
+    macro_materialization_table_default,
 ) -> List[Macro]:
     return [
         macro_test_unique,
         macro_default_test_unique,
         macro_test_not_null,
         macro_default_test_not_null,
+        macro_materialization_table_default,
     ]
 
 


Code review comments:
diff_hunk: @@ -50,3 +59,72 @@ def test_run_task_preserve_edges():
         task.get_graph_queue()
         # when we get the graph queue, preserve_edges is True
         mock_node_selector.get_graph_queue.assert_called_with(mock_spec, True)
+
+
+class TestModelRunner:
+    @pytest.fixture
+    def log_model_result_catcher(self) -> EventCatcher:
+        return EventCatcher(event_to_catch=LogModelResult)
+
+    @pytest.fixture
+    def model_runner(
+        self,
+        postgres_adapter: PostgresAdapter,
+        table_model: ModelNode,
+        runtime_config: RuntimeConfig,
+    ) -> ModelRunner:
+        return ModelRunner(
+            config=runtime_config,
+            adapter=postgres_adapter,
+            node=table_model,
+            node_index=1,
+            num_nodes=1,
+        )
+
+    @pytest.fixture
+    def run_result(self, table_model: ModelNode) -> RunResult:
+        return RunResult(
+            status=RunStatus.Success,
+            timing=[],
+            thread_id="an_id",
+            execution_time=0,
+            adapter_response={},
+            message="It did it",
+            failures=None,
+            node=table_model,
+        )
+
+    def test_print_result_line(
+        self,
+        log_model_result_catcher: EventCatcher,
+        model_runner: ModelRunner,
+        run_result: RunResult,
+    ) -> None:
+        # Setup way to catch events
+        event_manager = get_event_manager()
path: tests/unit/task/test_run.py
text: @user1:
Should this be some setup steps vs just something in a single test?

@author:
We can move this to the local `log_model_result_catcher` fixture. Additionally, the setup can be reduced by a few steps because, in the time since originally writing this, we've released a version of dbt-common which provides a `add_callback_to_manager` method. So the `log_model_result_catcher` fixture would become
```python
@user2.fixture
def log_model_result_catcher(self) -> EventCatcher:
    catcher = EventCatcher(event_to_catch=LogModelResult)
    add_callback_to_manager(catcher.catch)
    return catcher
```