feat(tests): enhance testing capabilities with new Dagster context

- Introduced `DagsterTestContext` for improved asset materialization in tests.
- Updated `sample_sqlmesh_project` and `sample_dagster_project` fixtures to create temporary projects for testing.
- Modified `SQLMeshResource` to support plan and run options overrides.
- Refactored existing test cases to utilize the new context and improve clarity.
- Updated `Makefile` to change asset materialization command and configuration for better usability.

Author: KLonge

Makefile

@@ -57,16 +57,18 @@ clean-dagster:
 	rm -rf sample/dagster_project/storage sample/dagster_project/logs sample/dagster_project/history
 
 clean-db:
-	$(PYTHON_CMD) -c "import duckdb; conn = duckdb.connect('db.db'); [conn.execute(cmd[0]) for cmd in conn.execute(\"\"\"SELECT 'DROP TABLE ' || table_schema || '.' || table_name || ' CASCADE;' as drop_cmd FROM information_schema.tables WHERE table_schema != 'sources' AND table_schema != 'information_schema' AND table_type = 'BASE TABLE'\"\"\").fetchall()]; [conn.execute(cmd[0]) for cmd in conn.execute(\"\"\"SELECT 'DROP VIEW ' || table_schema || '.' || table_name || ' CASCADE;' as drop_cmd FROM information_schema.tables WHERE table_schema != 'sources' AND table_schema != 'information_schema' AND table_type = 'VIEW'\"\"\").fetchall()]; conn.close()"
+	"$(PYTHON_CMD)" -c "import duckdb; conn = duckdb.connect('db.db'); [conn.execute(cmd[0]) for cmd in conn.execute(\"\"\"SELECT 'DROP TABLE ' || table_schema || '.' || table_name || ' CASCADE;' as drop_cmd FROM information_schema.tables WHERE table_schema != 'sources' AND table_schema != 'information_schema' AND table_type = 'BASE TABLE'\"\"").fetchall()]; [conn.execute(cmd[0]) for cmd in conn.execute(\"\"\"SELECT 'DROP VIEW ' || table_schema || '.' || table_name || ' CASCADE;' as drop_cmd FROM information_schema.tables WHERE table_schema != 'sources' AND table_schema != 'information_schema' AND table_type = 'VIEW'\"\"").fetchall()]; conn.close()"
 
 dagster-dev: clean-dagster
 	@DAGSTER_HOME="$(subst \,/,$(CURDIR))/sample/dagster_project" "$(PYTHON_CMD)" -m dagster dev -f sample/dagster_project/definitions.py -h 0.0.0.0
 
 dev: dagster-dev  # Alias for dagster-dev
 
 dagster-materialize:
-	"$(PYTHON_CMD)" -m dagster asset materialize -f "$(CURDIR)/sample/dagster_project/definitions.py" --select "test_source"
-
+	"$(PYTHON_CMD)" -m dagster asset materialize \
+		-f "$(CURDIR)/sample/dagster_project/definitions.py" \
+		--select "intermediate_model_1" \
+		--config-json '{"resources": {"sqlmesh": {"config": {"config": {"gateway": "local", "path": "C:\\\\Users\\\\kevin\\\\git_repos\\\\dagster-sqlmesh - Copy\\\\sample\\\\sqlmesh_project"}, "plan_options_override": {"skip_backfill": false}}}}}'
 
 sqlmesh-plan:
 	cd sample/sqlmesh_project && $(SQLMESH_CMD) plan

dagster_sqlmesh/conftest.py

@@ -1,6 +1,8 @@
+import json
 import logging
 import os
 import shutil
+import subprocess
 import sys
 import tempfile
 import typing as t
@@ -35,7 +37,7 @@ def setup_debug_logging_for_tests() -> None:
 
 
 @pytest.fixture
-def sample_sqlmesh_project() -> t.Generator[str, None, None]:
+def sample_sqlmesh_project() -> t.Iterator[str]:
     """Creates a temporary sqlmesh project by copying the sample project"""
     with tempfile.TemporaryDirectory() as tmp_dir:
         project_dir = shutil.copytree(
@@ -359,7 +361,7 @@ def plan_and_run(
 @pytest.fixture
 def sample_sqlmesh_test_context(
     sample_sqlmesh_project: str,
-) -> t.Generator[SQLMeshTestContext, None, None]:
+) -> t.Iterator[SQLMeshTestContext]:
     db_path = os.path.join(sample_sqlmesh_project, "db.db")
     config = SQLMeshConfig(
         gateways={
@@ -417,7 +419,7 @@ def permanent_sqlmesh_project() -> str:
 @pytest.fixture
 def model_change_test_context(
     permanent_sqlmesh_project: str,
-) -> t.Generator[SQLMeshTestContext, None, None]:
+) -> t.Iterator[SQLMeshTestContext]:
     """FOR DEBUGGING ONLY: Creates a SQLMesh test context specifically for testing model code changes.
 
     This fixture provides a context that allows modifying SQL model files and ensures
@@ -455,5 +457,109 @@ def model_change_test_context(
     # test_context.cleanup_modified_files()
 
 
+@dataclass
+class DagsterTestContext:
+    """A test context for running Dagster"""
+
+    project_path: str
+
+    def asset_materialisation(
+        self,
+        assets: list[str],
+        plan_options: PlanOptions | None = None,
+        run_options: RunOptions | None = None,
+    ) -> None:
+        """Materialises the given Dagster assets using CLI command.
+
+        Args:
+            assets: String of comma-separated asset names to materialize
+            plan_options: Optional SQLMesh plan options to pass to the config
+            run_options: Optional SQLMesh run options to pass to the config
+        """
+
+        # Construct the base config
+        config: dict[str, t.Any] = {
+            "resources": {
+                "sqlmesh": {
+                    "config": {
+                        "config": {"gateway": "local", "path": self.project_path}
+                    }
+                }
+            }
+        }
+
+        # Add plan options if provided
+        if plan_options:
+            config["resources"]["sqlmesh"]["config"]["plan_options_override"] = {
+                k: v for k, v in plan_options.items() if v is not None
+            }
+
+        # Add run options if provided
+        if run_options:
+            config["resources"]["sqlmesh"]["config"]["run_options_override"] = {
+                k: v for k, v in run_options.items() if v is not None
+            }
+
+        # Convert config to JSON string, escaping backslashes for Windows paths
+        config_json = json.dumps(config).replace("\\", "\\\\")
+
+        # Construct the command
+        cmd = [
+            sys.executable,
+            "-m",
+            "dagster",
+            "asset",
+            "materialize",
+            "-f",
+            os.path.join(self.project_path, "definitions.py"),
+            "--select",
+            ",".join(assets),
+            "--config-json",
+            config_json,
+        ]
+
+        # Run the command
+        subprocess.run(cmd, check=True)
+
+    def reset_assets(self) -> None:
+        """Resets the assets to the original state"""
+        self.asset_materialisation(assets=["reset_asset"])
+
+    def init_test_source(self) -> None:
+        """Initialises the test source"""
+        self.asset_materialisation(assets=["test_source"])
+
+
+
+@pytest.fixture
+def sample_dagster_project() -> t.Iterator[str]:
+    """Creates a temporary dagster project by copying the sample project"""
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        project_dir = shutil.copytree(
+            "sample",
+            tmp_dir,
+        )
+        dagster_project_dir = os.path.join(project_dir, "dagster_project")
+        sqlmesh_project_dir = os.path.join(project_dir, "sqlmesh_project")
+
+        db_path = os.path.join(sqlmesh_project_dir, "db.db")
+        if os.path.exists(db_path):
+            os.remove(os.path.join(sqlmesh_project_dir, "db.db"))
+
+        # Initialize the "source" data
+        yield str(dagster_project_dir)
+
+
+
+@pytest.fixture
+def sample_dagster_test_context(
+    sample_dagster_project: str,
+) -> t.Iterator[DagsterTestContext]:
+    test_context = DagsterTestContext(
+        project_path=os.path.join(sample_dagster_project),
+    )
+    yield test_context
+
+
 if __name__ == "__main__":
-    pytest.main([__file__])
+    pytest.main([__file__])