[components] Add windows testing to dagster-dg

azure-pipelines.yml

@@ -28,21 +28,36 @@ jobs:
         ${{ each py_version in parameters.py3_versions }}:
           ${{ each env_suffix in parameters.py3_env_suffixes }}:
             ${{ replace(py_version, '.', '') }}-windows-${{ env_suffix }}:
-              TOXENV: "py${{ replace(py_version, '.', '') }}-windows-${{ env_suffix }}"
-              python.version: "${{ py_version }}"
+              PACKAGE_ROOT: "python_modules\\dagster"
+              PATH_BACK_TO_REPO_ROOT: "..\\.."
+              TOX_ENV: "py${{ replace(py_version, '.', '') }}-windows-${{ env_suffix }}"
+              PYTHON_VERSION: "${{ py_version }}"
+
+        ${{ each py_version in parameters.py3_versions }}:
+          dagster-dg-${{ replace(py_version, '.', '') }}:
+            PACKAGE_ROOT: "python_modules\\libraries\\dagster-dg"
+            TOX_ENV: windows
+            PYTHON_VERSION: "${{ py_version }}"
+            PATH_BACK_TO_REPO_ROOT: "..\\..\\.."
     variables:
-      PYTHONLEGACYWINDOWSSTDIO: "1"
+      PYTHONUTF8: "1"
     steps:
       - task: UsePythonVersion@0
         inputs:
-          versionSpec: "$(python.version)"
+          versionSpec: "$(PYTHON_VERSION)"
           architecture: "x64"
       - script: pip install "tox<4.0.0" uv
         displayName: "Install tox & uv"
-      - script: cd python_modules\dagster && tox -e %TOXENV% && cd ..\..
+      # Use PowerShell for UTF-8 support. If we do not have proper UTF-8 support, there can be
+      # anomalies in the rendering of certain characters in command output, which can break tests
+      # that look at e.g. fancy box characters in CLI --help output.
+      - powershell: |
+          cd $env:PACKAGE_ROOT
+          tox -e $env:TOX_ENV
+          cd $env:PATH_BACK_TO_REPO_ROOT
         displayName: "Run tests"
       - task: PublishTestResults@2
         inputs:
           testResultsFiles: "**/test_results.xml"
-          testRunTitle: "dagster $(TOXENV)"
+          testRunTitle: "$(PACKAGE_ROOT) $(TOX_ENV)"
         condition: succeededOrFailed()

python_modules/libraries/dagster-dg/dagster_dg/cache.py

@@ -3,7 +3,7 @@
 import sys
 from collections.abc import Sequence
 from pathlib import Path
-from typing import Final, Literal, Optional, Union
+from typing import Final, Literal, Optional
 
 from typing_extensions import Self, TypeAlias
 
@@ -16,9 +16,7 @@
 
 _CACHE_CONTAINER_DIR_NAME: Final = "dg-cache"
 
-CachableDataType: TypeAlias = Union[
-    Literal["component_registry_data"], Literal["all_components_schema"]
-]
+CachableDataType: TypeAlias = Literal["component_registry_data", "all_components_schema"]
 
 
 def get_default_cache_dir() -> Path:
@@ -77,11 +75,16 @@ def clear_key(self, key: tuple[str, ...]) -> None:
             self.log(f"CACHE [clear-key]: {path}")
 
     def clear_all(self) -> None:
+        print("DELETING", self._root_path)
         shutil.rmtree(self._root_path)
+        assert not self._root_path.exists()
+        print("CONFIRMED DELETED", self._root_path)
         self.log(f"CACHE [clear-all]: {self._root_path}")
 
     def get(self, key: tuple[str, ...]) -> Optional[str]:
         path = self._get_path(key)
+        print("PATH IS", path)
+        print("PATH EXISTS", path.exists())
         if path.exists():
             self.log(f"CACHE [hit]: {path}")
             return path.read_text()
@@ -96,6 +99,16 @@ def set(self, key: tuple[str, ...], value: str) -> None:
         self.log(f"CACHE [write]: {path}")
 
     def _get_path(self, key: tuple[str, ...]) -> Path:
+        print("GETTING PATH")
+        print("ROOT PATH IS", self._root_path)
+        print("ROOT PATH str() IS", str(self._root_path))
+        print("ROOT PATH repr() IS", repr(self._root_path))
+        my_path = self._root_path
+        for k in key:
+            print("MY PATH IS", my_path)
+            my_path = my_path / k
+        print("MY PATH IS", my_path)
+        print("PATH IS", Path(self._root_path, *key))
         return Path(self._root_path, *key)
 
     def log(self, message: str) -> None:

python_modules/libraries/dagster-dg/dagster_dg/cli/dev.py

@@ -1,9 +1,6 @@
-import os
-import signal
 import subprocess
-import sys
 import time
-from collections.abc import Iterator, Mapping, Sequence
+from collections.abc import Iterator, Mapping
 from contextlib import contextmanager, nullcontext
 from pathlib import Path
 from tempfile import NamedTemporaryFile
@@ -17,7 +14,13 @@
 from dagster_dg.config import normalize_cli_config
 from dagster_dg.context import DgContext
 from dagster_dg.error import DgError
-from dagster_dg.utils import DgClickCommand, exit_with_error, pushd
+from dagster_dg.utils import (
+    DgClickCommand,
+    exit_with_error,
+    interrupt_subprocess,
+    open_subprocess,
+    pushd,
+)
 
 T = TypeVar("T")
 
@@ -132,7 +135,7 @@ def dev_command(
     with pushd(dg_context.root_path), temp_workspace_file_cm as workspace_file:
         if workspace_file:  # only non-None deployment context
             cmd.extend(["--workspace", workspace_file])
-        uv_run_dagster_dev_process = _open_subprocess(cmd)
+        uv_run_dagster_dev_process = open_subprocess(cmd)
         try:
             while True:
                 time.sleep(_CHECK_SUBPROCESS_INTERVAL)
@@ -151,7 +154,7 @@ def dev_command(
             # directly to the `dagster dev` process (the only child of the `uv run` process). This
             # will cause `dagster dev` to terminate which in turn will cause `uv run` to terminate.
             dagster_dev_pid = _get_child_process_pid(uv_run_dagster_dev_process)
-            _interrupt_subprocess(dagster_dev_pid)
+            interrupt_subprocess(dagster_dev_pid)
 
             try:
                 uv_run_dagster_dev_process.wait(timeout=10)
@@ -184,32 +187,9 @@ def _format_forwarded_option(option: str, value: object) -> list[str]:
     return [] if value is None else [option, str(value)]
 
 
-def _get_child_process_pid(proc: "subprocess.Popen") -> int:
-    children = psutil.Process(proc.pid).children(recursive=False)
+def _get_child_process_pid(proc: subprocess.Popen) -> int:
+    # Windows will sometimes return the parent process as its own child. Filter this out.
+    children = [p for p in psutil.Process(proc.pid).children(recursive=False) if p.pid != proc.pid]
     if len(children) != 1:
         raise ValueError(f"Expected exactly one child process, but found {len(children)}")
     return children[0].pid
-
-
-# Windows subprocess termination utilities. See here for why we send CTRL_BREAK_EVENT on Windows:
-# https://stefan.sofa-rockers.org/2013/08/15/handling-sub-process-hierarchies-python-linux-os-x/
-
-
-def _interrupt_subprocess(pid: int) -> None:
-    """Send CTRL_BREAK_EVENT on Windows, SIGINT on other platforms."""
-    if sys.platform == "win32":
-        os.kill(pid, signal.CTRL_BREAK_EVENT)
-    else:
-        os.kill(pid, signal.SIGINT)
-
-
-def _open_subprocess(command: Sequence[str]) -> "subprocess.Popen":
-    """Sets the correct flags to support graceful termination."""
-    creationflags = 0
-    if sys.platform == "win32":
-        creationflags = subprocess.CREATE_NEW_PROCESS_GROUP
-
-    return subprocess.Popen(
-        command,
-        creationflags=creationflags,
-    )

python_modules/libraries/dagster-dg/dagster_dg/completion.py

@@ -25,7 +25,7 @@ def install_completion(context: click.Context):
     shell, _ = shellingham.detect_shell()
     comp_class = click.shell_completion.get_completion_class(shell)
     if comp_class is None:
-        exit_with_error(f"Shell {shell} is not supported.")
+        exit_with_error(f"Shell `{shell}` is not supported.")
     else:
         comp_inst = comp_class(
             cli=context.command, ctx_args={}, prog_name="dg", complete_var="_DG_COMPLETE"

python_modules/libraries/dagster-dg/dagster_dg/context.py

@@ -21,6 +21,7 @@
     get_path_for_module,
     get_path_for_package,
     get_uv_command_env,
+    get_venv_executable,
     is_executable_available,
     is_package_installed,
     pushd,
@@ -212,7 +213,7 @@ def code_location_python_executable(self) -> Path:
             raise DgError(
                 "`code_location_python_executable` is only available in a code location context"
             )
-        return self.root_path / ".venv" / "bin" / "python"
+        return self.root_path / get_venv_executable(Path(".venv"))
 
     @cached_property
     def components_package_name(self) -> str:

python_modules/libraries/dagster-dg/dagster_dg/scaffold.py

@@ -72,7 +72,7 @@ def get_pyproject_toml_dev_dependencies(use_editable_dagster: bool) -> str:
 
 def get_pyproject_toml_uv_sources(editable_dagster_root: Path) -> str:
     lib_lines = [
-        f'{path.name} = {{ path = "{path}", editable = true }}'
+        f"{path.name} = {{ path = '{path}', editable = true }}"
         for path in _gather_dagster_packages(editable_dagster_root)
     ]
     return "\n".join(
@@ -140,7 +140,7 @@ def scaffold_component_type(dg_context: DgContext, name: str) -> None:
     scaffold_subtree(
         path=root_path,
         name_placeholder="COMPONENT_TYPE_NAME_PLACEHOLDER",
-        templates_path=os.path.join(os.path.dirname(__file__), "templates", "COMPONENT_TYPE"),
+        templates_path=str(Path(__file__).parent / "templates" / "COMPONENT_TYPE"),
         project_name=name,
         component_type_class_name=camelcase(name),
         name=name,

python_modules/libraries/dagster-dg/dagster_dg/utils/__init__.py

@@ -5,6 +5,8 @@
 import posixpath
 import re
 import shutil
+import signal
+import subprocess
 import sys
 import textwrap
 from collections.abc import Iterator, Mapping, Sequence
@@ -28,6 +30,10 @@
 CLI_CONFIG_KEY = "config"
 
 
+def get_venv_executable(venv_dir: Path) -> Path:
+    return venv_dir / ("Scripts" if sys.platform == "win32" else "bin") / "python"
+
+
 # Temporarily places a path at the front of sys.path, ensuring that any modules in that path are
 # importable.
 @contextlib.contextmanager
@@ -82,6 +88,12 @@ def is_executable_available(command: str) -> bool:
     return bool(shutil.which(command))
 
 
+# Short for "normalize path"-- use this to get the platform-correct string representation of an
+# existing string path.
+def npath(path: str):
+    return str(Path(path))
+
+
 # uv commands should be executed in an environment with no pre-existing VIRTUAL_ENV set. If this
 # variable is set (common during development) and does not match the venv resolved by uv, it prints
 # undesireable warnings.
@@ -283,6 +295,11 @@ def exit_with_error(error_message: str) -> Never:
     sys.exit(1)
 
 
+# ########################
+# ##### ERROR MESSAGES
+# ########################
+
+
 def _format_error_message(message: str) -> str:
     # width=10000 unwraps any hardwrapping
     return textwrap.fill(message, width=10000)
@@ -316,6 +333,34 @@ def _format_error_message(message: str) -> str:
 installed.
 """)
 
+# ########################
+# ##### SUBPROCESSES
+# ########################
+
+# Windows subprocess termination utilities. See here for why we send CTRL_BREAK_EVENT on Windows:
+# https://stefan.sofa-rockers.org/2013/08/15/handling-sub-process-hierarchies-python-linux-os-x/
+
+
+def interrupt_subprocess(pid: int, interrupt_signal: Any = None) -> None:
+    """Send CTRL_BREAK_EVENT on Windows, SIGINT on other platforms."""
+    if sys.platform == "win32":
+        os.kill(pid, interrupt_signal or signal.CTRL_BREAK_EVENT)
+    else:
+        os.kill(pid, interrupt_signal or signal.SIGINT)
+
+
+def open_subprocess(command: Sequence[str], **kwargs: Any) -> subprocess.Popen:
+    """Sets the correct flags to support graceful termination."""
+    creationflags = 0
+    if sys.platform == "win32":
+        creationflags = subprocess.CREATE_NEW_PROCESS_GROUP
+    return subprocess.Popen(
+        command,
+        creationflags=creationflags,
+        **kwargs,
+    )
+
+
 # ########################
 # ##### CUSTOM CLICK SUBCLASSES
 # ########################

python_modules/libraries/dagster-dg/dagster_dg_tests/cli_tests/component_command_tests/test_component_commands.py

@@ -4,7 +4,7 @@
 from pathlib import Path
 
 import pytest
-from dagster_dg.utils import ensure_dagster_dg_tests_import
+from dagster_dg.utils import ensure_dagster_dg_tests_import, npath
 
 ensure_dagster_dg_tests_import()
 
@@ -222,14 +222,14 @@ def test_component_scaffold_succeeds_scaffolded_component_type() -> None:
 # ##### REAL COMPONENTS
 
 
-dbt_project_path = "../stub_code_locations/dbt_project_location/components/jaffle_shop"
+dbt_project_path = Path("../stub_code_locations/dbt_project_location/components/jaffle_shop")
 
 
 @pytest.mark.parametrize(
     "params",
     [
         ["--json-params", json.dumps({"project_path": str(dbt_project_path)})],
-        ["--project-path", dbt_project_path],
+        ["--project-path", str(dbt_project_path)],
     ],
 )
 def test_scaffold_dbt_project_instance(params) -> None:
@@ -254,7 +254,7 @@ def test_scaffold_dbt_project_instance(params) -> None:
         assert component_yaml_path.exists()
         assert "type: dbt_project@dagster_components" in component_yaml_path.read_text()
         assert (
-            "stub_code_locations/dbt_project_location/components/jaffle_shop"
+            npath("stub_code_locations/dbt_project_location/components/jaffle_shop")
             in component_yaml_path.read_text()
         )
 

python_modules/libraries/dagster-dg/dagster_dg_tests/cli_tests/test_code_location_commands.py

@@ -102,25 +102,27 @@ def test_code_location_scaffold_editable_dagster_success(mode: str, monkeypatch)
         with open("code_locations/foo-bar/pyproject.toml") as f:
             toml = tomli.loads(f.read())
             assert toml["tool"]["uv"]["sources"]["dagster"] == {
-                "path": f"{dagster_git_repo_dir}/python_modules/dagster",
+                "path": f"{dagster_git_repo_dir / 'python_modules' / 'dagster'}",
                 "editable": True,
             }
             assert toml["tool"]["uv"]["sources"]["dagster-pipes"] == {
-                "path": f"{dagster_git_repo_dir}/python_modules/dagster-pipes",
+                "path": str(dagster_git_repo_dir / "python_modules" / "dagster-pipes"),
                 "editable": True,
             }
             assert toml["tool"]["uv"]["sources"]["dagster-webserver"] == {
-                "path": f"{dagster_git_repo_dir}/python_modules/dagster-webserver",
+                "path": str(dagster_git_repo_dir / "python_modules" / "dagster-webserver"),
                 "editable": True,
             }
             assert toml["tool"]["uv"]["sources"]["dagster-components"] == {
-                "path": f"{dagster_git_repo_dir}/python_modules/libraries/dagster-components",
+                "path": str(
+                    dagster_git_repo_dir / "python_modules" / "libraries" / "dagster-components"
+                ),
                 "editable": True,
             }
             # Check for presence of one random package with no component to ensure we are
             # preemptively adding all packages
             assert toml["tool"]["uv"]["sources"]["dagstermill"] == {
-                "path": f"{dagster_git_repo_dir}/python_modules/libraries/dagstermill",
+                "path": str(dagster_git_repo_dir / "python_modules" / "libraries" / "dagstermill"),
                 "editable": True,
             }