Repository: PrefectHQ/prefect
Pull request: Resolve dbt profiles and other improvements
<!-- 
Thanks for opening a pull request to Prefect! 
If this is your first contribution, please make sure to review our contribution guidelines: https://docs.prefect.io/contribute/index
-->

<!-- Include an overview of the proposed changes here -->
Closes #18430

This PR makes a number of improvements and fixes to `prefect-dbt` intended for inclusion in `prefect-dbt==0.7.2`:
- Fixes resolution of templated Prefect blocks and Prefect variables in `profiles.yml`
- Fixes a logging race condition, ensuring all logs about a node are associated with their corresponding Prefect task run
- Prevents skipped nodes from getting task runs so skipped nodes don't give a false impression that they ran, and don't create or update assets. Includes children of nodes whose tests fail if using or retrying the `build` command

### Checklist
<!-- These boxes may be checked after opening the pull request. -->

- [ ] This pull request references any related issue by including "closes `<link to issue>`"
  - If no issue exists and your change is not a small fix, please [create an issue](https://github.com/PrefectHQ/prefect/issues/new/choose) first.
- [x] If this pull request adds new functionality, it includes unit tests that cover the changes
- [ ] If this pull request removes docs files, it includes redirect settings in `mint.json`.
- [x] If this pull request adds functions or classes, it includes helpful docstrings.


Problem statement:
prefect-dbt - template rendering stopped working in 0.7.1
### Bug summary

During the release candidate for prefect dbt (0.7.0rc1) - template rendering as defined in the docs was working. 

However after upgrading to 0.7.1 - same flow no changes is erroring out with:

```
ValueError: Failed to invoke dbt command 'debug': Compilation Error
  Could not render {{ prefect.variables.target }}: 'prefect' is undefined
```

my profiles.yml looks similar to... :

```yaml
snowflake_dbt:
  target: "{{ prefect.variables.target }}" # complains about this on 0.7.1, fine in 0.7.0rc1
  outputs:
    dev:
      type: snowflake
      account: "{{ prefect.blocks.snowflake-credentials... }}"
      user: "{{ prefect.blocks.snowflake-credentials... }}"
      password: "{{ prefect.blocks.snowflake-credentials... }}"
      database: "{{ prefect.blocks.snowflake-connector... }}"
      schema: "{{ prefect.blocks.snowflake-connector...}}"
      warehouse: "{{ prefect.blocks.snowflake-connector... }}"
      threads: 12
```

Downgrade to the release candidate and I can run the flow.

flow code:

```python
from pathlib import Path

from prefect import flow, get_run_logger, task
from prefect_dbt import PrefectDbtRunner, PrefectDbtSettings
from prefect_github.repository import GitHubRepository
from pydantic import BaseModel, Field


class DbtConfig(BaseModel):
    """Represents a configuration for running dbt commands."""

    project_dir: str = Field(
        ...,
        description="The path to the dbt project directory.",
    )
    commands: list[str] = Field(
        ["debug"],
        description="The dbt commands to run. Defaults to ['debug']. Omit the dbt keyword.",
    )  


@task
def execute_dbt_command(
    settings: PrefectDbtSettings,
    command: str,
) -> None:
    logger = get_run_logger()
    logger.info(f"Executing dbt command: {command}")

    args = command.split(" ")
    PrefectDbtRunner(settings=settings).invoke(args=args)


@flow
def invoke_dbt(dbt_config: DbtConfig) -> None:
    """Invoke dbt commands for a given dbt project."""

    settings = PrefectDbtSettings(
        project_dir=Path(dbt_config.project_dir),
        profiles_dir=Path(dbt_config.project_dir) / ".prefect",
    )

    for command in dbt_config.commands:
        execute_dbt_command(settings, command)


if __name__ == "__main__":
    invoke_dbt(
        DbtConfig(
            project_dir="/path/to/dbt/project",
            commands=[
                "debug"
            ],
        )
    )
```



### Version info

```Text
Version:             3.4.7
API version:         0.8.4
Python version:      3.13.2
Git commit:          cd81d15a
Built:               Thu, Jun 26, 2025 09:16 PM
OS/Arch:             darwin/arm64
Profile:             dev
Server type:         cloud
Pydantic version:    2.11.7
Integrations:
  prefect-slack:     0.3.1
  prefect-dbt:       0.7.1
  prefect-github:    0.3.1
  prefect-snowflake: 0.28.5
  prefect-aws:       0.5.11
  prefect-shell:     0.3.1
```

### Additional context

_No response_

Hints from issue discussion:

Commit message: test for skipped nodes

Patch to review:
diff --git a/src/integrations/prefect-dbt/prefect_dbt/core/_tracker.py b/src/integrations/prefect-dbt/prefect_dbt/core/_tracker.py
index 933f273dc83f..8ca272378c8a 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/core/_tracker.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/core/_tracker.py
@@ -3,10 +3,12 @@
 """
 
 import threading
-from typing import Any, Union
+from typing import Any, Optional, Union
+from uuid import UUID
 
-from prefect.client.schemas.objects import State
+from prefect.client.schemas.objects import Flow, State
 from prefect.context import hydrated_context
+from prefect.logging.loggers import PrefectLogAdapter, get_logger
 from prefect.task_engine import run_task_sync
 from prefect.tasks import Task
 
@@ -19,12 +21,13 @@ class NodeTaskTracker:
 
     def __init__(self):
         self._tasks: dict[str, Task[Any, Any]] = {}
-        self._task_loggers: dict[str, Any] = {}
         self._task_results: dict[str, Any] = {}
         self._node_status: dict[str, dict[str, Any]] = {}
         self._node_complete: dict[str, bool] = {}
         self._node_dependencies: dict[str, list[str]] = {}
         self._node_events: dict[str, threading.Event] = {}
+        self._task_run_ids: dict[str, UUID] = {}
+        self._task_run_names: dict[str, str] = {}
 
     def start_task(self, node_id: str, task: Task[Any, Any]) -> None:
         """Start a task for a node."""
@@ -32,13 +35,30 @@ def start_task(self, node_id: str, task: Task[Any, Any]) -> None:
         self._node_complete[node_id] = False
         self._node_events[node_id] = threading.Event()
 
-    def set_task_logger(self, node_id: str, logger: Any) -> None:
-        """Set the logger for a task."""
-        self._task_loggers[node_id] = logger
-
-    def get_task_logger(self, node_id: str) -> Union[Any, None]:
+    def get_task_logger(
+        self,
+        node_id: str,
+        flow_run: Optional[dict[str, Any]] = None,
+        flow: Optional[Flow] = None,
+        **kwargs: Any,
+    ) -> PrefectLogAdapter:
         """Get the logger for a task."""
-        return self._task_loggers.get(node_id)
+        logger = PrefectLogAdapter(
+            get_logger("prefect.task_runs"),
+            extra={
+                **{
+                    "task_run_id": self.get_task_run_id(node_id),
+                    "flow_run_id": str(flow_run.get("id")) if flow_run else "<unknown>",
+                    "task_run_name": self.get_task_run_name(node_id),
+                    "task_name": "execute_dbt_node",
+                    "flow_run_name": flow_run.get("name") if flow_run else "<unknown>",
+                    "flow_name": flow.name if flow else "<unknown>",
+                },
+                **kwargs,
+            },
+        )
+
+        return logger
 
     def set_node_status(
         self, node_id: str, event_data: dict[str, Any], event_message: str
@@ -96,10 +116,27 @@ def get_node_dependencies(self, node_id: str) -> list[str]:
         """Get the dependencies for a node."""
         return self._node_dependencies.get(node_id, [])
 
+    def set_task_run_id(self, node_id: str, task_run_id: UUID) -> None:
+        """Set the task run ID for a node."""
+        self._task_run_ids[node_id] = task_run_id
+
+    def get_task_run_id(self, node_id: str) -> Union[UUID, None]:
+        """Get the task run ID for a node."""
+        return self._task_run_ids.get(node_id)
+
+    def set_task_run_name(self, node_id: str, task_run_name: str) -> None:
+        """Set the task run name for a node."""
+        self._task_run_names[node_id] = task_run_name
+
+    def get_task_run_name(self, node_id: str) -> Union[str, None]:
+        """Get the task run name for a node."""
+        return self._task_run_names.get(node_id)
+
     def run_task_in_thread(
         self,
         node_id: str,
         task: Task[Any, Any],
+        task_run_id: UUID,
         parameters: dict[str, Any],
         context: dict[str, Any],
     ) -> None:
@@ -116,6 +153,7 @@ def run_task():
 
                 state = run_task_sync(
                     task,
+                    task_run_id=task_run_id,
                     parameters=parameters,
                     wait_for=states,
                     context=context,
diff --git a/src/integrations/prefect-dbt/prefect_dbt/core/runner.py b/src/integrations/prefect-dbt/prefect_dbt/core/runner.py
index bebac42dfe7f..e5487664f8ae 100644
--- a/src/integrations/prefect-dbt/prefect_dbt/core/runner.py
+++ b/src/integrations/prefect-dbt/prefect_dbt/core/runner.py
@@ -16,13 +16,21 @@
 )
 from dbt.artifacts.schemas.run import RunExecutionResult
 from dbt.cli.main import dbtRunner
+from dbt.compilation import Linker
+from dbt.config.project import Project
+from dbt.config.renderer import DbtProjectYamlRenderer
 from dbt.config.runtime import RuntimeConfig
 from dbt.contracts.graph.manifest import Manifest
 from dbt.contracts.graph.nodes import ManifestNode
+from dbt.contracts.state import (
+    load_result_state,  # type: ignore[reportUnknownMemberType]
+)
+from dbt.graph.graph import Graph, UniqueId
 from dbt_common.events.base_types import EventLevel, EventMsg
 from google.protobuf.json_format import MessageToDict
 
 from prefect import get_client, get_run_logger
+from prefect._internal.uuid7 import uuid7
 from prefect.assets import Asset, AssetProperties
 from prefect.cache_policies import NO_CACHE
 from prefect.client.orchestration import PrefectClient
@@ -43,6 +51,11 @@
     NodeStatus.Fail,
     NodeStatus.RuntimeErr,
 ]
+SKIPPED_STATUSES = [
+    RunStatus.Skipped,
+    TestStatus.Skipped,
+    NodeStatus.Skipped,
+]
 MATERIALIZATION_NODE_TYPES = [
     NodeType.Model,
     NodeType.Seed,
@@ -72,8 +85,6 @@ def execute_dbt_node(
     2. Wait for the node to finish using efficient threading.Event
     3. Check the node's status and fail if it's in a failure state
     """
-    task_state.set_task_logger(node_id, get_run_logger())
-
     # Wait for the node to finish using efficient threading.Event
     task_state.wait_for_node_completion(node_id)
 
@@ -123,11 +134,14 @@ def __init__(
         self.include_compiled_code = include_compiled_code
         self._force_nodes_as_tasks = _force_nodes_as_tasks
 
-        self._project: Optional[RuntimeConfig] = None
+        self._project: Optional[Project] = None
         self._target_path: Optional[Path] = None
         self._profiles_dir: Optional[Path] = None
         self._project_dir: Optional[Path] = None
         self._log_level: Optional[EventLevel] = None
+        self._config: Optional[RuntimeConfig] = None
+        self._graph: Optional[Graph] = None
+        self._skipped_nodes: set[str] = set()
 
     @property
     def target_path(self) -> Path:
@@ -153,12 +167,27 @@ def manifest(self) -> Manifest:
         return self._manifest
 
     @property
-    def project(self) -> RuntimeConfig:
+    def project(self) -> Project:
         if self._project is None:
             self._set_project_from_project_dir()
             assert self._project is not None  # for type checking
         return self._project
 
+    @property
+    def graph(self) -> Graph:
+        if self._graph is None:
+            self._set_graph_from_manifest()
+            assert self._graph is not None  # for type checking
+        return self._graph
+
+    def _set_graph_from_manifest(self, add_test_edges: bool = False):
+        linker = Linker()
+        linker.link_graph(self.manifest)
+        if add_test_edges:
+            self.manifest.build_parent_and_child_maps()
+            linker.add_test_edges(self.manifest)
+        self._graph = Graph(linker.graph)
+
     def _set_manifest_from_project_dir(self):
         try:
             with open(
@@ -171,7 +200,8 @@ def _set_manifest_from_project_dir(self):
             )
 
     def _set_project_from_project_dir(self):
-        self._project = RuntimeConfig.from_args(self.project_dir)
+        renderer = DbtProjectYamlRenderer(profile=None, cli_vars=None)
+        self._project = Project.from_project_root(str(self.project_dir), renderer)
 
     def _get_node_prefect_config(
         self, manifest_node: ManifestNode
@@ -330,9 +360,14 @@ def _call_task(
             [node[0].unique_id for node in upstream_manifest_nodes],
         )
 
+        task_run_id = uuid7()
+
+        task_state.set_task_run_id(manifest_node.unique_id, task_run_id)
+
         task_state.run_task_in_thread(
             manifest_node.unique_id,
             task,
+            task_run_id=task_run_id,
             parameters={
                 "task_state": task_state,
                 "node_id": manifest_node.unique_id,
@@ -355,9 +390,19 @@ def _create_logging_callback(
 
         def logging_callback(event: EventMsg):
             event_data = MessageToDict(event.data, preserving_proto_field_name=True)
-            if event_data.get("node_info"):
-                node_id = self._get_dbt_event_node_id(event)
-                logger = task_state.get_task_logger(node_id)
+            if (
+                event_data.get("node_info")
+                and (node_id := self._get_dbt_event_node_id(event))
+                not in self._skipped_nodes
+            ):
+                flow_run_context: Optional[dict[str, Any]] = context.get(
+                    "flow_run_context"
+                )
+                logger = task_state.get_task_logger(
+                    node_id,
+                    flow_run_context.get("flow_run") if flow_run_context else None,
+                    flow_run_context.get("flow") if flow_run_context else None,
+                )
             else:
                 try:
                     with hydrated_context(context) as run_context:
@@ -392,6 +437,9 @@ def _create_node_started_callback(
         def node_started_callback(event: EventMsg):
             if event.info.name == "NodeStart":
                 node_id = self._get_dbt_event_node_id(event)
+                if node_id in self._skipped_nodes:
+                    return
+
                 manifest_node, prefect_config = self._get_manifest_node_and_config(
                     node_id
                 )
@@ -410,12 +458,16 @@ def node_started_callback(event: EventMsg):
     def _create_node_finished_callback(
         self,
         task_state: NodeTaskTracker,
+        add_test_edges: bool = False,
     ) -> Callable[[EventMsg], None]:
         """Creates a callback function for ending tasks when nodes finish."""
 
         def node_finished_callback(event: EventMsg):
             if event.info.name == "NodeFinished":
                 node_id = self._get_dbt_event_node_id(event)
+                if node_id in self._skipped_nodes:
+                    return
+
                 manifest_node, _ = self._get_manifest_node_and_config(node_id)
 
                 if manifest_node:
@@ -426,6 +478,23 @@ def node_finished_callback(event: EventMsg):
                         )
                         event_message = self.get_dbt_event_msg(event)
                         task_state.set_node_status(node_id, event_data, event_message)
+
+                        node_info: Optional[dict[str, Any]] = event_data.get(
+                            "node_info"
+                        )
+                        node_status: Optional[str] = (
+                            node_info.get("node_status") if node_info else None
+                        )
+
+                        if (
+                            node_status in SKIPPED_STATUSES
+                            or node_status in FAILURE_STATUSES
+                        ):
+                            self._set_graph_from_manifest(add_test_edges=add_test_edges)
+                            for dep_node_id in self.graph.get_dependent_nodes(  # type: ignore[reportUnknownMemberType]
+                                UniqueId(node_id)
+                            ):  # type: ignore[reportUnknownMemberType]
+                                self._skipped_nodes.add(dep_node_id)  # type: ignore[reportUnknownMemberType]
                     except Exception as e:
                         print(e)
 
@@ -509,11 +578,28 @@ def invoke(self, args: list[str], **kwargs: Any):
         )
         task_state = NodeTaskTracker()
 
+        add_test_edges = True if "build" in args_copy else False
+
+        if "retry" in args_copy:
+            previous_results = load_result_state(
+                Path(self.project_dir) / self.target_path / "run_results.json"
+            )
+            if not previous_results:
+                raise ValueError(
+                    f"Cannot retry. No previous results found at target path {self.target_path}"
+                )
+            previous_args = previous_results.args
+            self.previous_command_name = previous_args.get("which")
+            if self.previous_command_name == "build":
+                add_test_edges = True
+
         callbacks = (
             [
                 self._create_logging_callback(task_state, self.log_level, context),
                 self._create_node_started_callback(task_state, context),
-                self._create_node_finished_callback(task_state),
+                self._create_node_finished_callback(
+                    task_state, add_test_edges=add_test_edges
+                ),
             ]
             if in_flow_or_task_run or self._force_nodes_as_tasks
             else []
@@ -529,10 +615,13 @@ def invoke(self, args: list[str], **kwargs: Any):
             **kwargs,
         }
 
-        res = dbtRunner(callbacks=callbacks).invoke(  # type: ignore[reportUnknownMemberType]
-            args_copy,
-            **invoke_kwargs,
-        )
+        with self.settings.resolve_profiles_yml() as profiles_dir:
+            invoke_kwargs["profiles_dir"] = profiles_dir
+
+            res = dbtRunner(callbacks=callbacks).invoke(  # type: ignore[reportUnknownMemberType]
+                args_copy,
+                **invoke_kwargs,
+            )
 
         if not res.success and res.exception:
             raise ValueError(
diff --git a/src/integrations/prefect-dbt/tests/core/test_runner.py b/src/integrations/prefect-dbt/tests/core/test_runner.py
index 266675f4a6e4..91ad120ecfc4 100644
--- a/src/integrations/prefect-dbt/tests/core/test_runner.py
+++ b/src/integrations/prefect-dbt/tests/core/test_runner.py
@@ -3,13 +3,13 @@
 """
 
 from pathlib import Path
+from typing import Any
 from unittest.mock import Mock
 
 import pytest
 from dbt.artifacts.resources.types import NodeType
 from dbt.artifacts.schemas.results import RunStatus
 from dbt.artifacts.schemas.run import RunExecutionResult
-from dbt.config.runtime import RuntimeConfig
 from dbt.contracts.graph.manifest import Manifest
 from dbt.contracts.graph.nodes import ManifestNode
 from dbt_common.events.base_types import EventLevel, EventMsg
@@ -17,6 +17,307 @@
 from prefect_dbt.core.settings import PrefectDbtSettings
 
 
+@pytest.fixture
+def mock_dbt_runner_instance():
+    """Fixture providing a mock dbt runner instance."""
+    mock_instance = Mock()
+    mock_instance.invoke.return_value = Mock(success=True, exception=None)
+    return mock_instance
+
+
+@pytest.fixture
+def mock_dbt_runner_class(mock_dbt_runner_instance, monkeypatch):
+    """Fixture providing a mock dbt runner class."""
+    mock_class = Mock(return_value=mock_dbt_runner_instance)
+    monkeypatch.setattr("prefect_dbt.core.runner.dbtRunner", mock_class)
+    return mock_class
+
+
+@pytest.fixture
+def mock_settings():
+    """Fixture providing a mock settings object with resolve_profiles_yml method."""
+    mock_settings = Mock()
+    mock_settings.resolve_profiles_yml.return_value.__enter__ = Mock(
+        return_value="/mock/profiles/dir"
+    )
+    mock_settings.resolve_profiles_yml.return_value.__exit__ = Mock(return_value=None)
+    return mock_settings
+
+
+@pytest.fixture
+def mock_log_level():
+    """Fixture providing a mock log level with value attribute."""
+    mock_log_level = Mock()
+    mock_log_level.value = "info"
+    return mock_log_level
+
+
+@pytest.fixture
+def mock_runner_with_common_mocks(monkeypatch, mock_settings, mock_log_level):
+    """Fixture providing a runner with common mocks applied."""
+    runner = PrefectDbtRunner()
+
+    # Mock the callback methods
+    monkeypatch.setattr(runner, "_create_logging_callback", Mock())
+    monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
+    monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
+
+    # Mock the log_level property
+    monkeypatch.setattr(
+        PrefectDbtRunner, "log_level", property(lambda self: mock_log_level)
+    )
+
+    # Set mock settings
+    runner.settings = mock_settings
+
+    return runner
+
+
+@pytest.fixture
+def mock_manifest_node():
+    """Fixture providing a mock manifest node."""
+    mock_node = Mock(spec=ManifestNode)
+    mock_node.resource_type = NodeType.Model
+    mock_node.unique_id = "test_node"
+    mock_node.relation_name = "test_relation"
+    mock_node.name = "test_model"
+    mock_node.description = "Test model"
+    mock_node.config = Mock()
+    mock_node.config.meta = {}
+    return mock_node
+
+
+@pytest.fixture
+def mock_manifest():
+    """Fixture providing a mock manifest."""
+    mock_manifest = Mock(spec=Manifest)
+    mock_manifest.metadata = Mock()
+    mock_manifest.metadata.adapter_type = "snowflake"
+    mock_manifest.nodes = {}
+    return mock_manifest
+
+
+@pytest.fixture
+def mock_successful_result():
+    """Fixture providing a mock successful dbt result."""
+    mock_result = Mock()
+    mock_result.success = True
+    mock_result.exception = None
+    return mock_result
+
+
+@pytest.fixture
+def mock_dbt_runner_with_success(mock_successful_result):
+    """Fixture providing a mock dbt runner instance that returns successful results."""
+    mock_instance = Mock()
+    mock_instance.invoke.return_value = mock_successful_result
+    return mock_instance
+
+
+@pytest.fixture
+def mock_dbt_runner_class_with_success(mock_dbt_runner_with_success, monkeypatch):
+    """Fixture providing a mock dbt runner class that returns successful results."""
+    mock_class = Mock(return_value=mock_dbt_runner_with_success)
+    monkeypatch.setattr("prefect_dbt.core.runner.dbtRunner", mock_class)
+    return mock_class
+
+
+@pytest.fixture
+def mock_runner_with_all_mocks(
+    monkeypatch, mock_settings, mock_log_level, mock_dbt_runner_class_with_success
+):
+    """Fixture providing a runner with all common mocks applied."""
+    runner = PrefectDbtRunner()
+
+    # Mock the callback methods
+    monkeypatch.setattr(runner, "_create_logging_callback", Mock())
+    monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
+    monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
+
+    # Mock the log_level property
+    monkeypatch.setattr(
+        PrefectDbtRunner, "log_level", property(lambda self: mock_log_level)
+    )
+
+    # Set mock settings
+    runner.settings = mock_settings
+
+    return runner
+
+
+@pytest.fixture
+def mock_context_manager():
+    """Fixture providing a mock context manager for hydrated_context."""
+    mock_context = Mock()
+    mock_context.__enter__ = Mock()
+    mock_context.__exit__ = Mock()
+    return mock_context
+
+
+@pytest.fixture
+def mock_context_manager_with_patch(monkeypatch, mock_context_manager):
+    """Fixture providing a mock context manager with monkeypatch applied."""
+    monkeypatch.setattr(
+        "prefect_dbt.core._tracker.hydrated_context",
+        Mock(return_value=mock_context_manager),
+    )
+    return mock_context_manager
+
+
+@pytest.fixture
+def mock_failed_result():
+    """Fixture providing a mock failed dbt result."""
+    mock_result = Mock()
+    mock_result.success = False
+    mock_result.exception = "Test exception"
+    return mock_result
+
+
+@pytest.fixture
+def mock_dbt_runner_with_failure(mock_failed_result):
+    """Fixture providing a mock dbt runner instance that returns failed results."""
+    mock_instance = Mock()
+    mock_instance.invoke.return_value = mock_failed_result
+    return mock_instance
+
+
+@pytest.fixture
+def mock_dbt_runner_class_with_failure(mock_dbt_runner_with_failure, monkeypatch):
+    """Fixture providing a mock dbt runner class that returns failed results."""
+    mock_class = Mock(return_value=mock_dbt_runner_with_failure)
+    monkeypatch.setattr("prefect_dbt.core.runner.dbtRunner", mock_class)
+    return mock_class
+
+
+@pytest.fixture
+def mock_runner_with_callback_mocks(monkeypatch, mock_settings, mock_log_level):
+    """Fixture providing a runner with callback mocks applied."""
+    runner = PrefectDbtRunner()
+
+    # Mock the callback methods
+    mock_logging_callback = Mock()
+    mock_node_started_callback = Mock()
+    mock_node_finished_callback = Mock()
+    monkeypatch.setattr(
+        runner, "_create_logging_callback", Mock(return_value=mock_logging_callback)
+    )
+    monkeypatch.setattr(
+        runner,
+        "_create_node_started_callback",
+        Mock(return_value=mock_node_started_callback),
+    )
+    monkeypatch.setattr(
+        runner,
+        "_create_node_finished_callback",
+        Mock(return_value=mock_node_finished_callback),
+    )
+
+    # Mock the log_level property
+    monkeypatch.setattr(
+        PrefectDbtRunner, "log_level", property(lambda self: mock_log_level)
+    )
+
+    # Set mock settings
+    runner.settings = mock_settings
+
+    return (
+        runner,
+        mock_logging_callback,
+        mock_node_started_callback,
+        mock_node_finished_callback,
+    )
+
+
+@pytest.fixture
+def mock_runner_with_force_nodes_as_tasks(monkeypatch, mock_settings, mock_log_level):
+    """Fixture providing a runner with _force_nodes_as_tasks=True and common mocks."""
+    runner = PrefectDbtRunner(_force_nodes_as_tasks=True)
+
+    # Mock the callback methods
+    mock_logging_callback = Mock()
+    mock_node_started_callback = Mock()
+    mock_node_finished_callback = Mock()
+    monkeypatch.setattr(
+        runner, "_create_logging_callback", Mock(return_value=mock_logging_callback)
+    )
+    monkeypatch.setattr(
+        runner,
+        "_create_node_started_callback",
+        Mock(return_value=mock_node_started_callback),
+    )
+    monkeypatch.setattr(
+        runner,
+        "_create_node_finished_callback",
+        Mock(return_value=mock_node_finished_callback),
+    )
+
+    # Mock the log_level property
+    monkeypatch.setattr(
+        PrefectDbtRunner, "log_level", property(lambda self: mock_log_level)
+    )
+
+    # Set mock settings
+    runner.settings = mock_settings
+
+    return (
+        runner,
+        mock_logging_callback,
+        mock_node_started_callback,
+        mock_node_finished_callback,
+    )
+
+
+@pytest.fixture
+def mock_flow_context(monkeypatch):
+    """Fixture providing a mock flow context."""
+    context = {"flow_run_context": {"foo": "bar"}}
+    monkeypatch.setattr(
+        "prefect_dbt.core.runner.serialize_context",
+        Mock(return_value=context),
+    )
+    return context
+
+
+@pytest.fixture
+def mock_task_context(monkeypatch):
+    """Fixture providing a mock task context."""
+    context = {"task_run_context": {"foo": "bar"}}
+    monkeypatch.setattr(
+        "prefect_dbt.core.runner.serialize_context",
+        Mock(return_value=context),
+    )
+    return context
+
+
+@pytest.fixture
+def mock_empty_context(monkeypatch):
+    """Fixture providing a mock empty context."""
+    context = {}
+    monkeypatch.setattr(
+        "prefect_dbt.core.runner.serialize_context",
+        Mock(return_value=context),
+    )
+    return context
+
+
+@pytest.fixture
+def mock_project():
+    """Fixture providing a mock project."""
+    mock_project = Mock()
+    mock_project.project_name = "test_project"
+    return mock_project
+
+
+@pytest.fixture
+def mock_file_operations(monkeypatch):
+    """Fixture providing mock file operations."""
+    mock_open = Mock()
+    mock_open.return_value.__enter__ = Mock(return_value=Mock())
+    mock_open.return_value.__exit__ = Mock(return_value=None)
+    monkeypatch.setattr("builtins.open", mock_open)
+    return mock_open
+
+
 class TestPrefectDbtRunner:
     """Test cases focusing on PrefectDbtRunner outcomes and behavior."""
 
@@ -151,26 +452,14 @@ def test_runner_updates_settings_from_cli_flags(self):
         assert result_args == ["run"]
         assert runner._target_path == Path("/custom/path")
 
-    def test_runner_invokes_dbt_successfully(self, monkeypatch: pytest.MonkeyPatch):
+    def test_runner_invokes_dbt_successfully(
+        self,
+        mock_runner_with_all_mocks,
+        mock_dbt_runner_class_with_success,
+        monkeypatch,
+    ):
         """Test that runner invokes dbt commands successfully."""
-        runner = PrefectDbtRunner()
-
-        mock_result = Mock()
-        mock_result.success = True
-        mock_result.exception = None
-
-        mock_dbt_runner_instance = Mock()
-        mock_dbt_runner_instance.invoke.return_value = mock_result
-
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.dbtRunner",
-            Mock(return_value=mock_dbt_runner_instance),
-        )
-
-        # Mock the callback methods
-        monkeypatch.setattr(runner, "_create_logging_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
+        runner = mock_runner_with_all_mocks
 
         # Mock context to simulate no flow/task run
         monkeypatch.setattr(
@@ -179,59 +468,32 @@ def test_runner_invokes_dbt_successfully(self, monkeypatch: pytest.MonkeyPatch):
 
         result = runner.invoke(["run"])
 
-        assert result == mock_result
-        mock_dbt_runner_instance.invoke.assert_called_once()
+        assert result.success is True
+        assert result.exception is None
+        mock_dbt_runner_class_with_success.return_value.invoke.assert_called_once()
         # No assertion about callbacks in invoke call; callbacks are passed to dbtRunner constructor only.
 
     def test_runner_invokes_dbt_with_callbacks_in_flow_context(
-        self, monkeypatch: pytest.MonkeyPatch
+        self,
+        mock_runner_with_callback_mocks,
+        mock_dbt_runner_class_with_success,
+        mock_flow_context,
     ):
         """Test that runner adds callbacks when in flow context."""
-        runner = PrefectDbtRunner()
-
-        mock_result = Mock()
-        mock_result.success = True
-        mock_result.exception = None
-
-        mock_dbt_runner_instance = Mock()
-        mock_dbt_runner_instance.invoke.return_value = mock_result
-
-        mock_dbt_runner_class = Mock(return_value=mock_dbt_runner_instance)
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.dbtRunner",
-            mock_dbt_runner_class,
-        )
-
-        # Mock the callback methods
-        mock_logging_callback = Mock()
-        mock_node_started_callback = Mock()
-        mock_node_finished_callback = Mock()
-        monkeypatch.setattr(
-            runner, "_create_logging_callback", Mock(return_value=mock_logging_callback)
-        )
-        monkeypatch.setattr(
-            runner,
-            "_create_node_started_callback",
-            Mock(return_value=mock_node_started_callback),
-        )
-        monkeypatch.setattr(
+        (
             runner,
-            "_create_node_finished_callback",
-            Mock(return_value=mock_node_finished_callback),
-        )
-
-        # Mock context to simulate flow run (non-empty dict)
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.serialize_context",
-            Mock(return_value={"flow_run_context": {"foo": "bar"}}),
-        )
+            mock_logging_callback,
+            mock_node_started_callback,
+            mock_node_finished_callback,
+        ) = mock_runner_with_callback_mocks
 
         result = runner.invoke(["run"])
 
-        assert result == mock_result
-        mock_dbt_runner_instance.invoke.assert_called_once()
+        assert result.success is True
+        assert result.exception is None
+        mock_dbt_runner_class_with_success.return_value.invoke.assert_called_once()
         # Verify callbacks are added when in flow context
-        mock_dbt_runner_class.assert_called_once_with(
+        mock_dbt_runner_class_with_success.assert_called_once_with(
             callbacks=[
                 mock_logging_callback,
                 mock_node_started_callback,
@@ -240,109 +502,59 @@ def test_runner_invokes_dbt_with_callbacks_in_flow_context(
         )
 
     def test_runner_invokes_dbt_with_callbacks_in_task_context(
-        self, monkeypatch: pytest.MonkeyPatch
+        self,
+        mock_runner_with_callback_mocks,
+        mock_dbt_runner_class_with_success,
+        mock_task_context,
     ):
         """Test that runner adds callbacks when in task context."""
-        runner = PrefectDbtRunner()
-
-        mock_result = Mock()
-        mock_result.success = True
-        mock_result.exception = None
-
-        mock_dbt_runner_instance = Mock()
-        mock_dbt_runner_instance.invoke.return_value = mock_result
-
-        mock_dbt_runner_class = Mock(return_value=mock_dbt_runner_instance)
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.dbtRunner",
-            mock_dbt_runner_class,
-        )
-
-        # Mock the callback methods
-        mock_logging_callback = Mock()
-        mock_node_started_callback = Mock()
-        mock_node_finished_callback = Mock()
-        monkeypatch.setattr(
-            runner, "_create_logging_callback", Mock(return_value=mock_logging_callback)
-        )
-        monkeypatch.setattr(
-            runner,
-            "_create_node_started_callback",
-            Mock(return_value=mock_node_started_callback),
-        )
-        monkeypatch.setattr(
+        (
             runner,
-            "_create_node_finished_callback",
-            Mock(return_value=mock_node_finished_callback),
-        )
-
-        # Mock context to simulate task run (non-empty dict)
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.serialize_context",
-            Mock(return_value={"task_run_context": {"foo": "bar"}}),
-        )
+            mock_logging_callback,
+            mock_node_started_callback,
+            mock_node_finished_callback,
+        ) = mock_runner_with_callback_mocks
 
         result = runner.invoke(["run"])
 
-        assert result == mock_result
-        mock_dbt_runner_instance.invoke.assert_called_once()
+        assert result.success is True
+        assert result.exception is None
+        mock_dbt_runner_class_with_success.return_value.invoke.assert_called_once()
         # Verify callbacks are added when in task context
-        mock_dbt_runner_class.assert_called_once_with(
+        mock_dbt_runner_class_with_success.assert_called_once_with(
             callbacks=[
                 mock_logging_callback,
                 mock_node_started_callback,
                 mock_node_finished_callback,
             ]
         )
+        call_kwargs = mock_dbt_runner_class_with_success.return_value.invoke.call_args[
+            1
+        ]
+        assert call_kwargs["log_level"] == "none"
+        assert call_kwargs["log_level_file"] == "info"
 
     def test_runner_invokes_dbt_with_force_nodes_as_tasks(
-        self, monkeypatch: pytest.MonkeyPatch
+        self,
+        mock_runner_with_force_nodes_as_tasks,
+        mock_dbt_runner_class_with_success,
+        mock_empty_context,
     ):
         """Test that runner adds callbacks when force_nodes_as_tasks is True."""
-        runner = PrefectDbtRunner(_force_nodes_as_tasks=True)
-
-        mock_result = Mock()
-        mock_result.success = True
-        mock_result.exception = None
-
-        mock_dbt_runner_instance = Mock()
-        mock_dbt_runner_instance.invoke.return_value = mock_result
-
-        mock_dbt_runner_class = Mock(return_value=mock_dbt_runner_instance)
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.dbtRunner",
-            mock_dbt_runner_class,
-        )
-
-        # Mock the callback methods
-        mock_logging_callback = Mock()
-        mock_node_started_callback = Mock()
-        mock_node_finished_callback = Mock()
-        monkeypatch.setattr(
-            runner, "_create_logging_callback", Mock(return_value=mock_logging_callback)
-        )
-        monkeypatch.setattr(
-            runner,
-            "_create_node_started_callback",
-            Mock(return_value=mock_node_started_callback),
-        )
-        monkeypatch.setattr(
+        (
             runner,
-            "_create_node_finished_callback",
-            Mock(return_value=mock_node_finished_callback),
-        )
-
-        # Mock context to simulate no flow/task run
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.serialize_context", Mock(return_value={})
-        )
+            mock_logging_callback,
+            mock_node_started_callback,
+            mock_node_finished_callback,
+        ) = mock_runner_with_force_nodes_as_tasks
 
         result = runner.invoke(["run"])
 
-        assert result == mock_result
-        mock_dbt_runner_instance.invoke.assert_called_once()
+        assert result.success is True
+        assert result.exception is None
+        mock_dbt_runner_class_with_success.return_value.invoke.assert_called_once()
         # Verify callbacks are added when _force_nodes_as_tasks is True
-        mock_dbt_runner_class.assert_called_once_with(
+        mock_dbt_runner_class_with_success.assert_called_once_with(
             callbacks=[
                 mock_logging_callback,
                 mock_node_started_callback,
@@ -351,27 +563,13 @@ def test_runner_invokes_dbt_with_force_nodes_as_tasks(
         )
 
     def test_runner_sets_log_level_none_in_flow_context(
-        self, monkeypatch: pytest.MonkeyPatch
+        self,
+        mock_runner_with_all_mocks,
+        mock_dbt_runner_class_with_success,
+        monkeypatch,
     ):
         """Test that runner sets log_level to 'none' when in flow context."""
-        runner = PrefectDbtRunner()
-
-        mock_result = Mock()
-        mock_result.success = True
-        mock_result.exception = None
-
-        mock_dbt_runner_instance = Mock()
-        mock_dbt_runner_instance.invoke.return_value = mock_result
-
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.dbtRunner",
-            Mock(return_value=mock_dbt_runner_instance),
-        )
-
-        # Mock the callback methods
-        monkeypatch.setattr(runner, "_create_logging_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
+        runner = mock_runner_with_all_mocks
 
         # Mock context to simulate flow run (non-empty dict)
         monkeypatch.setattr(
@@ -381,60 +579,174 @@ def test_runner_sets_log_level_none_in_flow_context(
 
         result = runner.invoke(["run"])
 
-        assert result == mock_result
-        mock_dbt_runner_instance.invoke.assert_called_once()
+        assert result.success is True
+        assert result.exception is None
+        mock_dbt_runner_class_with_success.return_value.invoke.assert_called_once()
         # Verify log_level is set to 'none' and log_level_file uses the original level as string
-        call_kwargs = mock_dbt_runner_instance.invoke.call_args[1]
+        call_kwargs = mock_dbt_runner_class_with_success.return_value.invoke.call_args[
+            1
+        ]
         assert call_kwargs["log_level"] == "none"
-        assert call_kwargs["log_level_file"] == runner.log_level.value
+        assert call_kwargs["log_level_file"] == "info"
 
     def test_runner_sets_log_level_none_in_task_context(
-        self, monkeypatch: pytest.MonkeyPatch
+        self,
+        mock_runner_with_callback_mocks,
+        mock_dbt_runner_class_with_success,
+        mock_task_context,
     ):
         """Test that runner sets log_level to 'none' when in task context."""
-        runner = PrefectDbtRunner()
+        runner, _, _, _ = mock_runner_with_callback_mocks
 
-        mock_result = Mock()
-        mock_result.success = True
-        mock_result.exception = None
+        result = runner.invoke(["run"])
 
-        mock_dbt_runner_instance = Mock()
-        mock_dbt_runner_instance.invoke.return_value = mock_result
+        assert result.success is True
+        assert result.exception is None
+        mock_dbt_runner_class_with_success.return_value.invoke.assert_called_once()
+        # Verify log_level is set to 'none' and log_level_file uses the original level as string
+        call_kwargs = mock_dbt_runner_class_with_success.return_value.invoke.call_args[
+            1
+        ]
+        assert call_kwargs["log_level"] == "none"
+        assert call_kwargs["log_level_file"] == "info"
 
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.dbtRunner",
-            Mock(return_value=mock_dbt_runner_instance),
-        )
+    def test_runner_uses_original_log_level_outside_context(
+        self,
+        mock_runner_with_callback_mocks,
+        mock_dbt_runner_class_with_success,
+        mock_empty_context,
+    ):
+        """Test that runner uses original log_level when not in flow/task context."""
+        runner, _, _, _ = mock_runner_with_callback_mocks
 
-        # Mock the callback methods
-        monkeypatch.setattr(runner, "_create_logging_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
+        result = runner.invoke(["run"])
 
-        # Mock context to simulate task run (non-empty dict)
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.serialize_context",
-            Mock(return_value={"task_run_context": {"foo": "bar"}}),
+        assert result.success is True
+        assert result.exception is None
+        mock_dbt_runner_class_with_success.return_value.invoke.assert_called_once()
+        # Verify log_level uses the original level when not in context
+        call_kwargs = mock_dbt_runner_class_with_success.return_value.invoke.call_args[
+            1
+        ]
+        assert call_kwargs["log_level"] == "info"
+        assert call_kwargs["log_level_file"] == "info"
+
+    def test_runner_uses_original_log_level_with_force_nodes_as_tasks(
+        self,
+        mock_runner_with_force_nodes_as_tasks,
+        mock_dbt_runner_class_with_success,
+        mock_empty_context,
+    ):
+        """Test that runner uses original log_level when force_nodes_as_tasks is True but not in context."""
+        runner, _, _, _ = mock_runner_with_force_nodes_as_tasks
+
+        result = runner.invoke(["run"])
+
+        assert result.success is True
+        assert result.exception is None
+        mock_dbt_runner_class_with_success.return_value.invoke.assert_called_once()
+        # Verify log_level uses the original level even with _force_nodes_as_tasks
+        call_kwargs = mock_dbt_runner_class_with_success.return_value.invoke.call_args[
+            1
+        ]
+        assert call_kwargs["log_level"] == "info"
+        assert call_kwargs["log_level_file"] == "info"
+
+    def test_runner_force_nodes_as_tasks_with_flow_context(
+        self,
+        mock_runner_with_force_nodes_as_tasks,
+        mock_dbt_runner_class_with_success,
+        mock_flow_context,
+    ):
+        """Test that runner behavior when _force_nodes_as_tasks is True and in flow context."""
+        (
+            runner,
+            mock_logging_callback,
+            mock_node_started_callback,
+            mock_node_finished_callback,
+        ) = mock_runner_with_force_nodes_as_tasks
+
+        result = runner.invoke(["run"])
+
+        assert result.success is True
+        assert result.exception is None
+        mock_dbt_runner_class_with_success.return_value.invoke.assert_called_once()
+        # Verify callbacks are added and log_level is 'none' when in flow context
+        mock_dbt_runner_class_with_success.assert_called_once_with(
+            callbacks=[
+                mock_logging_callback,
+                mock_node_started_callback,
+                mock_node_finished_callback,
+            ]
         )
+        call_kwargs = mock_dbt_runner_class_with_success.return_value.invoke.call_args[
+            1
+        ]
+        assert call_kwargs["log_level"] == "none"
+        assert call_kwargs["log_level_file"] == "info"
+
+    def test_runner_force_nodes_as_tasks_with_task_context(
+        self,
+        mock_runner_with_force_nodes_as_tasks,
+        mock_dbt_runner_class_with_success,
+        mock_task_context,
+    ):
+        """Test that runner behavior when _force_nodes_as_tasks is True and in task context."""
+        (
+            runner,
+            mock_logging_callback,
+            mock_node_started_callback,
+            mock_node_finished_callback,
+        ) = mock_runner_with_force_nodes_as_tasks
 
         result = runner.invoke(["run"])
 
-        assert result == mock_result
-        mock_dbt_runner_instance.invoke.assert_called_once()
-        # Verify log_level is set to 'none' and log_level_file uses the original level as string
-        call_kwargs = mock_dbt_runner_instance.invoke.call_args[1]
+        assert result.success is True
+        assert result.exception is None
+        mock_dbt_runner_class_with_success.return_value.invoke.assert_called_once()
+        # Verify callbacks are added and log_level is 'none' when in task context
+        mock_dbt_runner_class_with_success.assert_called_once_with(
+            callbacks=[
+                mock_logging_callback,
+                mock_node_started_callback,
+                mock_node_finished_callback,
+            ]
+        )
+        call_kwargs = mock_dbt_runner_class_with_success.return_value.invoke.call_args[
+            1
+        ]
         assert call_kwargs["log_level"] == "none"
-        assert call_kwargs["log_level_file"] == runner.log_level.value
+        assert call_kwargs["log_level_file"] == "info"
 
-    def test_runner_uses_original_log_level_outside_context(
+    def test_runner_handles_dbt_exceptions(
+        self,
+        mock_runner_with_callback_mocks,
+        mock_dbt_runner_class_with_failure,
+        mock_empty_context,
+    ):
+        """Test that runner handles dbt exceptions gracefully."""
+        runner, _, _, _ = mock_runner_with_callback_mocks
+
+        with pytest.raises(ValueError, match="Failed to invoke dbt command"):
+            runner.invoke(["run"])
+
+    def test_runner_handles_dbt_failures_with_raise_on_failure_true(
         self, monkeypatch: pytest.MonkeyPatch
     ):
-        """Test that runner uses original log_level when not in flow/task context."""
-        runner = PrefectDbtRunner()
+        """Test that runner raises on dbt failures when raise_on_failure is True."""
+        runner = PrefectDbtRunner(raise_on_failure=True)
 
         mock_result = Mock()
-        mock_result.success = True
+        mock_result.success = False
         mock_result.exception = None
+        mock_result.result = Mock(spec=RunExecutionResult)
+        mock_result.result.results = [
+            Mock(
+                status=RunStatus.Error,
+                node=Mock(resource_type="model", name="test_model"),
+                message="Test error",
+            )
+        ]
 
         mock_dbt_runner_instance = Mock()
         mock_dbt_runner_instance.invoke.return_value = mock_result
@@ -454,24 +766,36 @@ def test_runner_uses_original_log_level_outside_context(
             "prefect_dbt.core.runner.serialize_context", Mock(return_value={})
         )
 
-        result = runner.invoke(["run"])
+        # Mock the settings resolve_profiles_yml method to prevent file access
+        mock_settings = Mock()
+        mock_settings.resolve_profiles_yml.return_value.__enter__ = Mock(
+            return_value="/mock/profiles/dir"
+        )
+        mock_settings.resolve_profiles_yml.return_value.__exit__ = Mock(
+            return_value=None
+        )
+        runner.settings = mock_settings
 
-        assert result == mock_result
-        mock_dbt_runner_instance.invoke.assert_called_once()
-        # Verify log_level uses the original level when not in context
-        call_kwargs = mock_dbt_runner_instance.invoke.call_args[1]
-        assert call_kwargs["log_level"] == runner.log_level
-        assert call_kwargs["log_level_file"] == runner.log_level
+        with pytest.raises(ValueError, match="Failures detected during invocation"):
+            runner.invoke(["run"])
 
-    def test_runner_uses_original_log_level_with_force_nodes_as_tasks(
+    def test_runner_handles_dbt_failures_with_raise_on_failure_false(
         self, monkeypatch: pytest.MonkeyPatch
     ):
-        """Test that runner uses original log_level when force_nodes_as_tasks is True but not in context."""
-        runner = PrefectDbtRunner(_force_nodes_as_tasks=True)
+        """Test that runner doesn't raise on dbt failures when raise_on_failure is False."""
+        runner = PrefectDbtRunner(raise_on_failure=False)
 
         mock_result = Mock()
-        mock_result.success = True
+        mock_result.success = False
         mock_result.exception = None
+        mock_result.result = Mock(spec=RunExecutionResult)
+        mock_result.result.results = [
+            Mock(
+                status=RunStatus.Error,
+                node=Mock(resource_type="model", name="test_model"),
+                message="Test error",
+            )
+        ]
 
         mock_dbt_runner_instance = Mock()
         mock_dbt_runner_instance.invoke.return_value = mock_result
@@ -491,273 +815,15 @@ def test_runner_uses_original_log_level_with_force_nodes_as_tasks(
             "prefect_dbt.core.runner.serialize_context", Mock(return_value={})
         )
 
-        result = runner.invoke(["run"])
-
-        assert result == mock_result
-        mock_dbt_runner_instance.invoke.assert_called_once()
-        # Verify log_level uses the original level even with _force_nodes_as_tasks
-        call_kwargs = mock_dbt_runner_instance.invoke.call_args[1]
-        assert call_kwargs["log_level"] == runner.log_level
-        assert call_kwargs["log_level_file"] == runner.log_level
-
-    def test_runner_force_nodes_as_tasks_with_flow_context(
-        self, monkeypatch: pytest.MonkeyPatch
-    ):
-        """Test that runner behavior when _force_nodes_as_tasks is True and in flow context."""
-        runner = PrefectDbtRunner(_force_nodes_as_tasks=True)
-
-        mock_result = Mock()
-        mock_result.success = True
-        mock_result.exception = None
-
-        mock_dbt_runner_instance = Mock()
-        mock_dbt_runner_instance.invoke.return_value = mock_result
-
-        mock_dbt_runner_class = Mock(return_value=mock_dbt_runner_instance)
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.dbtRunner",
-            mock_dbt_runner_class,
+        # Mock the settings resolve_profiles_yml method to prevent file access
+        mock_settings = Mock()
+        mock_settings.resolve_profiles_yml.return_value.__enter__ = Mock(
+            return_value="/mock/profiles/dir"
         )
-
-        # Mock the callback methods
-        mock_logging_callback = Mock()
-        mock_node_started_callback = Mock()
-        mock_node_finished_callback = Mock()
-        monkeypatch.setattr(
-            runner, "_create_logging_callback", Mock(return_value=mock_logging_callback)
-        )
-        monkeypatch.setattr(
-            runner,
-            "_create_node_started_callback",
-            Mock(return_value=mock_node_started_callback),
-        )
-        monkeypatch.setattr(
-            runner,
-            "_create_node_finished_callback",
-            Mock(return_value=mock_node_finished_callback),
-        )
-
-        # Mock context to simulate flow run (non-empty dict)
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.serialize_context",
-            Mock(return_value={"flow_run_context": {"foo": "bar"}}),
-        )
-
-        result = runner.invoke(["run"])
-
-        assert result == mock_result
-        mock_dbt_runner_instance.invoke.assert_called_once()
-        # Verify callbacks are added and log_level is 'none' when in flow context
-        mock_dbt_runner_class.assert_called_once_with(
-            callbacks=[
-                mock_logging_callback,
-                mock_node_started_callback,
-                mock_node_finished_callback,
-            ]
-        )
-        call_kwargs = mock_dbt_runner_instance.invoke.call_args[1]
-        assert call_kwargs["log_level"] == "none"
-        assert call_kwargs["log_level_file"] == runner.log_level
-
-    def test_runner_force_nodes_as_tasks_with_task_context(
-        self, monkeypatch: pytest.MonkeyPatch
-    ):
-        """Test that runner behavior when _force_nodes_as_tasks is True and in task context."""
-        runner = PrefectDbtRunner(_force_nodes_as_tasks=True)
-
-        mock_result = Mock()
-        mock_result.success = True
-        mock_result.exception = None
-
-        mock_dbt_runner_instance = Mock()
-        mock_dbt_runner_instance.invoke.return_value = mock_result
-
-        mock_dbt_runner_class = Mock(return_value=mock_dbt_runner_instance)
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.dbtRunner",
-            mock_dbt_runner_class,
-        )
-
-        # Mock the callback methods
-        mock_logging_callback = Mock()
-        mock_node_started_callback = Mock()
-        mock_node_finished_callback = Mock()
-        monkeypatch.setattr(
-            runner, "_create_logging_callback", Mock(return_value=mock_logging_callback)
-        )
-        monkeypatch.setattr(
-            runner,
-            "_create_node_started_callback",
-            Mock(return_value=mock_node_started_callback),
-        )
-        monkeypatch.setattr(
-            runner,
-            "_create_node_finished_callback",
-            Mock(return_value=mock_node_finished_callback),
-        )
-
-        # Mock context to simulate task run (non-empty dict)
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.serialize_context",
-            Mock(return_value={"task_run_context": {"foo": "bar"}}),
-        )
-
-        result = runner.invoke(["run"])
-
-        assert result == mock_result
-        mock_dbt_runner_instance.invoke.assert_called_once()
-        # Verify callbacks are added and log_level is 'none' when in task context
-        mock_dbt_runner_class.assert_called_once_with(
-            callbacks=[
-                mock_logging_callback,
-                mock_node_started_callback,
-                mock_node_finished_callback,
-            ]
-        )
-        call_kwargs = mock_dbt_runner_instance.invoke.call_args[1]
-        assert call_kwargs["log_level"] == "none"
-        assert call_kwargs["log_level_file"] == runner.log_level
-
-    def test_runner_context_detection_edge_cases(self, monkeypatch: pytest.MonkeyPatch):
-        """Test edge cases for context detection."""
-        runner = PrefectDbtRunner()
-
-        mock_result = Mock()
-        mock_result.success = True
-        mock_result.exception = None
-
-        mock_dbt_runner_instance = Mock()
-        mock_dbt_runner_instance.invoke.return_value = mock_result
-
-        mock_dbt_runner_class = Mock(return_value=mock_dbt_runner_instance)
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.dbtRunner",
-            mock_dbt_runner_class,
-        )
-
-        # Mock the callback methods
-        monkeypatch.setattr(runner, "_create_logging_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
-
-        # Test with None values in context
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.serialize_context",
-            Mock(return_value={"flow_run_context": None, "task_run_context": None}),
-        )
-
-        result = runner.invoke(["run"])
-
-        assert result == mock_result
-        mock_dbt_runner_instance.invoke.assert_called_once()
-        # Verify no callbacks when context values are None
-        mock_dbt_runner_class.assert_called_once_with(callbacks=[])
-        call_kwargs = mock_dbt_runner_instance.invoke.call_args[1]
-        assert call_kwargs["log_level"] == runner.log_level
-
-    def test_runner_handles_dbt_exceptions(self, monkeypatch: pytest.MonkeyPatch):
-        """Test that runner handles dbt exceptions gracefully."""
-        runner = PrefectDbtRunner()
-
-        mock_result = Mock()
-        mock_result.success = False
-        mock_result.exception = "Test exception"
-
-        mock_dbt_runner_instance = Mock()
-        mock_dbt_runner_instance.invoke.return_value = mock_result
-
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.dbtRunner",
-            Mock(return_value=mock_dbt_runner_instance),
-        )
-
-        # Mock the callback methods
-        monkeypatch.setattr(runner, "_create_logging_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
-
-        # Mock context to simulate no flow/task run
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.serialize_context", Mock(return_value={})
-        )
-
-        with pytest.raises(ValueError, match="Failed to invoke dbt command"):
-            runner.invoke(["run"])
-
-    def test_runner_handles_dbt_failures_with_raise_on_failure_true(
-        self, monkeypatch: pytest.MonkeyPatch
-    ):
-        """Test that runner raises on dbt failures when raise_on_failure is True."""
-        runner = PrefectDbtRunner(raise_on_failure=True)
-
-        mock_result = Mock()
-        mock_result.success = False
-        mock_result.exception = None
-        mock_result.result = Mock(spec=RunExecutionResult)
-        mock_result.result.results = [
-            Mock(
-                status=RunStatus.Error,
-                node=Mock(resource_type="model", name="test_model"),
-                message="Test error",
-            )
-        ]
-
-        mock_dbt_runner_instance = Mock()
-        mock_dbt_runner_instance.invoke.return_value = mock_result
-
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.dbtRunner",
-            Mock(return_value=mock_dbt_runner_instance),
-        )
-
-        # Mock the callback methods
-        monkeypatch.setattr(runner, "_create_logging_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
-
-        # Mock context to simulate no flow/task run
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.serialize_context", Mock(return_value={})
-        )
-
-        with pytest.raises(ValueError, match="Failures detected during invocation"):
-            runner.invoke(["run"])
-
-    def test_runner_handles_dbt_failures_with_raise_on_failure_false(
-        self, monkeypatch: pytest.MonkeyPatch
-    ):
-        """Test that runner doesn't raise on dbt failures when raise_on_failure is False."""
-        runner = PrefectDbtRunner(raise_on_failure=False)
-
-        mock_result = Mock()
-        mock_result.success = False
-        mock_result.exception = None
-        mock_result.result = Mock(spec=RunExecutionResult)
-        mock_result.result.results = [
-            Mock(
-                status=RunStatus.Error,
-                node=Mock(resource_type="model", name="test_model"),
-                message="Test error",
-            )
-        ]
-
-        mock_dbt_runner_instance = Mock()
-        mock_dbt_runner_instance.invoke.return_value = mock_result
-
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.dbtRunner",
-            Mock(return_value=mock_dbt_runner_instance),
-        )
-
-        # Mock the callback methods
-        monkeypatch.setattr(runner, "_create_logging_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
-
-        # Mock context to simulate no flow/task run
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.serialize_context", Mock(return_value={})
+        mock_settings.resolve_profiles_yml.return_value.__exit__ = Mock(
+            return_value=None
         )
+        runner.settings = mock_settings
 
         result = runner.invoke(["run"])
         assert result == mock_result
@@ -786,51 +852,30 @@ def test_runner_creates_callbacks_successfully(self):
         assert callable(node_finished_callback)
 
     def test_runner_handles_kwargs_and_cli_flags_together(
-        self, monkeypatch: pytest.MonkeyPatch
+        self,
+        mock_runner_with_callback_mocks,
+        mock_dbt_runner_class_with_success,
+        mock_empty_context,
     ):
         """Test that runner handles both kwargs and CLI flags together."""
-        runner = PrefectDbtRunner()
-
-        mock_result = Mock()
-        mock_result.success = True
-        mock_result.exception = None
-
-        mock_dbt_runner_instance = Mock()
-        mock_dbt_runner_instance.invoke.return_value = mock_result
-
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.dbtRunner",
-            Mock(return_value=mock_dbt_runner_instance),
-        )
-
-        # Mock the callback methods
-        monkeypatch.setattr(runner, "_create_logging_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
-        monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
-
-        # Mock context to simulate no flow/task run
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.serialize_context", Mock(return_value={})
-        )
+        runner, _, _, _ = mock_runner_with_callback_mocks
 
         # Test with both kwargs and CLI flags
         result = runner.invoke(
             ["--target-path", "/cli/path", "run"], target_path="/kwargs/path"
         )
 
-        assert result == mock_result
+        assert result.success is True
+        assert result.exception is None
         # CLI flag should take precedence
         assert runner._target_path == Path("/cli/path")
 
     def test_get_upstream_manifest_nodes_and_configs_returns_correct_structure(
-        self, monkeypatch: pytest.MonkeyPatch
+        self, mock_manifest
     ):
         """Test that _get_upstream_manifest_nodes_and_configs returns correct structure."""
         runner = PrefectDbtRunner()
 
-        # Mock manifest and nodes
-        mock_manifest = Mock(spec=Manifest)
-
         # Create properly configured mock nodes
         mock_node_1 = Mock(spec=ManifestNode)
         mock_node_1.relation_name = "test_relation_1"
@@ -866,13 +911,12 @@ def test_get_upstream_manifest_nodes_and_configs_returns_correct_structure(
             assert isinstance(item[1], dict)  # config dict
 
     def test_get_upstream_manifest_nodes_and_configs_handles_missing_nodes(
-        self, monkeypatch: pytest.MonkeyPatch
+        self, mock_manifest
     ):
         """Test that _get_upstream_manifest_nodes_and_configs handles missing nodes gracefully."""
         runner = PrefectDbtRunner()
 
         # Mock manifest with missing nodes
-        mock_manifest = Mock(spec=Manifest)
         mock_manifest.nodes = {}  # Empty nodes dict
         runner._manifest = mock_manifest
 
@@ -886,13 +930,12 @@ def test_get_upstream_manifest_nodes_and_configs_handles_missing_nodes(
         assert result == []
 
     def test_get_upstream_manifest_nodes_and_configs_handles_missing_relation_name(
-        self, monkeypatch: pytest.MonkeyPatch
+        self, mock_manifest
     ):
         """Test that _get_upstream_manifest_nodes_and_configs handles missing relation_name."""
         runner = PrefectDbtRunner()
 
         # Mock manifest and nodes
-        mock_manifest = Mock(spec=Manifest)
         mock_node_without_relation = Mock(spec=ManifestNode)
         mock_node_without_relation.relation_name = None  # Missing relation_name
         mock_node_without_relation.config = Mock()
@@ -908,28 +951,13 @@ def test_get_upstream_manifest_nodes_and_configs_handles_missing_relation_name(
             runner._get_upstream_manifest_nodes_and_configs(mock_node)
 
     def test_call_task_with_enable_assets_true_creates_materializing_task(
-        self, monkeypatch: pytest.MonkeyPatch
+        self, mock_manifest, mock_manifest_node, monkeypatch
     ):
         """Test that _call_task creates MaterializingTask when enable_assets is True."""
         runner = PrefectDbtRunner()
 
-        # Mock manifest with adapter type
-        mock_manifest = Mock(spec=Manifest)
-        mock_manifest.metadata = Mock()
-        mock_manifest.metadata.adapter_type = "snowflake"
-        mock_manifest.nodes = {}
         runner._manifest = mock_manifest
 
-        # Mock manifest node
-        mock_node = Mock(spec=ManifestNode)
-        mock_node.resource_type = NodeType.Model
-        mock_node.unique_id = "test_node"
-        mock_node.relation_name = "test_relation"
-        mock_node.name = "test_model"
-        mock_node.description = "Test model"
-        mock_node.config = Mock()
-        mock_node.config.meta = {}
-
         # Mock task state
         mock_task_state = Mock()
 
@@ -964,7 +992,7 @@ def test_call_task_with_enable_assets_true_creates_materializing_task(
         context = {"test": "context"}
         enable_assets = True
 
-        runner._call_task(mock_task_state, mock_node, context, enable_assets)
+        runner._call_task(mock_task_state, mock_manifest_node, context, enable_assets)
 
         # Verify MaterializingTask was created and started
         mock_task_state.start_task.assert_called_once_with(
@@ -974,28 +1002,13 @@ def test_call_task_with_enable_assets_true_creates_materializing_task(
         mock_task_state.run_task_in_thread.assert_called_once()
 
     def test_call_task_with_enable_assets_false_creates_regular_task(
-        self, monkeypatch: pytest.MonkeyPatch
+        self, mock_manifest, mock_manifest_node, monkeypatch
     ):
         """Test that _call_task creates regular Task when enable_assets is False."""
         runner = PrefectDbtRunner()
 
-        # Mock manifest with adapter type
-        mock_manifest = Mock(spec=Manifest)
-        mock_manifest.metadata = Mock()
-        mock_manifest.metadata.adapter_type = "snowflake"
-        mock_manifest.nodes = {}
         runner._manifest = mock_manifest
 
-        # Mock manifest node
-        mock_node = Mock(spec=ManifestNode)
-        mock_node.resource_type = NodeType.Model
-        mock_node.unique_id = "test_node"
-        mock_node.relation_name = "test_relation"
-        mock_node.name = "test_model"
-        mock_node.description = "Test model"
-        mock_node.config = Mock()
-        mock_node.config.meta = {}
-
         # Mock task state
         mock_task_state = Mock()
 
@@ -1019,7 +1032,7 @@ def test_call_task_with_enable_assets_false_creates_regular_task(
         context = {"test": "context"}
         enable_assets = False
 
-        runner._call_task(mock_task_state, mock_node, context, enable_assets)
+        runner._call_task(mock_task_state, mock_manifest_node, context, enable_assets)
 
         # Verify regular Task was created and started
         mock_task_state.start_task.assert_called_once_with("test_node", mock_task)
@@ -1027,21 +1040,15 @@ def test_call_task_with_enable_assets_false_creates_regular_task(
         mock_task_state.run_task_in_thread.assert_called_once()
 
     def test_call_task_handles_missing_adapter_type(
-        self, monkeypatch: pytest.MonkeyPatch
+        self, mock_manifest, mock_manifest_node
     ):
         """Test that _call_task handles missing adapter type gracefully."""
         runner = PrefectDbtRunner()
 
         # Mock manifest without adapter type
-        mock_manifest = Mock(spec=Manifest)
-        mock_manifest.metadata = Mock()
         mock_manifest.metadata.adapter_type = None
         runner._manifest = mock_manifest
 
-        # Mock manifest node
-        mock_node = Mock(spec=ManifestNode)
-        mock_node.resource_type = NodeType.Model
-
         # Mock task state
         mock_task_state = Mock()
 
@@ -1049,19 +1056,16 @@ def test_call_task_handles_missing_adapter_type(
         enable_assets = True
 
         with pytest.raises(ValueError, match="Adapter type not found in manifest"):
-            runner._call_task(mock_task_state, mock_node, context, enable_assets)
+            runner._call_task(
+                mock_task_state, mock_manifest_node, context, enable_assets
+            )
 
     def test_call_task_handles_missing_relation_name_for_assets(
-        self, monkeypatch: pytest.MonkeyPatch
+        self, mock_manifest, monkeypatch
     ):
         """Test that _call_task handles missing relation_name when creating assets."""
         runner = PrefectDbtRunner()
 
-        # Mock manifest with adapter type
-        mock_manifest = Mock(spec=Manifest)
-        mock_manifest.metadata = Mock()
-        mock_manifest.metadata.adapter_type = "snowflake"
-        mock_manifest.nodes = {}
         runner._manifest = mock_manifest
 
         # Mock manifest node without relation_name
@@ -1092,114 +1096,58 @@ def test_runner_handles_project_loading_successfully(
         """Test that runner can load project successfully."""
         mock_project = Mock()
 
-        # Mock RuntimeConfig.from_args
+        # Mock Project.from_project_root
+        mock_from_project_root = Mock(return_value=mock_project)
         monkeypatch.setattr(
-            "prefect_dbt.core.runner.RuntimeConfig.from_args",
-            Mock(return_value=mock_project),
+            "prefect_dbt.core.runner.Project.from_project_root",
+            mock_from_project_root,
         )
 
         runner = PrefectDbtRunner()
         runner._project_dir = Path("/test/project")
 
-        # Access project property to trigger loading
-        result = runner.project
-
-        assert result == mock_project
-        RuntimeConfig.from_args.assert_called_once_with(Path("/test/project"))
-
-    def test_runner_handles_project_loading_with_settings(
-        self, monkeypatch: pytest.MonkeyPatch
-    ):
-        """Test that runner uses settings.project_dir when _project_dir is not set."""
-        mock_project = Mock()
-
-        # Mock RuntimeConfig.from_args
-        monkeypatch.setattr(
-            "prefect_dbt.core.runner.RuntimeConfig.from_args",
-            Mock(return_value=mock_project),
-        )
-
-        # Create settings with project_dir
-        settings = PrefectDbtSettings()
-        settings.project_dir = Path("/settings/project")
-
-        runner = PrefectDbtRunner(settings=settings)
-
-        # Access project property to trigger loading
-        result = runner.project
-
-        assert result == mock_project
-        RuntimeConfig.from_args.assert_called_once_with(Path("/settings/project"))
-
-    def test_runner_get_compiled_code_path_uses_project_name(
-        self, monkeypatch: pytest.MonkeyPatch
-    ):
-        """Test that _get_compiled_code_path uses project.project_name correctly."""
-        runner = PrefectDbtRunner()
-        runner._project_dir = Path("/test/project")
-        runner._target_path = Path("target")
-
-        # Mock project with project_name
-        mock_project = Mock()
-        mock_project.project_name = "test_project"
-        runner._project = mock_project
-
-        # Mock manifest node
-        mock_node = Mock(spec=ManifestNode)
-        mock_node.original_file_path = "models/test_model.sql"
-
-        result = runner._get_compiled_code_path(mock_node)
-
-        # Should use project.project_name instead of project_dir name
-        expected_path = (
-            Path("/test/project")
-            / "target"
-            / "compiled"
-            / "test_project"
-            / "models/test_model.sql"
-        )
-        assert result == expected_path
-
-    def test_runner_get_compiled_code_path_handles_project_name_with_slash(
-        self, monkeypatch: pytest.MonkeyPatch
-    ):
-        """Test that _get_compiled_code_path handles project_name with slashes correctly."""
-        runner = PrefectDbtRunner()
-        runner._project_dir = Path("/test/project")
-        runner._target_path = Path("target")
-
-        # Mock project with project_name containing slashes
-        mock_project = Mock()
-        mock_project.project_name = "org/test_project"
-        runner._project = mock_project
+        # Access project property to trigger loading
+        result = runner.project
 
-        # Mock manifest node
-        mock_node = Mock(spec=ManifestNode)
-        mock_node.original_file_path = "models/test_model.sql"
+        assert result == mock_project
+        mock_from_project_root.assert_called_once()
 
-        result = runner._get_compiled_code_path(mock_node)
+    def test_runner_handles_project_loading_with_settings(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Test that runner uses settings.project_dir when _project_dir is not set."""
+        mock_project = Mock()
 
-        # Should extract the last part after splitting by slash
-        expected_path = (
-            Path("/test/project")
-            / "target"
-            / "compiled"
-            / "test_project"
-            / "models/test_model.sql"
+        # Mock Project.from_project_root
+        mock_from_project_root = Mock(return_value=mock_project)
+        monkeypatch.setattr(
+            "prefect_dbt.core.runner.Project.from_project_root",
+            mock_from_project_root,
         )
-        assert result == expected_path
 
-    def test_runner_get_compiled_code_path_handles_project_name_without_slash(
+        # Create settings with project_dir
+        settings = PrefectDbtSettings()
+        settings.project_dir = Path("/settings/project")
+
+        runner = PrefectDbtRunner(settings=settings)
+
+        # Access project property to trigger loading
+        result = runner.project
+
+        assert result == mock_project
+        mock_from_project_root.assert_called_once()
+
+    def test_runner_get_compiled_code_path_uses_project_name(
         self, monkeypatch: pytest.MonkeyPatch
     ):
-        """Test that _get_compiled_code_path handles project_name without slashes correctly."""
+        """Test that _get_compiled_code_path uses project.project_name correctly."""
         runner = PrefectDbtRunner()
         runner._project_dir = Path("/test/project")
         runner._target_path = Path("target")
 
-        # Mock project with simple project_name
+        # Mock project with project_name
         mock_project = Mock()
-        mock_project.project_name = "simple_project"
+        mock_project.project_name = "test_project"
         runner._project = mock_project
 
         # Mock manifest node
@@ -1208,12 +1156,12 @@ def test_runner_get_compiled_code_path_handles_project_name_without_slash(
 
         result = runner._get_compiled_code_path(mock_node)
 
-        # Should use the project_name as is
+        # Should use project.project_name instead of project_dir name
         expected_path = (
             Path("/test/project")
             / "target"
             / "compiled"
-            / "simple_project"
+            / "test_project"
             / "models/test_model.sql"
         )
         assert result == expected_path
@@ -1230,10 +1178,11 @@ def test_runner_get_compiled_code_path_triggers_project_loading(
         mock_project = Mock()
         mock_project.project_name = "test_project"
 
-        # Mock RuntimeConfig.from_args
+        # Mock Project.from_project_root
+        mock_from_project_root = Mock(return_value=mock_project)
         monkeypatch.setattr(
-            "prefect_dbt.core.runner.RuntimeConfig.from_args",
-            Mock(return_value=mock_project),
+            "prefect_dbt.core.runner.Project.from_project_root",
+            mock_from_project_root,
         )
 
         # Mock manifest node
@@ -1243,7 +1192,7 @@ def test_runner_get_compiled_code_path_triggers_project_loading(
         result = runner._get_compiled_code_path(mock_node)
 
         # Should trigger project loading
-        RuntimeConfig.from_args.assert_called_once_with(Path("/test/project"))
+        mock_from_project_root.assert_called_once()
         expected_path = (
             Path("/test/project")
             / "target"
@@ -1339,6 +1288,446 @@ def test_runner_get_compiled_code_with_file_not_found(
         # Should return empty string when file doesn't exist
         assert result == ""
 
+    def test_runner_invoke_uses_resolve_profiles_yml_context_manager(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Test that invoke method uses resolve_profiles_yml context manager with actual templated profiles.yml."""
+        runner = PrefectDbtRunner()
+
+        mock_result = Mock()
+        mock_result.success = True
+        mock_result.exception = None
+
+        mock_dbt_runner_instance = Mock()
+        mock_dbt_runner_instance.invoke.return_value = mock_result
+
+        mock_dbt_runner_class = Mock(return_value=mock_dbt_runner_instance)
+        monkeypatch.setattr(
+            "prefect_dbt.core.runner.dbtRunner",
+            mock_dbt_runner_class,
+        )
+
+        # Mock the callback methods
+        monkeypatch.setattr(runner, "_create_logging_callback", Mock())
+        monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
+        monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
+
+        # Mock context to simulate no flow/task run
+        monkeypatch.setattr(
+            "prefect_dbt.core.runner.serialize_context", Mock(return_value={})
+        )
+
+        # Create a mock profiles.yml with templated values
+        mock_profiles_yml = {
+            "my_project": {
+                "targets": {
+                    "dev": {
+                        "type": "snowflake",
+                        "account": "{{ prefect.blocks.snowflake-credentials.account }}",
+                        "user": "{{ prefect.blocks.snowflake-credentials.user }}",
+                        "password": "{{ prefect.blocks.snowflake-credentials.password }}",
+                        "database": "{{ prefect.variables.database_name }}",
+                        "warehouse": "{{ prefect.variables.warehouse_name }}",
+                        "schema": "{{ prefect.variables.schema_name }}",
+                    }
+                }
+            }
+        }
+
+        # Mock the load_profiles_yml method to return our templated profiles
+        mock_load_method = Mock(return_value=mock_profiles_yml)
+        monkeypatch.setattr(
+            "prefect_dbt.core.settings.PrefectDbtSettings.load_profiles_yml",
+            mock_load_method,
+        )
+
+        # Mock the resolve functions to return resolved values
+        mock_resolved_profiles = {
+            "my_project": {
+                "targets": {
+                    "dev": {
+                        "type": "snowflake",
+                        "account": "test-account.snowflakecomputing.com",
+                        "user": "test_user",
+                        "password": "test_password",
+                        "database": "test_database",
+                        "warehouse": "test_warehouse",
+                        "schema": "test_schema",
+                    }
+                }
+            }
+        }
+
+        # Mock the resolve_block_document_references function
+        mock_resolve_blocks = Mock(return_value=mock_resolved_profiles)
+        monkeypatch.setattr(
+            "prefect_dbt.core.settings.resolve_block_document_references",
+            mock_resolve_blocks,
+        )
+
+        # Mock the resolve_variables function
+        mock_resolve_vars = Mock(return_value=mock_resolved_profiles)
+        monkeypatch.setattr(
+            "prefect_dbt.core.settings.resolve_variables",
+            mock_resolve_vars,
+        )
+
+        # Mock run_coro_as_sync to return the resolved profiles directly
+        def mock_run_coro(coro) -> Any:
+            return mock_resolved_profiles
+
+        mock_run_coro_func = Mock(side_effect=mock_run_coro)
+        monkeypatch.setattr(
+            "prefect_dbt.core.settings.run_coro_as_sync",
+            mock_run_coro_func,
+        )
+
+        # Mock yaml.dump to verify it's called with resolved profiles
+        mock_yaml_dump = Mock()
+        monkeypatch.setattr("prefect_dbt.core.settings.yaml.dump", mock_yaml_dump)
+
+        # Mock Path.write_text to verify the resolved profiles are written
+        mock_write_text = Mock()
+        monkeypatch.setattr("pathlib.Path.write_text", mock_write_text)
+
+        # Call invoke with additional kwargs to test preservation
+        result = runner.invoke(["run"], target="dev", threads=4, custom_arg="value")
+
+        assert result == mock_result
+
+        # Verify the profiles were loaded and resolved
+        mock_load_method.assert_called_once()
+        mock_run_coro_func.assert_called()
+        mock_yaml_dump.assert_called_once_with(
+            mock_resolved_profiles, default_style=None, default_flow_style=False
+        )
+        mock_write_text.assert_called_once()
+
+        # Verify dbtRunner was called with the resolved profiles_dir and preserved kwargs
+        mock_dbt_runner_instance.invoke.assert_called_once()
+        call_kwargs = mock_dbt_runner_instance.invoke.call_args[1]
+        assert call_kwargs["target"] == "dev"
+        assert call_kwargs["threads"] == 4
+        assert call_kwargs["custom_arg"] == "value"
+
+        # Verify the profiles_dir was changed from the original (indicating resolution was used)
+        assert call_kwargs["profiles_dir"] != str(runner.profiles_dir)
+
+    def test_runner_loads_graph(self, monkeypatch: pytest.MonkeyPatch):
+        """Test that runner loads graph."""
+        runner = PrefectDbtRunner()
+
+        # Mock manifest
+        mock_manifest = Mock(spec=Manifest)
+        runner._manifest = mock_manifest
+
+        # Mock Linker
+        mock_linker = Mock()
+        monkeypatch.setattr(
+            "prefect_dbt.core.runner.Linker", Mock(return_value=mock_linker)
+        )
+
+        # Mock Graph
+        mock_graph = Mock()
+        monkeypatch.setattr(
+            "prefect_dbt.core.runner.Graph", Mock(return_value=mock_graph)
+        )
+
+        # Access graph property to trigger loading (graph is None initially)
+        result = runner.graph
+
+        # Verify graph was created correctly
+        assert result == mock_graph
+        # Verify Linker was used to link the graph
+        mock_linker.link_graph.assert_called_once_with(mock_manifest)
+
+    def test_runner_loads_graph_with_test_edges_for_build_command(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Test that runner loads graph with test edges when 'build' is in args."""
+        runner = PrefectDbtRunner()
+
+        # Mock manifest
+        mock_manifest = Mock(spec=Manifest)
+        runner._manifest = mock_manifest
+
+        # Mock _set_graph_from_manifest method
+        mock_set_graph = Mock()
+        monkeypatch.setattr(runner, "_set_graph_from_manifest", mock_set_graph)
+
+        # Mock dbtRunner and successful result
+        mock_result = Mock()
+        mock_result.success = True
+        mock_result.exception = None
+
+        mock_dbt_runner_instance = Mock()
+        mock_dbt_runner_instance.invoke.return_value = mock_result
+
+        mock_dbt_runner_class = Mock(return_value=mock_dbt_runner_instance)
+        monkeypatch.setattr("prefect_dbt.core.runner.dbtRunner", mock_dbt_runner_class)
+
+        # Mock the callback methods
+        monkeypatch.setattr(runner, "_create_logging_callback", Mock())
+        monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
+        monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
+
+        # Mock context to simulate flow run (so callbacks are created)
+        monkeypatch.setattr(
+            "prefect_dbt.core.runner.serialize_context",
+            Mock(return_value={"flow_run_context": {"foo": "bar"}}),
+        )
+
+        # Mock settings
+        mock_settings = Mock()
+        mock_settings.resolve_profiles_yml.return_value.__enter__ = Mock(
+            return_value="/mock/profiles/dir"
+        )
+        mock_settings.resolve_profiles_yml.return_value.__exit__ = Mock(
+            return_value=None
+        )
+        runner.settings = mock_settings
+
+        # Call invoke with 'build' command
+        runner.invoke(["build"])
+
+        # Verify _create_node_finished_callback was called with add_test_edges=True
+        runner._create_node_finished_callback.assert_called_once()
+        call_args = runner._create_node_finished_callback.call_args
+        assert call_args[1]["add_test_edges"] is True
+
+    def test_runner_loads_graph_with_test_edges_for_retry_build(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Test that runner loads graph with test edges when retrying 'build' command."""
+        runner = PrefectDbtRunner()
+
+        # Mock manifest
+        mock_manifest = Mock(spec=Manifest)
+        runner._manifest = mock_manifest
+
+        # Mock _set_graph_from_manifest method
+        mock_set_graph = Mock()
+        monkeypatch.setattr(runner, "_set_graph_from_manifest", mock_set_graph)
+
+        # Mock dbtRunner and successful result
+        mock_result = Mock()
+        mock_result.success = True
+        mock_result.exception = None
+
+        mock_dbt_runner_instance = Mock()
+        mock_dbt_runner_instance.invoke.return_value = mock_result
+
+        mock_dbt_runner_class = Mock(return_value=mock_dbt_runner_instance)
+        monkeypatch.setattr("prefect_dbt.core.runner.dbtRunner", mock_dbt_runner_class)
+
+        # Mock the callback methods
+        monkeypatch.setattr(runner, "_create_logging_callback", Mock())
+        monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
+        monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
+
+        # Mock context to simulate flow run (so callbacks are created)
+        monkeypatch.setattr(
+            "prefect_dbt.core.runner.serialize_context",
+            Mock(return_value={"flow_run_context": {"foo": "bar"}}),
+        )
+
+        # Mock settings
+        mock_settings = Mock()
+        mock_settings.resolve_profiles_yml.return_value.__enter__ = Mock(
+            return_value="/mock/profiles/dir"
+        )
+        mock_settings.resolve_profiles_yml.return_value.__exit__ = Mock(
+            return_value=None
+        )
+        runner.settings = mock_settings
+
+        # Mock load_result_state to return previous build results
+        mock_previous_results = Mock()
+        mock_previous_results.args = {"which": "build"}
+        monkeypatch.setattr(
+            "prefect_dbt.core.runner.load_result_state",
+            Mock(return_value=mock_previous_results),
+        )
+
+        # Mock project_dir and target_path
+        runner._project_dir = Path("/test/project")
+        runner._target_path = Path("target")
+
+        # Call invoke with 'retry' command
+        runner.invoke(["retry"])
+
+        # Verify _create_node_finished_callback was called with add_test_edges=True
+        runner._create_node_finished_callback.assert_called_once()
+        call_args = runner._create_node_finished_callback.call_args
+        assert call_args[1]["add_test_edges"] is True
+
+    def test_runner_does_not_load_test_edges_for_non_build_commands(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Test that runner does not load test edges for non-build commands."""
+        runner = PrefectDbtRunner()
+
+        # Mock manifest
+        mock_manifest = Mock(spec=Manifest)
+        runner._manifest = mock_manifest
+
+        # Mock _set_graph_from_manifest method
+        mock_set_graph = Mock()
+        monkeypatch.setattr(runner, "_set_graph_from_manifest", mock_set_graph)
+
+        # Mock dbtRunner and successful result
+        mock_result = Mock()
+        mock_result.success = True
+        mock_result.exception = None
+
+        mock_dbt_runner_instance = Mock()
+        mock_dbt_runner_instance.invoke.return_value = mock_result
+
+        mock_dbt_runner_class = Mock(return_value=mock_dbt_runner_instance)
+        monkeypatch.setattr("prefect_dbt.core.runner.dbtRunner", mock_dbt_runner_class)
+
+        # Mock the callback methods
+        monkeypatch.setattr(runner, "_create_logging_callback", Mock())
+        monkeypatch.setattr(runner, "_create_node_started_callback", Mock())
+        monkeypatch.setattr(runner, "_create_node_finished_callback", Mock())
+
+        # Mock context to simulate flow run (so callbacks are created)
+        monkeypatch.setattr(
+            "prefect_dbt.core.runner.serialize_context",
+            Mock(return_value={"flow_run_context": {"foo": "bar"}}),
+        )
+
+        # Mock settings
+        mock_settings = Mock()
+        mock_settings.resolve_profiles_yml.return_value.__enter__ = Mock(
+            return_value="/mock/profiles/dir"
+        )
+        mock_settings.resolve_profiles_yml.return_value.__exit__ = Mock(
+            return_value=None
+        )
+        runner.settings = mock_settings
+
+        # Call invoke with 'run' command (not build)
+        runner.invoke(["run"])
+
+        # Verify _create_node_finished_callback was called with add_test_edges=False
+        runner._create_node_finished_callback.assert_called_once()
+        call_args = runner._create_node_finished_callback.call_args
+        assert call_args[1]["add_test_edges"] is False
+
+    def test_runner_handles_retry_without_previous_results(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Test that runner handles retry when no previous results exist."""
+        runner = PrefectDbtRunner()
+
+        # Mock project_dir and target_path
+        runner._project_dir = Path("/test/project")
+        runner._target_path = Path("target")
+
+        # Mock load_result_state to return None (no previous results)
+        monkeypatch.setattr(
+            "prefect_dbt.core.runner.load_result_state", Mock(return_value=None)
+        )
+
+        # Call invoke with 'retry' command
+        with pytest.raises(ValueError, match="Cannot retry. No previous results found"):
+            runner.invoke(["retry"])
+
+    def test_runner_node_finished_callback_rebuilds_graph_with_test_edges(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Test that the node finished callback rebuilds graph with test edges when needed."""
+        runner = PrefectDbtRunner()
+
+        # Mock _set_graph_from_manifest method
+        mock_set_graph = Mock()
+        monkeypatch.setattr(runner, "_set_graph_from_manifest", mock_set_graph)
+
+        # Mock graph with get_dependent_nodes method
+        mock_graph = Mock()
+        mock_graph.get_dependent_nodes.return_value = ["dep1", "dep2"]
+        runner._graph = mock_graph
+
+        # Create task state
+        task_state = Mock()
+
+        # Create callback with add_test_edges=True
+        callback = runner._create_node_finished_callback(
+            task_state, add_test_edges=True
+        )
+
+        # Create mock event for failed node
+        mock_event = Mock()
+        mock_event.info.name = "NodeFinished"
+        mock_event.data.node_info.unique_id = "test_node"
+
+        # Mock MessageToDict to return event data with failed status
+        mock_event_data = {"node_info": {"node_status": "error"}}
+        monkeypatch.setattr(
+            "prefect_dbt.core.runner.MessageToDict", Mock(return_value=mock_event_data)
+        )
+
+        # Mock get_dbt_event_msg
+        monkeypatch.setattr(
+            runner, "get_dbt_event_msg", Mock(return_value="Node failed")
+        )
+
+        # Mock _get_manifest_node_and_config to return a node
+        mock_node = Mock()
+        monkeypatch.setattr(
+            runner, "_get_manifest_node_and_config", Mock(return_value=(mock_node, {}))
+        )
+
+        # Call the callback
+        callback(mock_event)
+
+        # Verify graph was rebuilt with test edges
+        mock_set_graph.assert_called_once_with(add_test_edges=True)
+
+        # Verify dependent nodes were added to skipped nodes
+        assert "dep1" in runner._skipped_nodes
+        assert "dep2" in runner._skipped_nodes
+
+    def test_runner_does_not_create_tasks_for_skipped_nodes(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        """Test that runner does not create tasks for nodes that are marked as skipped."""
+        runner = PrefectDbtRunner()
+
+        # Add a node to the skipped nodes set
+        skipped_node_id = "skipped_node"
+        runner._skipped_nodes.add(skipped_node_id)
+
+        # Create task state
+        task_state = Mock()
+
+        # Create the node started callback
+        callback = runner._create_node_started_callback(task_state, {})
+
+        # Create mock event for the skipped node
+        mock_event = Mock()
+        mock_event.info.name = "NodeStart"
+        mock_event.data.node_info.unique_id = skipped_node_id
+
+        # Mock _get_manifest_node_and_config to return a node (should not be called)
+        mock_node = Mock()
+        mock_get_config = Mock(return_value=(mock_node, {}))
+        monkeypatch.setattr(runner, "_get_manifest_node_and_config", mock_get_config)
+
+        # Mock _call_task (should not be called)
+        mock_call_task = Mock()
+        monkeypatch.setattr(runner, "_call_task", mock_call_task)
+
+        # Call the callback
+        callback(mock_event)
+
+        # Verify that _get_manifest_node_and_config was not called
+        mock_get_config.assert_not_called()
+
+        # Verify that _call_task was not called
+        mock_call_task.assert_not_called()
+
 
 class TestExecuteDbtNode:
     """Test cases focusing on execute_dbt_node behavior."""
@@ -1364,7 +1753,6 @@ def test_execute_dbt_node_completes_successfully(
         execute_dbt_node(task_state, node_id, asset_id)
 
         # Verify expected interactions
-        task_state.set_task_logger.assert_called_once_with(node_id, mock_logger)
         task_state.wait_for_node_completion.assert_called_once_with(node_id)
         task_state.get_node_status.assert_called_once_with(node_id)
 
@@ -1402,7 +1790,6 @@ def test_execute_dbt_node_handles_no_status(self, monkeypatch: pytest.MonkeyPatc
         execute_dbt_node(task_state, node_id, asset_id)
 
         # Verify function completes without error
-        task_state.set_task_logger.assert_called_once()
         task_state.wait_for_node_completion.assert_called_once_with(node_id)
         task_state.get_node_status.assert_called_once_with(node_id)
 
diff --git a/src/integrations/prefect-dbt/tests/core/test_settings.py b/src/integrations/prefect-dbt/tests/core/test_settings.py
index 0e33c03c1c12..97a3dc9a9ffe 100644
--- a/src/integrations/prefect-dbt/tests/core/test_settings.py
+++ b/src/integrations/prefect-dbt/tests/core/test_settings.py
@@ -294,3 +294,207 @@ def test_settings_handle_complex_profiles_structure():
             == "dev-db.example.com"
         )
         assert loaded_profiles["test_profile"]["targets"]["test"]["type"] == "sqlite"
+
+
+@pytest.fixture
+def mock_prefect_blocks(monkeypatch: pytest.MonkeyPatch):
+    """Fixture to mock Prefect blocks for testing."""
+
+    # Mock block document references resolution
+    def mock_resolve_blocks(data, value_transformer=None):
+        # Simulate resolving block references
+        if isinstance(data, dict):
+            resolved = {}
+            for key, value in data.items():
+                if isinstance(value, str) and "{{ prefect.blocks.secret." in value:
+                    # Extract block name and return mock value
+                    block_name = value.split("prefect.blocks.secret.")[1].split(" }}")[
+                        0
+                    ]
+                    mock_values = {
+                        "db-host": "localhost",
+                        "db-password": "secret_password_123",
+                        "db-user": "dbt_user",
+                    }
+                    resolved[key] = mock_values.get(
+                        block_name, f"resolved_{block_name}"
+                    )
+                elif isinstance(value, (dict, list)):
+                    resolved[key] = mock_resolve_blocks(value, value_transformer)
+                else:
+                    resolved[key] = value
+            return resolved
+        elif isinstance(data, list):
+            return [mock_resolve_blocks(item, value_transformer) for item in data]
+        else:
+            return data
+
+    # Mock the async function to return resolved data directly
+    async def mock_async_resolve_blocks(data, value_transformer=None):
+        return mock_resolve_blocks(data, value_transformer)
+
+    monkeypatch.setattr(
+        "prefect_dbt.core.settings.resolve_block_document_references",
+        Mock(side_effect=mock_async_resolve_blocks),
+    )
+
+    # Mock run_coro_as_sync to return the resolved data
+    resolved_data = {
+        "my_profile": {
+            "targets": {
+                "dev": {
+                    "type": "postgres",
+                    "host": "localhost",
+                    "password": "secret_password_123",
+                    "user": "dbt_user",
+                }
+            }
+        }
+    }
+    monkeypatch.setattr(
+        "prefect_dbt.core.settings.run_coro_as_sync",
+        Mock(side_effect=lambda coro: resolved_data),
+    )
+
+
+@pytest.fixture
+def mock_prefect_variables(monkeypatch: pytest.MonkeyPatch):
+    """Fixture to mock Prefect variables for testing."""
+
+    # Mock variable resolution
+    def mock_resolve_vars(data):
+        # Simulate resolving variable references
+        if isinstance(data, dict):
+            resolved = {}
+            for key, value in data.items():
+                if isinstance(value, str) and "{{ prefect.variables." in value:
+                    # Extract variable name and return mock value
+                    var_name = value.split("prefect.variables.")[1].split(" }}")[0]
+                    mock_values = {
+                        "DB_HOST": "prod-db.example.com",
+                        "DB_PASSWORD": "prod_password_456",
+                        "DB_USER": "prod_user",
+                    }
+                    resolved[key] = mock_values.get(var_name, f"resolved_{var_name}")
+                elif isinstance(value, (dict, list)):
+                    resolved[key] = mock_resolve_vars(value)
+                else:
+                    resolved[key] = value
+            return resolved
+        elif isinstance(data, list):
+            return [mock_resolve_vars(item) for item in data]
+        else:
+            return data
+
+    # Mock the async function to return resolved data directly
+    async def mock_async_resolve_vars(data):
+        return mock_resolve_vars(data)
+
+    monkeypatch.setattr(
+        "prefect_dbt.core.settings.resolve_variables",
+        Mock(side_effect=mock_async_resolve_vars),
+    )
+
+    # Mock run_coro_as_sync to return the resolved data
+    resolved_data = {
+        "my_profile": {
+            "targets": {
+                "dev": {
+                    "type": "postgres",
+                    "host": "prod-db.example.com",
+                    "password": "prod_password_456",
+                    "user": "prod_user",
+                }
+            }
+        }
+    }
+    monkeypatch.setattr(
+        "prefect_dbt.core.settings.run_coro_as_sync",
+        Mock(side_effect=lambda coro: resolved_data),
+    )
+
+
+def test_settings_resolve_profiles_yml_with_block_references(mock_prefect_blocks):
+    """Test that resolve_profiles_yml resolves Prefect block references correctly."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        temp_path = Path(temp_dir)
+
+        # Create profiles.yml with block references
+        profiles_content = {
+            "my_profile": {
+                "targets": {
+                    "dev": {
+                        "type": "postgres",
+                        "host": "{{ prefect.blocks.secret.db-host }}",
+                        "password": "{{ prefect.blocks.secret.db-password }}",
+                        "user": "{{ prefect.blocks.secret.db-user }}",
+                    }
+                }
+            }
+        }
+
+        profiles_file = temp_path / "profiles.yml"
+        with open(profiles_file, "w") as f:
+            yaml.dump(profiles_content, f)
+
+        settings = PrefectDbtSettings(profiles_dir=temp_path)
+
+        with settings.resolve_profiles_yml() as temp_dir_path:
+            temp_profiles_path = Path(temp_dir_path) / "profiles.yml"
+            with open(temp_profiles_path, "r") as f:
+                content = yaml.safe_load(f)
+
+            # Verify block references were resolved
+            assert content["my_profile"]["targets"]["dev"]["host"] == "localhost"
+            assert (
+                content["my_profile"]["targets"]["dev"]["password"]
+                == "secret_password_123"
+            )
+            assert content["my_profile"]["targets"]["dev"]["user"] == "dbt_user"
+
+        # Verify cleanup happens
+        assert not Path(temp_dir_path).exists()
+
+
+def test_settings_resolve_profiles_yml_with_variable_references(mock_prefect_variables):
+    """Test that resolve_profiles_yml resolves Prefect variable references correctly."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        temp_path = Path(temp_dir)
+
+        # Create profiles.yml with variable references
+        profiles_content = {
+            "my_profile": {
+                "targets": {
+                    "dev": {
+                        "type": "postgres",
+                        "host": "{{ prefect.variables.DB_HOST }}",
+                        "password": "{{ prefect.variables.DB_PASSWORD }}",
+                        "user": "{{ prefect.variables.DB_USER }}",
+                    }
+                }
+            }
+        }
+
+        profiles_file = temp_path / "profiles.yml"
+        with open(profiles_file, "w") as f:
+            yaml.dump(profiles_content, f)
+
+        settings = PrefectDbtSettings(profiles_dir=temp_path)
+
+        with settings.resolve_profiles_yml() as temp_dir_path:
+            temp_profiles_path = Path(temp_dir_path) / "profiles.yml"
+            with open(temp_profiles_path, "r") as f:
+                content = yaml.safe_load(f)
+
+            # Verify variable references were resolved
+            assert (
+                content["my_profile"]["targets"]["dev"]["host"] == "prod-db.example.com"
+            )
+            assert (
+                content["my_profile"]["targets"]["dev"]["password"]
+                == "prod_password_456"
+            )
+            assert content["my_profile"]["targets"]["dev"]["user"] == "prod_user"
+
+        # Verify cleanup happens
+        assert not Path(temp_dir_path).exists()
diff --git a/src/integrations/prefect-dbt/tests/core/test_tracker.py b/src/integrations/prefect-dbt/tests/core/test_tracker.py
index df6ae0350d8e..500fb52f90af 100644
--- a/src/integrations/prefect-dbt/tests/core/test_tracker.py
+++ b/src/integrations/prefect-dbt/tests/core/test_tracker.py
@@ -9,35 +9,96 @@
 import pytest
 from prefect_dbt.core._tracker import NodeTaskTracker
 
+from prefect._internal.uuid7 import uuid7
+
+
+@pytest.fixture
+def tracker():
+    """Fixture providing a NodeTaskTracker instance."""
+    return NodeTaskTracker()
+
+
+@pytest.fixture
+def mock_task():
+    """Fixture providing a mock task."""
+    return Mock()
+
+
+@pytest.fixture
+def mock_state():
+    """Fixture providing a mock state."""
+    return Mock()
+
+
+@pytest.fixture
+def mock_context_manager(monkeypatch):
+    """Fixture providing a mock context manager for hydrated_context."""
+    mock_context = Mock()
+    mock_context.__enter__ = Mock()
+    mock_context.__exit__ = Mock()
+    monkeypatch.setattr(
+        "prefect_dbt.core._tracker.hydrated_context",
+        Mock(return_value=mock_context),
+    )
+    return mock_context
+
+
+@pytest.fixture
+def mock_run_task_sync(monkeypatch, mock_state):
+    """Fixture providing a mock run_task_sync function."""
+    monkeypatch.setattr(
+        "prefect_dbt.core._tracker.run_task_sync", Mock(return_value=mock_state)
+    )
+    return mock_state
 
-def test_tracker_manages_node_lifecycle():
-    """Test that tracker properly manages the complete lifecycle of a node."""
-    tracker = NodeTaskTracker()
-    node_id = "test_node"
-    mock_task = Mock()
 
+@pytest.fixture
+def sample_node_id():
+    """Fixture providing a sample node ID."""
+    return "test_node"
+
+
+@pytest.fixture
+def sample_task_run_id():
+    """Fixture providing a sample task run ID."""
+    return uuid7()
+
+
+@pytest.fixture
+def sample_parameters():
+    """Fixture providing sample parameters."""
+    return {"param": "value"}
+
+
+@pytest.fixture
+def sample_context():
+    """Fixture providing sample context."""
+    return {"context": "data"}
+
+
+def test_tracker_manages_node_lifecycle(tracker, sample_node_id, mock_task):
+    """Test that tracker properly manages the complete lifecycle of a node."""
     # Start the node
-    tracker.start_task(node_id, mock_task)
-    assert not tracker.is_node_complete(node_id)
+    tracker.start_task(sample_node_id, mock_task)
+    assert not tracker.is_node_complete(sample_node_id)
 
     # Set node status (completes the node)
     event_data = {"status": "success", "node_info": {"node_status": "success"}}
-    tracker.set_node_status(node_id, event_data, "Node completed successfully")
+    tracker.set_node_status(sample_node_id, event_data, "Node completed successfully")
 
     # Verify node is complete
-    assert tracker.is_node_complete(node_id)
-    status = tracker.get_node_status(node_id)
+    assert tracker.is_node_complete(sample_node_id)
+    status = tracker.get_node_status(sample_node_id)
+    assert status is not None
     assert status["event_data"] == event_data
     assert status["event_message"] == "Node completed successfully"
 
 
-def test_tracker_handles_multiple_nodes_independently():
+def test_tracker_handles_multiple_nodes_independently(tracker, mock_task):
     """Test that tracker can manage multiple nodes without interference."""
-    tracker = NodeTaskTracker()
-
     # Set up two nodes
-    tracker.start_task("node1", Mock())
-    tracker.start_task("node2", Mock())
+    tracker.start_task("node1", mock_task)
+    tracker.start_task("node2", mock_task)
 
     # Complete one node
     tracker.set_node_status("node1", {"status": "success"}, "Node 1 done")
@@ -115,54 +176,29 @@ def test_tracker_manages_dependencies():
     assert tracker.get_node_dependencies("unknown_node") == []
 
 
-def test_tracker_manages_loggers():
-    """Test that tracker properly manages task loggers."""
-    tracker = NodeTaskTracker()
-    node_id = "test_node"
-    mock_logger = Mock()
-
-    # Set and retrieve logger
-    tracker.set_task_logger(node_id, mock_logger)
-    retrieved_logger = tracker.get_task_logger(node_id)
-    assert retrieved_logger == mock_logger
-
-    # Test None for unknown node
-    assert tracker.get_task_logger("unknown_node") is None
-
-
-def test_tracker_thread_execution_outcomes(monkeypatch: pytest.MonkeyPatch):
+def test_tracker_thread_execution_outcomes(
+    tracker,
+    sample_node_id,
+    mock_task,
+    sample_task_run_id,
+    sample_parameters,
+    sample_context,
+    mock_run_task_sync,
+    mock_context_manager,
+):
     """Test that tracker properly handles thread execution outcomes."""
-    tracker = NodeTaskTracker()
-    node_id = "test_node"
-    mock_task = Mock()
-    parameters = {"param": "value"}
-    context = {"context": "data"}
+    # Mock successful execution is already set up by mock_run_task_sync fixture
 
-    # Mock successful execution
-    mock_state = Mock()
-
-    monkeypatch.setattr(
-        "prefect_dbt.core._tracker.run_task_sync", Mock(return_value=mock_state)
+    tracker.run_task_in_thread(
+        sample_node_id, mock_task, sample_task_run_id, sample_parameters, sample_context
     )
-    monkeypatch.setattr("prefect_dbt.core._tracker.hydrated_context", Mock())
-
-    # Set up context manager mock
-    mock_context = Mock()
-    mock_context.__enter__ = Mock()
-    mock_context.__exit__ = Mock()
-    monkeypatch.setattr(
-        "prefect_dbt.core._tracker.hydrated_context",
-        Mock(return_value=mock_context),
-    )
-
-    tracker.run_task_in_thread(node_id, mock_task, parameters, context)
 
     # Wait for thread completion
     time.sleep(0.2)
 
     # Verify result was stored
-    result = tracker.get_task_result(node_id)
-    assert result == mock_state
+    result = tracker.get_task_result(sample_node_id)
+    assert result == mock_run_task_sync
 
 
 def test_tracker_handles_execution_with_dependencies(monkeypatch: pytest.MonkeyPatch):
@@ -170,6 +206,7 @@ def test_tracker_handles_execution_with_dependencies(monkeypatch: pytest.MonkeyP
     tracker = NodeTaskTracker()
     node_id = "test_node"
     mock_task = Mock()
+    task_run_id = uuid7()
 
     # Set up dependencies with results
     tracker.set_node_dependencies(node_id, ["dep1", "dep2"])
@@ -197,7 +234,7 @@ def test_tracker_handles_execution_with_dependencies(monkeypatch: pytest.MonkeyP
         Mock(return_value=mock_context),
     )
 
-    tracker.run_task_in_thread(node_id, mock_task, parameters, context)
+    tracker.run_task_in_thread(node_id, mock_task, task_run_id, parameters, context)
 
     # Wait for thread completion
     time.sleep(0.2)
@@ -212,6 +249,7 @@ def test_tracker_handles_execution_failures(monkeypatch: pytest.MonkeyPatch):
     tracker = NodeTaskTracker()
     node_id = "test_node"
     mock_task = Mock()
+    task_run_id = uuid7()
     parameters = {"param": "value"}
     context = {"context": "data"}
 
@@ -234,7 +272,7 @@ def test_tracker_handles_execution_failures(monkeypatch: pytest.MonkeyPatch):
     # The exception should cause the thread to fail, but we can't easily test that
     # since the exception would be raised in the thread context
     # Instead, we test that the function can be called without error
-    tracker.run_task_in_thread(node_id, mock_task, parameters, context)
+    tracker.run_task_in_thread(node_id, mock_task, task_run_id, parameters, context)
 
     # Wait for thread completion
     time.sleep(0.2)
@@ -251,6 +289,7 @@ def test_tracker_handles_no_state_returned(monkeypatch: pytest.MonkeyPatch):
     tracker = NodeTaskTracker()
     node_id = "test_node"
     mock_task = Mock()
+    task_run_id = uuid7()
     parameters = {"param": "value"}
     context = {"context": "data"}
 
@@ -269,7 +308,7 @@ def test_tracker_handles_no_state_returned(monkeypatch: pytest.MonkeyPatch):
         Mock(return_value=mock_context),
     )
 
-    tracker.run_task_in_thread(node_id, mock_task, parameters, context)
+    tracker.run_task_in_thread(node_id, mock_task, task_run_id, parameters, context)
 
     # Wait for thread completion
     time.sleep(0.2)
@@ -279,6 +318,91 @@ def test_tracker_handles_no_state_returned(monkeypatch: pytest.MonkeyPatch):
     assert result is None
 
 
+def test_get_task_logger_creates_logger_with_task_id():
+    """Test that get_task_logger creates a logger with the correct task ID."""
+    tracker = NodeTaskTracker()
+    node_id = "test_node"
+    task_run_id = uuid7()
+
+    # Set up task run ID
+    tracker.set_task_run_id(node_id, task_run_id)
+    tracker.set_task_run_name(node_id, "test_task_run")
+
+    # Create logger
+    logger = tracker.get_task_logger(node_id)
+
+    # Verify logger has correct task run ID
+    assert logger.extra["task_run_id"] == task_run_id
+    assert logger.extra["task_run_name"] == "test_task_run"
+    assert logger.extra["task_name"] == "execute_dbt_node"
+
+
+def test_get_task_logger_with_flow_context():
+    """Test that get_task_logger includes flow context when provided."""
+    tracker = NodeTaskTracker()
+    node_id = "test_node"
+    task_run_id = uuid7()
+
+    tracker.set_task_run_id(node_id, task_run_id)
+    tracker.set_task_run_name(node_id, "test_task_run")
+
+    flow_run = {"id": "flow-run-123", "name": "test_flow_run"}
+    flow = Mock()
+    flow.name = "test_flow"
+
+    logger = tracker.get_task_logger(node_id, flow_run=flow_run, flow=flow)
+
+    assert logger.extra["flow_run_id"] == "flow-run-123"
+    assert logger.extra["flow_run_name"] == "test_flow_run"
+    assert logger.extra["flow_name"] == "test_flow"
+
+
+def test_task_id_tracking():
+    """Test that task run IDs and names are properly tracked."""
+    tracker = NodeTaskTracker()
+    node_id = "test_node"
+    task_run_id = uuid7()
+    task_run_name = "my_task_run"
+
+    # Initially no task run ID should be set
+    assert tracker.get_task_run_id(node_id) is None
+    assert tracker.get_task_run_name(node_id) is None
+
+    # Set task run ID and name
+    tracker.set_task_run_id(node_id, task_run_id)
+    tracker.set_task_run_name(node_id, task_run_name)
+
+    # Verify they are stored correctly
+    assert tracker.get_task_run_id(node_id) == task_run_id
+    assert tracker.get_task_run_name(node_id) == task_run_name
+
+
+def test_task_run_name_tracking():
+    """Test that task run names are properly tracked and updated."""
+    tracker = NodeTaskTracker()
+    node_id = "test_node"
+
+    # Initially no task run name should be set
+    assert tracker.get_task_run_name(node_id) is None
+
+    # Set initial task run name
+    initial_name = "initial_task_run"
+    tracker.set_task_run_name(node_id, initial_name)
+    assert tracker.get_task_run_name(node_id) == initial_name
+
+    # Update task run name
+    updated_name = "updated_task_run"
+    tracker.set_task_run_name(node_id, updated_name)
+    assert tracker.get_task_run_name(node_id) == updated_name
+
+    # Test with different node
+    other_node = "other_node"
+    other_name = "other_task_run"
+    tracker.set_task_run_name(other_node, other_name)
+    assert tracker.get_task_run_name(other_node) == other_name
+    assert tracker.get_task_run_name(node_id) == updated_name  # Original unchanged
+
+
 def test_tracker_comprehensive_workflow() -> None:
     """Test a comprehensive workflow using the tracker."""
     tracker = NodeTaskTracker()
@@ -290,7 +414,6 @@ def test_tracker_comprehensive_workflow() -> None:
     # Start all nodes
     for node in nodes:
         tracker.start_task(node, mock_tasks[node])
-        tracker.set_task_logger(node, Mock())
 
     # Set up dependencies: model_3 depends on model_1 and model_2
     tracker.set_node_dependencies("model_3", ["model_1", "model_2"])


Code review comments:
diff_hunk: @@ -153,12 +167,27 @@ def manifest(self) -> Manifest:
         return self._manifest
 
     @property
-    def project(self) -> RuntimeConfig:
+    def project(self) -> Project:
         if self._project is None:
             self._set_project_from_project_dir()
             assert self._project is not None  # for type checking
         return self._project
 
+    @property
+    def graph(self) -> Graph:
+        if self._graph is None:
+            self._set_graph_from_manifest()
+            assert self._graph is not None  # for type checking
path: src/integrations/prefect-dbt/prefect_dbt/core/runner.py
text: @user1:
```suggestion
            assert self._graph is not None 
```

diff_hunk: @@ -294,3 +294,207 @@ def test_settings_handle_complex_profiles_structure():
             == "dev-db.example.com"
         )
         assert loaded_profiles["test_profile"]["targets"]["test"]["type"] == "sqlite"
+
+
+@pytest.fixture
+def mock_prefect_blocks(monkeypatch: pytest.MonkeyPatch):
+    """Fixture to mock Prefect blocks for testing."""
+
+    # Mock block document references resolution
+    def mock_resolve_blocks(data, value_transformer=None):
+        # Simulate resolving block references
+        if isinstance(data, dict):
+            resolved = {}
+            for key, value in data.items():
+                if isinstance(value, str) and "{{ prefect.blocks.secret." in value:
+                    # Extract block name and return mock value
+                    block_name = value.split("prefect.blocks.secret.")[1].split(" }}")[
+                        0
+                    ]
+                    mock_values = {
+                        "db-host": "localhost",
+                        "db-password": "secret_password_123",
+                        "db-user": "dbt_user",
+                    }
+                    resolved[key] = mock_values.get(
+                        block_name, f"resolved_{block_name}"
+                    )
+                elif isinstance(value, (dict, list)):
+                    resolved[key] = mock_resolve_blocks(value, value_transformer)
+                else:
+                    resolved[key] = value
+            return resolved
+        elif isinstance(data, list):
+            return [mock_resolve_blocks(item, value_transformer) for item in data]
+        else:
+            return data
path: src/integrations/prefect-dbt/tests/core/test_settings.py
text: @user1:
why do we need this stuff?

@author:
We definitely don't, I've still been struggling with making tests that are comprehensive but not repetitive. I'm considering starting from scratch because I'm not satisfied with how they're looking right now.