[components] Reenable dg check definitions in workspace test

Author: smackesey

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

@@ -27,14 +27,11 @@
     NOT_PROJECT_ERROR_MESSAGE,
     NOT_WORKSPACE_ERROR_MESSAGE,
     NOT_WORKSPACE_OR_PROJECT_ERROR_MESSAGE,
-    ensure_loadable_path,
     exit_with_error,
     generate_missing_dagster_components_in_local_venv_error_message,
-    get_path_for_module,
     get_toml_value,
     get_venv_executable,
     has_toml_value,
-    is_package_installed,
     pushd,
     resolve_local_venv,
     strip_activated_venv_from_env_vars,
@@ -217,7 +214,7 @@ def get_cache_key(self, data_type: Union[CachableDataType, str]) -> tuple[str, s
 
     def get_cache_key_for_module(self, module_name: str) -> tuple[str, str, str]:
         if module_name.startswith(self.root_module_name):
-            path = self.get_path_for_module(module_name)
+            path = self.get_path_for_local_module(module_name)
             env_hash = hash_paths([path], includes=["*.py"])
             path_parts = [str(part) for part in path.parts if part != "/"]
             return ("_".join(path_parts), env_hash, "local_component_registry")
@@ -306,7 +303,7 @@ def components_module_name(self) -> str:
     def components_path(self) -> Path:
         if not self.is_project:
             raise DgError("`components_path` is only available in a Dagster project context")
-        return self.get_path_for_module(self.components_module_name)
+        return self.get_path_for_local_module(self.components_module_name)
 
     def get_component_instance_names(self) -> Iterable[str]:
         return [str(instance_path.name) for instance_path in self.components_path.iterdir()]
@@ -327,7 +324,7 @@ def definitions_module_name(self) -> str:
 
     @cached_property
     def definitions_path(self) -> Path:
-        return self.get_path_for_module(self.definitions_module_name)
+        return self.get_path_for_local_module(self.definitions_module_name)
 
     # ########################
     # ##### COMPONENT LIBRARY METHODS
@@ -353,7 +350,7 @@ def default_components_library_module(self) -> str:
     def default_components_library_path(self) -> Path:
         if not self.is_component_library:
             raise DgError("`components_lib_path` is only available in a component library context")
-        return self.get_path_for_module(self.default_components_library_module)
+        return self.get_path_for_local_module(self.default_components_library_module)
 
     @cached_property
     def _dagster_components_entry_points(self) -> Mapping[str, str]:
@@ -450,16 +447,20 @@ def environment_desc(self) -> str:
     def pyproject_toml_path(self) -> Path:
         return self.root_path / "pyproject.toml"
 
-    def get_path_for_module(self, module_name: str) -> Path:
-        with ensure_loadable_path(self.root_path):
-            parts = module_name.split(".")
-            for i in range(len(parts)):
-                leading_module_name = ".".join(parts[: i + 1])
-                if not is_package_installed(leading_module_name):
-                    raise DgError(
-                        f"Module `{leading_module_name}` is not installed in the current environment."
-                    )
-            return Path(get_path_for_module(module_name))
+    def get_path_for_local_module(self, module_name: str) -> Path:
+        if not self.is_project and not self.is_component_library:
+            raise DgError(
+                "`get_path_for_local_module` is only available in a project or component library context"
+            )
+        if not module_name.startswith(self.root_module_name):
+            raise DgError(f"Module `{module_name}` is not part of the current project.")
+        relative_path = Path(*module_name.split("."))
+        if (self.root_path / relative_path).exists():
+            return self.root_path / relative_path
+        elif (self.root_path / relative_path).with_suffix(".py").exists():
+            return (self.root_path / relative_path).with_suffix(".py")
+        else:
+            raise DgError(f"Cannot find module `{module_name}` in the current project.")
 
 
 # ########################

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

@@ -1,5 +1,4 @@
 import contextlib
-import importlib.util
 import json
 import os
 import posixpath
@@ -9,7 +8,6 @@
 import textwrap
 from collections.abc import Iterable, Iterator, Mapping, Sequence
 from fnmatch import fnmatch
-from importlib.machinery import ModuleSpec
 from pathlib import Path
 from typing import Any, Optional, TypeVar, Union
 
@@ -61,44 +59,6 @@ def install_to_venv(venv_dir: Path, install_args: list[str]) -> None:
     subprocess.run(command, check=True)
 
 
-# Temporarily places a path at the front of sys.path, ensuring that any modules in that path are
-# importable.
-@contextlib.contextmanager
-def ensure_loadable_path(path: Path) -> Iterator[None]:
-    orig_path = sys.path.copy()
-    sys.path.insert(0, str(path))
-    try:
-        yield
-    finally:
-        sys.path = orig_path
-
-
-def is_package_installed(package_name: str) -> bool:
-    try:
-        return bool(importlib.util.find_spec(package_name))
-    except ModuleNotFoundError:
-        return False
-
-
-def _get_spec_for_module(module_name: str) -> ModuleSpec:
-    spec = importlib.util.find_spec(module_name)
-    if not spec:
-        raise DgError(f"Cannot find module: {module_name}")
-    return spec
-
-
-def get_path_for_module(module_name: str) -> str:
-    spec = _get_spec_for_module(module_name)
-    submodule_search_locations = spec.submodule_search_locations
-    if submodule_search_locations:  # branch module (i.e. package), a directory
-        return submodule_search_locations[0]
-    else:  # leaf module, not a directory
-        file_path = spec.origin
-        if not file_path:
-            raise DgError(f"Cannot find file path for module: {module_name}")
-        return file_path
-
-
 def is_valid_json(value: str) -> bool:
     try:
         json.loads(value)

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

@@ -397,7 +397,7 @@ def test_scaffold_component_command_with_non_matching_module_name():
             "qux",
         )
         assert_runner_result(result, exit_0=False)
-        assert "Module `foo_bar` is not installed" in str(result.exception)
+        assert "Cannot find module `foo_bar.lib`" in str(result.exception)
 
 
 @pytest.mark.parametrize("in_workspace", [True, False])
@@ -456,7 +456,7 @@ def test_scaffold_component_fails_components_package_does_not_exist() -> None:
             "qux",
         )
         assert_runner_result(result, exit_0=False)
-        assert "Module `foo_bar._defs` is not installed" in str(result.exception)
+        assert "Cannot find module `foo_bar._defs`" in str(result.exception)
 
 
 def test_scaffold_component_succeeds_scaffolded_component_type() -> None:

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

@@ -11,8 +11,7 @@
 )
 
 
-# @pytest.mark.skipif(is_windows(), reason="Temporarily skipping (signal issues in CLI)..")
-@pytest.mark.skip  # Weirdness with environment, temporarily skipping
+@pytest.mark.skipif(is_windows(), reason="Temporarily skipping (signal issues in CLI)..")
 def test_validate_command_deployment_context_success():
     dagster_git_repo_dir = str(discover_git_root(Path(__file__)))
     with ProxyRunner.test() as runner, isolated_example_workspace(runner, create_venv=True):