refactor(tests): enhance command execution in DagsterTestContext

- Updated `_run_command` method to accept an optional `cwd` parameter for dynamic directory changes during command execution.
- Adjusted test case to specify the working directory for asset materialization, ensuring correct file placement.
- Added a sleep delay in the test to manage timing issues during execution.
- Cleaned up print statements for better clarity in command execution context.

Author: KLonge

dagster_sqlmesh/conftest.py

@@ -541,7 +541,7 @@ class DagsterTestContext:
     dagster_project_path: str
     sqlmesh_project_path: str
 
-    def _run_command(self, cmd: list[str]) -> None:
+    def _run_command(self, cmd: list[str], cwd: str | None = None) -> None:
         """Execute a command and stream its output in real-time.
 
         Args:
@@ -590,10 +590,10 @@ def stream_output(
 
         print(f"Running command: {' '.join(cmd)}")
         print(f"Current working directory: {os.getcwd()}")
-        print(f"Changing to directory: {self.dagster_project_path}")
 
-        # Change to the dagster project directory before running the command
-        os.chdir(self.dagster_project_path)
+        if cwd:
+            print(f"Changing to directory: {cwd}")
+            os.chdir(cwd)
 
         process = subprocess.Popen(
             cmd,
@@ -699,7 +699,8 @@ def asset_materialisation(
             config_json,
         ]
 
-        self._run_command(cmd)
+        # Change to the sqlmesh project directory before running the command (for some reason asset materialize needs to be run from the dirctory you want the db.db file to be in - feel free to investigate)
+        self._run_command(cmd=cmd, cwd=self.sqlmesh_project_path)
 
     def reset_assets(self) -> None:
         """Resets the assets to the original state"""

dagster_sqlmesh/controller/tests_plan_and_run/test_model_code_change.py

@@ -1,4 +1,5 @@
 import logging
+import time
 
 import pytest
 
@@ -237,11 +238,13 @@ def test_given_model_chain_when_running_with_different_flags_then_behaves_as_exp
 
     # sample_dagster_test_context.asset_materialisation(assets=["intermediate_model_1"], plan_options=PlanOptions(skip_backfill=True, enable_preview=True, skip_tests=True))
 
+    time.sleep(5)
+
     intermediate_model_1_df = (
         sample_sqlmesh_test_context.query(
             """
         SELECT *
-        FROM sqlmesh_example.intermediate_model_1
+        FROM sqlmesh_example__dev.intermediate_model_1
         """,
             return_df=True,
         )