Added the inmanta module download command, that allows downloading Inmanta modules in their source format from a Python package repository. (Issue #7820, PR #9397)

# Description

Added the `inmanta module download` command, that allows downloading Inmanta modules in their source format from a Python package repository.

closes #7820

# Self Check:

- [x] Attached issue to pull request
- [x] Changelog entry
- [x] Type annotations are present
- [x] Code is clear and sufficiently documented
- [x] No (preventable) type errors (check using make mypy or make mypy-diff)
- [x] Sufficient test cases (reproduces the bug/tests the requested feature)
- [x] Correct, in line with design
- [x] End user documentation is included or an issue is created for end-user documentation
- [ ] ~~If this PR fixes a race condition in the test suite, also push the fix to the relevant stable branche(s) (see [test-fixes](https://internal.inmanta.com/development/core/tasks/build-master.html#test-fixes) for more info)~~

Author: arnaudsjs

changelogs/unreleased/add-inmanta-module-download-cmd.yml

@@ -0,0 +1,8 @@
+---
+description: "Added the `inmanta module download` command, that allows downloading Inmanta modules in their source format from a Python package repository."
+issue-nr: 7820
+issue-repo: inmanta-core
+change-type: minor
+destination-branches: [master, iso8]
+sections:
+  feature: "{{description}}"

src/inmanta/env.py

@@ -359,6 +359,21 @@ def compose_uninstall_command(cls, python_path: str, pkg_names: Sequence[str]) -
         """
         return [python_path, "-m", "pip", "uninstall", "-y", *pkg_names]
 
+    @classmethod
+    def compose_download_command(
+        cls, python_path: str, pkg_requirement: inmanta.util.CanonicalRequirement, no_deps: bool, no_binary: str | None
+    ) -> list[str]:
+        """
+        Returns the pip command to download a python package.
+        """
+        command = [python_path, "-m", "pip", "download"]
+        if no_deps:
+            command.append("--no-deps")
+        if no_binary:
+            command.extend(["--no-binary", no_binary])
+        command.append(str(pkg_requirement))
+        return command
+
     @classmethod
     def compose_list_command(
         cls, python_path: str, format: Optional[PipListFormat] = None, only_editable: bool = False
@@ -828,6 +843,15 @@ def get_installed_packages(self, only_editable: bool = False) -> dict[Normalized
         output = CommandRunner(LOGGER_PIP).run_command_and_log_output(cmd, stderr=subprocess.DEVNULL, env=os.environ.copy())
         return {canonicalize_name(r["name"]): packaging.version.Version(r["version"]) for r in json.loads(output)}
 
+    def download_python_package(self, pkg_requirement: inmanta.util.CanonicalRequirement, output_directory: str) -> None:
+        """
+        Download the python package that satisfies the constraint pkg_requirement as a source distributin package.
+        """
+        cmd = PipCommandBuilder.compose_download_command(
+            python_path=self.python_path, pkg_requirement=pkg_requirement, no_deps=True, no_binary=pkg_requirement.name
+        )
+        CommandRunner(LOGGER_PIP).run_command_and_log_output(cmd, env=os.environ.copy(), cwd=output_directory)
+
     def install_for_config(
         self,
         requirements: Sequence[inmanta.util.CanonicalRequirement],

src/inmanta/moduletool.py

@@ -20,6 +20,7 @@
 import configparser
 import datetime
 import enum
+import gzip
 import inspect
 import itertools
 import logging
@@ -53,7 +54,7 @@
 import packaging.requirements
 import toml
 from build.env import DefaultIsolatedEnv
-from inmanta import const
+from inmanta import const, env
 from inmanta.command import CLIException, ShowUsageException
 from inmanta.const import CF_CACHE_DIR
 from inmanta.module import (
@@ -626,6 +627,71 @@ def modules_parser_config(cls, parser: ArgumentParser, parent_parsers: abc.Seque
             "this message will also be used as the commit message.",
         )
         release.add_argument("-a", "--all", dest="commit_all", help="Use commit -a", action="store_true")
+        download = subparser.add_parser(
+            "download",
+            help="Download the source distribution of an Inmanta module from a Python package repository,"
+            " extract it and convert it to its source format.",
+            parents=parent_parsers,
+        )
+        download.add_argument(
+            "module_req",
+            help="The name of the module, optionally with a version constraint.",
+        )
+        download.add_argument(
+            "--install",
+            dest="install",
+            help="Install the downloaded module in editable mode into the active Python environment.",
+            action="store_true",
+        )
+        download.add_argument(
+            "-d",
+            "--directory",
+            dest="directory",
+            help="Download the module in this directory instead of the current working directory.",
+        )
+
+    def download(self, module_req: str, install: bool, directory: str | None) -> None:
+        if directory is None:
+            directory = os.getcwd()
+        module_requirement = InmantaModuleRequirement.parse(module_req)
+        module_name = module_requirement.name
+        with tempfile.TemporaryDirectory() as path_tmp_dir:
+            # Download the python package
+            download_dir = os.path.join(path_tmp_dir, "download")
+            os.mkdir(download_dir)
+            env.process_env.download_python_package(
+                pkg_requirement=module_requirement.get_python_package_requirement(),
+                output_directory=download_dir,
+            )
+            files_download_dir = os.listdir(download_dir)
+            assert len(files_download_dir) == 1
+            path_python_source_package = os.path.join(download_dir, files_download_dir[0])
+            # Extract the package
+            extract_dir = os.path.join(path_tmp_dir, "extract")
+            os.mkdir(extract_dir)
+            with gzip.open(filename=path_python_source_package, mode="rb") as tar_file_obj:
+                with tarfile.TarFile(mode="r", fileobj=tar_file_obj) as tar:
+                    tar.extractall(path=extract_dir, filter="data")
+            files_extract_dir = os.listdir(extract_dir)
+            assert len(files_extract_dir) == 1
+            path_extracted_pkg = os.path.join(extract_dir, files_extract_dir[0])
+            # Convert to source format
+            try:
+                # Remove this file as it will be replace by the one present in the inmanta_plugins/<mod-name> directory.
+                os.remove(os.path.join(path_extracted_pkg, "setup.cfg"))
+            except FileNotFoundError:
+                pass
+            files_and_dirs_to_move = ["model", "templates", "files", "setup.cfg"]
+            for file_or_dir in files_and_dirs_to_move:
+                fq_path = os.path.join(path_extracted_pkg, "inmanta_plugins", module_name, file_or_dir)
+                if os.path.exists(fq_path):
+                    shutil.move(src=fq_path, dst=path_extracted_pkg)
+            # Move to desired output directory
+            destination_dir = os.path.join(directory, module_name)
+            shutil.copytree(src=path_extracted_pkg, dst=destination_dir)
+            # Install in editable mode if requested
+            if install:
+                env.process_env.install_from_source(paths=[env.LocalPackagePath(path=destination_dir, editable=True)])
 
     def add(self, module_req: str, v2: bool = True, override: bool = False) -> None:
         """

tests/data/modules_v2/elaboratev2module/tests/test_test.py

@@ -0,0 +1,2 @@
+def test_test():
+    pass

tests/moduletool/test_download.py

@@ -0,0 +1,192 @@
+"""
+Copyright 2025 Inmanta
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+Contact: [email protected]
+"""
+
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+
+import pytest
+
+import packaging.utils
+from inmanta import env, module, moduletool
+from libpip2pi.commands import dir2pi
+from packaging import version
+from utils import module_from_template
+
+
+@pytest.fixture(scope="session")
+def pip_index(modules_v2_dir: str) -> str:
+    """
+    Returns the path to a pip index that contains several version for the same python package.
+    """
+    with tempfile.TemporaryDirectory() as root_dir:
+        source_dir = os.path.join(root_dir, "source")
+        build_dir = os.path.join(root_dir, "build")
+        index_dir = os.path.join(build_dir, "simple")
+
+        modules_to_build = [
+            ("elaboratev2module", version.Version("1.2.3"), False),
+            ("elaboratev2module", version.Version("2.3.4"), False),
+            ("elaboratev2module", version.Version("2.3.5"), True),
+            ("minimalv2module", version.Version("1.1.1"), False),
+        ]
+
+        for module_name, mod_version, is_prerelease in modules_to_build:
+            template_dir = os.path.join(modules_v2_dir, module_name)
+            module_dir = os.path.join(source_dir, module_name)
+            module_from_template(
+                source_dir=template_dir,
+                dest_dir=module_dir,
+                new_version=mod_version,
+            )
+            moduletool.ModuleTool().build(
+                path=module_dir,
+                output_dir=build_dir,
+                dev_build=is_prerelease,
+                wheel=True,
+                sdist=True,
+            )
+            shutil.rmtree(module_dir)
+
+        # The setuptools and wheel packages are required by `pip download`
+        subprocess.check_call([sys.executable, "-m", "pip", "download", "setuptools", "wheel"], cwd=build_dir)
+        dir2pi(argv=["dir2pi", build_dir])
+        yield index_dir
+
+
+def execute_inmanta_download(module_req: str, install: bool, download_dir: str | None) -> str:
+    """
+    Executes the `inmanta module download` command and returns the ModuleV2 object for the
+    downloaded module.
+    """
+    if download_dir is None:
+        download_dir = os.getcwd()
+    module_name: str = module.InmantaModuleRequirement.parse(module_req).name
+    assert not os.listdir(download_dir)
+    m = moduletool.ModuleTool()
+    m.download(module_req=module_req, install=install, directory=download_dir)
+    files = os.listdir(download_dir)
+    assert len(files) == 1
+    assert files[0] == module_name
+    return module.ModuleV2(project=None, path=os.path.join(download_dir, module_name))
+
+
+def assert_files_in_module(
+    module_dir: str, module_name: str, has_files_dir: bool, has_templates_dir: bool, has_tests_dir: bool
+) -> None:
+    """
+    Verify that the directory structure of the extracted python package is correct.
+
+    :param has_files_dir: True iff the given module has a files directory.
+    :param has_templates_dir: True iff the given module has a templates directory.
+    :param has_tests_dir: True iff the given module has a tests directory.
+    """
+    assert os.path.exists(os.path.join(module_dir, "inmanta_plugins"))
+    for file_or_dir, must_exist_in_root_dir in [
+        ("tests", has_tests_dir),
+        ("files", has_files_dir),
+        ("templates", has_templates_dir),
+        ("setup.cfg", True),
+    ]:
+        assert not os.path.exists(os.path.join(module_dir, "inmanta_plugins", module_name, file_or_dir))
+        assert os.path.exists(os.path.join(module_dir, file_or_dir)) == must_exist_in_root_dir
+
+
+def test_download_module(tmpdir, monkeypatch, tmpvenv_active, pip_index: str):
+    """
+    Test the `inmanta module download` command with a version constraint.
+    """
+    monkeypatch.setenv("PIP_INDEX_URL", pip_index)
+    monkeypatch.setenv("PIP_PRE", "false")
+    download_dir = os.path.join(tmpdir, "download")
+
+    # Test download package without constraint
+    os.mkdir(download_dir)
+    mod: module.ModuleV2 = execute_inmanta_download(module_req="elaboratev2module", install=False, download_dir=download_dir)
+    assert_files_in_module(
+        module_dir=mod.path, module_name="elaboratev2module", has_files_dir=True, has_templates_dir=True, has_tests_dir=True
+    )
+    assert mod.version == version.Version("2.3.4")
+    shutil.rmtree(download_dir)
+
+    # Test download package with constraint
+    os.mkdir(download_dir)
+    mod: module.ModuleV2 = execute_inmanta_download(
+        module_req="elaboratev2module~=1.2.0", install=False, download_dir=download_dir
+    )
+    assert_files_in_module(
+        module_dir=mod.path, module_name="elaboratev2module", has_files_dir=True, has_templates_dir=True, has_tests_dir=True
+    )
+    assert mod.version == version.Version("1.2.3")
+    shutil.rmtree(download_dir)
+
+    # Test download package with --pre
+    os.mkdir(download_dir)
+    monkeypatch.setenv("PIP_PRE", "true")
+    mod: module.ModuleV2 = execute_inmanta_download(module_req="elaboratev2module", install=False, download_dir=download_dir)
+    assert_files_in_module(
+        module_dir=mod.path, module_name="elaboratev2module", has_files_dir=True, has_templates_dir=True, has_tests_dir=True
+    )
+    assert mod.version.base_version == "2.3.5"
+    assert mod.version.is_prerelease
+    shutil.rmtree(download_dir)
+
+    # Test downloading a package that doesn't have any of the optional directories
+    # (e.g. files, templates, tests).
+    monkeypatch.setenv("PIP_PRE", "false")
+    os.mkdir(download_dir)
+    mod: module.ModuleV2 = execute_inmanta_download(module_req="minimalv2module", install=False, download_dir=download_dir)
+    assert_files_in_module(
+        module_dir=mod.path, module_name="minimalv2module", has_files_dir=False, has_templates_dir=False, has_tests_dir=False
+    )
+    assert mod.version == version.Version("1.1.1")
+
+
+def test_download_cwd(tmpdir, monkeypatch, tmpvenv_active, pip_index: str):
+    """
+    Test the `inmanta module download` command when downloading to the current working directory.
+    """
+    monkeypatch.setenv("PIP_INDEX_URL", pip_index)
+    monkeypatch.setenv("PIP_PRE", "false")
+    download_dir = os.path.join(tmpdir, "download")
+    os.mkdir(download_dir)
+    monkeypatch.chdir(download_dir)
+    mod: module.ModuleV2 = execute_inmanta_download(module_req="elaboratev2module", install=False, download_dir=None)
+    assert_files_in_module(
+        module_dir=mod.path, module_name="elaboratev2module", has_files_dir=True, has_templates_dir=True, has_tests_dir=True
+    )
+
+
+def test_download_install(tmpdir, monkeypatch, tmpvenv_active, pip_index: str):
+    """
+    Test the install option of the `inmanta module download` command.
+    """
+    monkeypatch.setenv("PIP_INDEX_URL", pip_index)
+    monkeypatch.setenv("PIP_PRE", "false")
+    download_dir = os.path.join(tmpdir, "download")
+    os.mkdir(download_dir)
+    pkg_name = "inmanta-module-minimalv2module"
+    pkgs_installed_in_editable_mode: dict[packaging.utils.NormalizedName, version.Version]
+    pkgs_installed_in_editable_mode = env.process_env.get_installed_packages(only_editable=True)
+    assert pkg_name not in pkgs_installed_in_editable_mode
+    execute_inmanta_download(module_req="minimalv2module", install=True, download_dir=download_dir)
+    pkgs_installed_in_editable_mode = env.process_env.get_installed_packages(only_editable=True)
+    assert pkg_name in pkgs_installed_in_editable_mode
+    assert pkgs_installed_in_editable_mode[pkg_name] == version.Version("1.1.1")