Create hatch command wrappers (#21135)

* Create wrapper around hatch commands and refactor `ddev env test` to use
it. This simplifies the logic for environment selection.

* Fix changelog number

Author: AAraKKe

ddev/changelog.d/21135.added

@@ -0,0 +1 @@
+Add a utils.hatch module to centralize hatch operations

ddev/src/ddev/cli/env/test.py

@@ -9,6 +9,7 @@
 
 if TYPE_CHECKING:
     from ddev.cli.application import Application
+    from ddev.utils.hatch import Environment
 
 
 @click.command('test')
@@ -72,13 +73,16 @@ def test_command(
     \b
     https://datadoghq.dev/integrations-core/testing/
     """
+    from functools import partial
+
     from ddev.cli.env.start import start
     from ddev.cli.env.stop import stop
     from ddev.cli.test import test
     from ddev.config.constants import AppEnvVars
     from ddev.e2e.config import EnvDataStorage
     from ddev.e2e.constants import E2EMetadata
     from ddev.utils.ci import running_in_ci
+    from ddev.utils.hatch import HatchCommandError, list_environment_names
     from ddev.utils.structures import EnvVars
 
     app: Application = ctx.obj
@@ -93,29 +97,19 @@ def test_command(
     if environment == 'active':
         env_names = active_envs
     else:
-        import json
-        import sys
-
-        with integration.path.as_cwd():
-            env_data_output = app.platform.check_command_output(
-                [sys.executable, '-m', 'hatch', '--no-color', '--no-interactive', 'env', 'show', '--json']
+        try:
+            env_names = list_environment_names(
+                app.platform,
+                integration,
+                filters=[
+                    is_e2e_environment,
+                    partial(uses_python_version, python_filter=python_filter),
+                    partial(uses_platform, platform=app.platform.name),
+                    partial(is_selected_environment, environment_name=environment),
+                ],
             )
-            try:
-                environments = json.loads(env_data_output)
-            except json.JSONDecodeError:
-                app.abort(f'Failed to parse environments for `{integration.name}`:\n{repr(env_data_output)}')
-
-        no_python_filter = python_filter is None
-        all_environments = environment == 'all'
-
-        env_names = [
-            name
-            for name, data in environments.items()
-            if data.get('e2e-env', False)
-            and (not data.get('platforms') or app.platform.name in data['platforms'])
-            and (no_python_filter or data.get('python') == python_filter)
-            and (name == environment or all_environments)
-        ]
+        except HatchCommandError as error:
+            app.abort(f'Failed to list environments for `{integration.name}`:\n{error}')
 
     if not env_names:
         app.display_info(f"Selected target {integration.name!r} disabled by e2e-env option.")
@@ -155,3 +149,19 @@ def test_command(
                 )
         finally:
             ctx.invoke(stop, intg_name=intg_name, environment=env_name, ignore_state=env_active)
+
+
+def is_e2e_environment(environment: Environment) -> bool:
+    return environment.e2e_env
+
+
+def uses_python_version(environment: Environment, python_filter: str | None) -> bool:
+    return python_filter is None or environment.python == python_filter
+
+
+def uses_platform(environment: Environment, platform: str) -> bool:
+    return not environment.platforms or platform in environment.platforms
+
+
+def is_selected_environment(environment: Environment, environment_name: str) -> bool:
+    return environment.name == environment_name or environment_name == 'all'

ddev/src/ddev/cli/test/__init__.py

@@ -130,8 +130,8 @@ def test(
     import sys
 
     from ddev.testing.constants import EndToEndEnvVars, TestEnvVars
-    from ddev.testing.hatch import get_hatch_env_vars
     from ddev.utils.ci import running_in_ci
+    from ddev.utils.hatch import get_hatch_env_vars
 
     if target_spec is None:
         target_spec = 'changed'

ddev/src/ddev/e2e/run.py

@@ -8,7 +8,7 @@
 from typing import Generator
 
 from ddev.e2e.constants import E2EEnvVars
-from ddev.testing.hatch import get_hatch_env_vars
+from ddev.utils.hatch import get_hatch_env_vars
 from ddev.utils.structures import EnvVars
 
 

ddev/src/ddev/testing/hatch.py

@@ -0,0 +1,122 @@
+# (C) Datadog, Inc. 2025-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Protocol, overload
+
+from pydantic import BaseModel, Field, RootModel, model_validator
+
+if TYPE_CHECKING:
+    from collections.abc import Collection
+    from typing import Any, Literal
+
+    from ddev.integration.core import Integration
+
+    from .platform import Platform
+
+
+class HatchCommandError(Exception):
+    pass
+
+
+class Environment(BaseModel):
+    """Represents a single environment configuration."""
+
+    name: str
+    type: str
+    dependencies: list[str] = Field(default_factory=list)
+    test_env: bool = Field(alias='test-env')
+    e2e_env: bool = Field(alias='e2e-env')
+    benchmark_env: bool = Field(alias='benchmark-env')
+    latest_env: bool = Field(alias='latest-env')
+    python: str | None = None
+    scripts: dict[str, list[str]] = Field(default_factory=dict)
+    platforms: list[str] = Field(default_factory=list)
+    pre_install_commands: list[str] = Field(default_factory=list, alias='pre-install-commands')
+    post_install_commands: list[str] = Field(default_factory=list, alias='post-install-commands')
+    skip_install: bool = Field(False, alias='skip-install')
+
+
+class HatchEnvironmentConfiguration(RootModel[list[Environment]]):
+    """
+    A root model that parses the top-level dictionary returned by the `hatch env show --json` command
+    into a list of Environment models.
+    """
+
+    @model_validator(mode='before')
+    @classmethod
+    def transform_dict_to_list_with_name(cls, data: object) -> object:
+        if isinstance(data, list):
+            return data
+
+        if isinstance(data, dict):
+            return [Environment(**value, name=key) for key, value in data.items()]
+
+        raise ValueError(f'Invalid data type: {type(data)}. Expected a list or a dictionary.')
+
+
+class EnvironmentFilter(Protocol):
+    def __call__(self, environment: Environment) -> bool: ...
+
+
+@overload
+def env_show(platform: Platform, integration: Integration, as_json: Literal[True]) -> dict[str, Any]: ...
+
+
+@overload
+def env_show(platform: Platform, integration: Integration) -> dict[str, Any]: ...
+
+
+@overload
+def env_show(platform: Platform, integration: Integration, as_json: Literal[False]) -> str: ...
+
+
+def env_show(platform: Platform, integration: Integration, as_json: bool = True) -> dict[str, Any] | str:
+    import json
+    import sys
+
+    with integration.path.as_cwd():
+        command = [sys.executable, '-m', 'hatch', '--no-color', '--no-interactive', 'env', 'show']
+        if as_json:
+            command.append('--json')
+
+        env_data_output = platform.check_command_output(command)
+
+        try:
+            if as_json:
+                return json.loads(env_data_output)
+            return env_data_output
+        except json.JSONDecodeError as error:
+            raise HatchCommandError(
+                f'Failed to parse environments for {integration.name!r}: {env_data_output!r}'
+            ) from error
+
+
+def list_environment_names(
+    platform: Platform, integration: Integration, filters: Collection[EnvironmentFilter], match_all: bool = True
+) -> list[str]:
+    """
+    List the names of the environments that match the given filters.
+
+    If `match_all` is True, all filters must match. If False, any filter can match.
+    """
+    hatch_output = env_show(platform, integration)
+    matching_rule = all if match_all else any
+
+    return [
+        env.name
+        for env in HatchEnvironmentConfiguration.model_validate(hatch_output).root
+        if matching_rule(filter(env) for filter in filters)
+    ]
+
+
+def get_hatch_env_vars(*, verbosity: int) -> dict[str, str]:
+    env_vars = {}
+
+    if verbosity > 0:
+        env_vars['HATCH_VERBOSE'] = str(verbosity)
+    elif verbosity < 0:
+        env_vars['HATCH_QUIET'] = str(abs(verbosity))
+
+    return env_vars

ddev/src/ddev/utils/hatch.py

@@ -16,6 +16,21 @@
 from tests.helpers.mocks import MockPopen
 from tests.helpers.runner import CliRunner
 
+BASE_ENV_CONFIG = {
+    'type': 'virtual',
+    'dependencies': [],
+    'test-env': True,
+    'e2e-env': True,
+    'benchmark-env': True,
+    'latest-env': True,
+    'python': '3.12',
+    'scripts': {},
+    'platforms': [],
+    'pre-install-commands': [],
+    'post-install-commands': [],
+    'skip-install': False,
+}
+
 
 def setup(
     mocker: MockerFixture,
@@ -96,7 +111,7 @@ def test_env_vars_repo(
     predicate: Callable[[Result], bool],
     mock_commands: tuple[MockType, MockType, MockType],
 ):
-    setup(mocker, write_result_file, hatch_json_output={'py3.12': {'e2e-env': e2e_env}})
+    setup(mocker, write_result_file, hatch_json_output={'py3.12': {**BASE_ENV_CONFIG, 'e2e-env': e2e_env}})
     mocker.patch.object(EnvData, 'read_metadata', return_value={})
 
     result = ddev('env', 'test', 'postgres', 'py3.12')
@@ -119,7 +134,11 @@ def test_environment_runs_for_enabled_environments(
     setup(
         mocker,
         write_result_file,
-        hatch_json_output={'py3.12': {'e2e-env': True}, 'py3.13': {'e2e-env': False}, 'py3.13-v1': {'e2e-env': True}},
+        hatch_json_output={
+            'py3.12': BASE_ENV_CONFIG,
+            'py3.13': {**BASE_ENV_CONFIG, 'e2e-env': False},
+            'py3.13-v1': BASE_ENV_CONFIG,
+        },
     )
     with mocker.patch.object(EnvData, 'read_metadata', return_value={}):
         result = ddev('env', 'test', 'postgres', environment)
@@ -145,7 +164,7 @@ def test_runningin_ci_triggers_all_environments_when_not_supplied(
     mocker: MockerFixture,
     mock_commands: tuple[MockType, MockType, MockType],
 ):
-    setup(mocker, write_result_file, hatch_json_output={'py3.12': {'e2e-env': True}, 'py3.13': {'e2e-env': True}})
+    setup(mocker, write_result_file, hatch_json_output={'py3.12': BASE_ENV_CONFIG, 'py3.13': BASE_ENV_CONFIG})
     mocker.patch('ddev.utils.ci.running_in_ci', return_value=True)
 
     with mocker.patch.object(EnvData, 'read_metadata', return_value={}):
@@ -161,7 +180,7 @@ def test_run_only_active_environments_when_not_running_in_ci_and_active_environm
     mocker: MockerFixture,
     mock_commands: tuple[MockType, MockType, MockType],
 ):
-    setup(mocker, write_result_file, hatch_json_output={'py3.12': {'e2e-env': True}, 'py3.13': {'e2e-env': True}})
+    setup(mocker, write_result_file, hatch_json_output={'py3.12': BASE_ENV_CONFIG, 'py3.13': BASE_ENV_CONFIG})
     mocker.patch('ddev.utils.ci.running_in_ci', return_value=False)
 
     with (