[components] Ensure dg component check uses configured components pat…

…h and other tweaks (#27386)

## Summary & Motivation

This makes a few tweaks around the new `dg component check` command and
it's tests:

- Adds a clause to ensure this is running inside a code location
(something we do for all `dg component ...` commands)
- Change to use the parametrized component path derived from context
instead of a hardcoded components path (which will break if the user
tries to configure the components path)
- Does a little refactoring around the tests to make them more coherent
with the rest of `dagster-dg` tests. In particular:
- makes the `create_code_location_from_components` utility build off of
`isolated_example_code_location_foo_bar`, which is used everywhere else
    - nests the `check` tests with the other `dg component ...` tests

## How I Tested These Changes

New unit tests.

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

@@ -285,14 +285,14 @@ def component_check_command(
 
     cli_config = normalize_cli_config(global_options, context)
     dg_context = DgContext.from_config_file_discovery_and_cli_config(Path.cwd(), cli_config)
+    if not dg_context.is_code_location:
+        exit_with_error("This command must be run inside a Dagster code location directory.")
 
     validation_errors: list[tuple[Optional[str], ValidationError, ValueAndSourcePositionTree]] = []
 
     component_contents_by_dir = {}
     local_component_dirs = set()
-    for component_dir in (
-        dg_context.root_path / dg_context.root_package_name / "components"
-    ).iterdir():
+    for component_dir in dg_context.components_path.iterdir():
         if resolved_paths and not any(
             path == component_dir or path in component_dir.parents for path in resolved_paths
         ):

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

@@ -39,7 +39,10 @@ def ensure_loadable_path(path: Path) -> Iterator[None]:
 
 
 def is_package_installed(package_name: str) -> bool:
-    return bool(importlib.util.find_spec(package_name))
+    try:
+        return bool(importlib.util.find_spec(package_name))
+    except ModuleNotFoundError:
+        return False
 
 
 def get_path_for_package(package_name: str) -> str:

python_modules/libraries/dagster-dg/dagster_dg_tests/cli_tests/test_check.py → python_modules/libraries/dagster-dg/dagster_dg_tests/cli_tests/component_command_tests/test_check.py

@@ -1,10 +1,7 @@
 import contextlib
-import os
 import re
 import shutil
-import tempfile
 from collections.abc import Iterator
-from contextlib import contextmanager
 from pathlib import Path
 from typing import Optional
 
@@ -17,13 +14,19 @@
     ComponentValidationTestCase,
     msg_includes_all_of,
 )
-from dagster_dg.utils import discover_git_root, ensure_dagster_dg_tests_import
+from dagster_dg.utils import ensure_dagster_dg_tests_import, pushd
 
 ensure_dagster_dg_tests_import()
-from dagster_dg_tests.utils import ProxyRunner, clear_module_from_cache
+from dagster_dg_tests.utils import (
+    ProxyRunner,
+    assert_runner_result,
+    isolated_example_code_location_foo_bar,
+    isolated_example_deployment_foo,
+    modify_pyproject_toml,
+)
 
 COMPONENT_INTEGRATION_TEST_DIR = (
-    Path(__file__).parent.parent.parent.parent
+    Path(__file__).parent.parent.parent.parent.parent
     / "dagster-components"
     / "dagster_components_tests"
     / "integration_tests"
@@ -44,52 +47,52 @@
 ]
 
 
-@contextmanager
-def new_cwd(path: str) -> Iterator[None]:
-    old = os.getcwd()
-    try:
-        os.chdir(path)
-        yield
-    finally:
-        os.chdir(old)
-
-
 @contextlib.contextmanager
 def create_code_location_from_components(
     runner: ProxyRunner, *src_paths: str, local_component_defn_to_inject: Optional[Path] = None
 ) -> Iterator[Path]:
     """Scaffolds a code location with the given components in a temporary directory,
     injecting the provided local component defn into each component's __init__.py.
     """
-    dagster_git_repo_dir = str(discover_git_root(Path(__file__)))
-    with tempfile.TemporaryDirectory() as tmpdir, new_cwd(tmpdir):
-        runner.invoke(
-            "code-location",
-            "scaffold",
-            "--use-editable-dagster",
-            dagster_git_repo_dir,
-            "my_location",
-        )
-
-        code_location_dir = Path(tmpdir) / "my_location"
-        assert code_location_dir.exists()
-
-        (code_location_dir / "lib").mkdir(parents=True, exist_ok=True)
-        (code_location_dir / "lib" / "__init__.py").touch()
+    origin_paths = [COMPONENT_INTEGRATION_TEST_DIR / src_path for src_path in src_paths]
+    with isolated_example_code_location_foo_bar(runner, component_dirs=origin_paths):
         for src_path in src_paths:
-            component_name = src_path.split("/")[-1]
+            components_dir = Path.cwd() / "foo_bar" / "components" / src_path.split("/")[-1]
+            if local_component_defn_to_inject:
+                shutil.copy(local_component_defn_to_inject, components_dir / "__init__.py")
 
-            components_dir = code_location_dir / "my_location" / "components" / component_name
-            components_dir.mkdir(parents=True, exist_ok=True)
+        yield Path.cwd()
 
-            origin_path = COMPONENT_INTEGRATION_TEST_DIR / src_path
 
-            shutil.copytree(origin_path, components_dir, dirs_exist_ok=True)
-            if local_component_defn_to_inject:
-                shutil.copy(local_component_defn_to_inject, components_dir / "__init__.py")
+def test_component_check_outside_code_location_fails() -> None:
+    with ProxyRunner.test() as runner, isolated_example_deployment_foo(runner):
+        result = runner.invoke("component", "check")
+        assert_runner_result(result, exit_0=False)
+        assert "must be run inside a Dagster code location directory" in result.output
+
+
+def test_component_check_succeeds_non_default_component_package() -> None:
+    with (
+        ProxyRunner.test() as runner,
+        create_code_location_from_components(
+            runner,
+        ),
+    ):
+        with modify_pyproject_toml() as pyproject_toml:
+            pyproject_toml["tool"]["dg"]["component_package"] = "foo_bar._components"
+
+        # We need to do all of this copying here rather than relying on the code location setup
+        # fixture because that fixture assumes a default component package.
+        component_src_path = COMPONENT_INTEGRATION_TEST_DIR / BASIC_VALID_VALUE.component_path
+        component_name = component_src_path.name
+        components_dir = Path.cwd() / "foo_bar" / "_components" / component_name
+        components_dir.mkdir(parents=True, exist_ok=True)
+        shutil.copytree(component_src_path, components_dir, dirs_exist_ok=True)
+        assert BASIC_VALID_VALUE.component_type_filepath
+        shutil.copy(BASIC_VALID_VALUE.component_type_filepath, components_dir / "__init__.py")
 
-        with clear_module_from_cache("my_location"):
-            yield code_location_dir
+        result = runner.invoke("component", "check")
+        assert_runner_result(result, exit_0=True)
 
 
 @pytest.mark.parametrize(
@@ -109,7 +112,7 @@ def test_validation_cli(test_case: ComponentValidationTestCase) -> None:
             local_component_defn_to_inject=test_case.component_type_filepath,
         ) as tmpdir,
     ):
-        with new_cwd(str(tmpdir)):
+        with pushd(tmpdir):
             result = runner.invoke("component", "check")
             if test_case.should_error:
                 assert result.exit_code != 0, str(result.stdout)
@@ -142,14 +145,14 @@ def test_validation_cli_multiple_components(scope_check_run: bool) -> None:
             local_component_defn_to_inject=BASIC_MISSING_VALUE.component_type_filepath,
         ) as tmpdir,
     ):
-        with new_cwd(str(tmpdir)):
+        with pushd(str(tmpdir)):
             result = runner.invoke(
                 "component",
                 "check",
                 *(
                     [
-                        str(Path("my_location") / "components" / "basic_component_missing_value"),
-                        str(Path("my_location") / "components" / "basic_component_invalid_value"),
+                        str(Path("foo_bar") / "components" / "basic_component_missing_value"),
+                        str(Path("foo_bar") / "components" / "basic_component_invalid_value"),
                     ]
                     if scope_check_run
                     else []
@@ -173,11 +176,11 @@ def test_validation_cli_multiple_components_filter() -> None:
             local_component_defn_to_inject=BASIC_MISSING_VALUE.component_type_filepath,
         ) as tmpdir,
     ):
-        with new_cwd(str(tmpdir)):
+        with pushd(tmpdir):
             result = runner.invoke(
                 "component",
                 "check",
-                str(Path("my_location") / "components" / "basic_component_missing_value"),
+                str(Path("foo_bar") / "components" / "basic_component_missing_value"),
             )
             assert result.exit_code != 0, str(result.stdout)
 
@@ -200,7 +203,7 @@ def test_validation_cli_local_component_cache() -> None:
             local_component_defn_to_inject=BASIC_VALID_VALUE.component_type_filepath,
         ) as code_location_dir,
     ):
-        with new_cwd(str(code_location_dir)):
+        with pushd(code_location_dir):
             result = runner.invoke("component", "check")
             assert re.search(
                 r"CACHE \[write\].*basic_component_success.*local_component_registry", result.stdout
@@ -223,14 +226,14 @@ def test_validation_cli_local_component_cache() -> None:
             # Update local component type, to invalidate cache
             contents = (
                 code_location_dir
-                / "my_location"
+                / "foo_bar"
                 / "components"
                 / "basic_component_success"
                 / "__init__.py"
             ).read_text()
             (
                 code_location_dir
-                / "my_location"
+                / "foo_bar"
                 / "components"
                 / "basic_component_success"
                 / "__init__.py"

python_modules/libraries/dagster-dg/dagster_dg_tests/utils.py

@@ -1,3 +1,4 @@
+import shutil
 import sys
 import traceback
 from collections.abc import Iterator, Sequence
@@ -32,34 +33,43 @@ def isolated_example_deployment_foo(runner: Union[CliRunner, "ProxyRunner"]) ->
 # of hyphenation.
 @contextmanager
 def isolated_example_code_location_foo_bar(
-    runner: Union[CliRunner, "ProxyRunner"], in_deployment: bool = True, skip_venv: bool = False
+    runner: Union[CliRunner, "ProxyRunner"],
+    in_deployment: bool = True,
+    skip_venv: bool = False,
+    component_dirs: Sequence[Path] = [],
 ) -> Iterator[None]:
+    """Scaffold a code location named foo_bar in an isolated filesystem.
+
+    Args:
+        runner: The runner to use for invoking commands.
+        in_deployment: Whether the code location should be scaffolded inside a deployment directory.
+        skip_venv: Whether to skip creating a virtual environment when scaffolding the code location.
+        component_dirs: A list of component directories that will be copied into the code location component root.
+    """
     runner = ProxyRunner(runner) if isinstance(runner, CliRunner) else runner
     dagster_git_repo_dir = str(discover_git_root(Path(__file__)))
     if in_deployment:
-        with isolated_example_deployment_foo(runner):
-            runner.invoke(
-                "code-location",
-                "scaffold",
-                "--use-editable-dagster",
-                dagster_git_repo_dir,
-                *(["--no-use-dg-managed-environment"] if skip_venv else []),
-                "foo-bar",
-            )
-            with clear_module_from_cache("foo_bar"), pushd("code_locations/foo-bar"):
-                yield
+        fs_context = isolated_example_deployment_foo(runner)
+        code_loc_path = "code_locations/foo-bar"
     else:
-        with runner.isolated_filesystem():
-            runner.invoke(
-                "code-location",
-                "scaffold",
-                "--use-editable-dagster",
-                dagster_git_repo_dir,
-                *(["--no-use-dg-managed-environment"] if skip_venv else []),
-                "foo-bar",
-            )
-            with clear_module_from_cache("foo_bar"), pushd("foo-bar"):
-                yield
+        fs_context = runner.isolated_filesystem()
+        code_loc_path = "foo-bar"
+    with fs_context:
+        runner.invoke(
+            "code-location",
+            "scaffold",
+            "--use-editable-dagster",
+            dagster_git_repo_dir,
+            *(["--no-use-dg-managed-environment"] if skip_venv else []),
+            "foo-bar",
+        )
+        with clear_module_from_cache("foo_bar"), pushd(code_loc_path):
+            for src_dir in component_dirs:
+                component_name = src_dir.name
+                components_dir = Path.cwd() / "foo_bar" / "components" / component_name
+                components_dir.mkdir(parents=True, exist_ok=True)
+                shutil.copytree(src_dir, components_dir, dirs_exist_ok=True)
+            yield
 
 
 @contextmanager