package_managers: rpm: Allow specifying DNF options via CLI input JSON

Introduce a new input JSON attribute 'options' for the PoC rpm package
manager allowing consumers to pass arbitrary DNF repo configuration
options that would be applied later through the inject-files command.

The general schema for the input JSON is as follows:

{
    type: <package_manager>
    path: <relative_path>
    options: {
        <namespace>: {
            <key1>..<keyN>: Union[str, Any]<val1>
        }
    }
}

which translated to DNF repo options for RPM might then look like:

{
    type: rpm
    path: .
    options: {
        dnf: {
             repoid1: {gpgcheck: 0, deltarpm: 1},
             repoid2: {fastestmirro: 1, sslverify: 0}
        }
    }
}

This implementation is trying to be generic enough to be potentially
later extended to other package managers as well which also means that
due to the fairly complex (multi-level nested dictionary) input
a new intermediary model _DNFOptions was needed to be introduced in
this patch, but most importantly a custom 'before' model validator for
this model had to be implemented to generate stable and deterministic
errors which could be reliably unit-tested (default pydantic error
messages for such complex structures are madness).

Worth noting would be that unless options were specified, they would
not appear in the build-config.json --> exclude_none=True

Signed-off-by: Erik Skultety <[email protected]>

cachi2/core/models/input.py

@@ -1,5 +1,17 @@
 from pathlib import Path
-from typing import TYPE_CHECKING, Annotated, Callable, ClassVar, Literal, Optional, TypeVar, Union
+from typing import (
+    TYPE_CHECKING,
+    Annotated,
+    Any,
+    Callable,
+    ClassVar,
+    Dict,
+    List,
+    Literal,
+    Optional,
+    TypeVar,
+    Union,
+)
 
 import pydantic
 
@@ -65,6 +77,60 @@ def _path_is_relative(cls, path: Path) -> Path:
         return check_sane_relpath(path)
 
 
+class _DNFOptions(pydantic.BaseModel, extra="forbid"):
+    """DNF options model.
+
+    DNF options can be provided via 2 'streams':
+        1) global /etc/dnf/dnf.conf OR
+        2) /etc/yum.repos.d/.repo files
+
+    Config options are specified via INI format based on sections. There are 2 types of sections:
+        1) global 'main' - either global repo options or DNF control-only options
+            - NOTE: there must always ever be a single "main" section
+
+        2) <repoid> sections - options tied specifically to a given defined repo
+
+    [1] https://man7.org/linux/man-pages/man5/dnf.conf.5.html
+    """
+
+    # Don't model all known DNF options for validation purposes - it's user's responsibility!
+    dnf: Dict[Union[Literal["main"], str], Dict[str, Any]]
+
+    @pydantic.model_validator(mode="before")
+    def _validate_dnf_options(cls, data: Any, info: pydantic.ValidationInfo) -> Optional[Dict]:
+        """Fail if the user passes unexpected configuration options namespace."""
+
+        def _raise_unexpected_type(repr_: str, *prefixes: str) -> None:
+            loc = ".".join(prefixes + (repr_,))
+            raise ValueError(f"Unexpected data type for '{loc}' in input JSON: expected 'dict'")
+
+        prefixes: List[str] = ["options"]
+
+        if not data:
+            return None
+
+        if not isinstance(data, dict):
+            _raise_unexpected_type(data, *prefixes)
+
+        if "dnf" not in data:
+            raise ValueError(f"Missing required namespace attribute in '{data}': 'dnf'")
+
+        if diff := set(cls.model_fields) - set(data.keys()):
+            raise ValueError(f"Extra attributes passed in '{data}': {diff}")
+
+        prefixes.append("dnf")
+        options_scope = data["dnf"]
+        if not isinstance(options_scope, dict):
+            _raise_unexpected_type(options_scope, *prefixes)
+
+        for repo, repo_options in options_scope.items():
+            prefixes.append(repo)
+            if not isinstance(repo_options, dict):
+                _raise_unexpected_type(repo_options, *prefixes)
+
+        return data
+
+
 class GomodPackageInput(_PackageInputBase):
     """Accepted input for a gomod package."""
 
@@ -104,6 +170,7 @@ class RpmPackageInput(_PackageInputBase):
     """Accepted input for a rpm package."""
 
     type: Literal["rpm"]
+    options: Optional[Union[_DNFOptions]] = None
 
 
 class YarnPackageInput(_PackageInputBase):

cachi2/core/models/output.py

@@ -1,7 +1,7 @@
 import logging
 import string
 from pathlib import Path
-from typing import Dict, Literal, Optional, Set
+from typing import Any, Dict, Literal, Optional, Set
 
 import pydantic
 
@@ -133,6 +133,7 @@ class BuildConfig(pydantic.BaseModel):
 
     environment_variables: list[EnvironmentVariable] = []
     project_files: list[ProjectFile] = []
+    options: Optional[Dict[str, Any]] = None
 
     @pydantic.field_validator("environment_variables")
     def _unique_env_vars(cls, env_vars: list[EnvironmentVariable]) -> list[EnvironmentVariable]:
@@ -170,6 +171,7 @@ def from_obj_list(
         components: list[Component],
         environment_variables: Optional[list[EnvironmentVariable]] = None,
         project_files: Optional[list[ProjectFile]] = None,
+        options: Optional[Dict[str, Any]] = None,
     ) -> "RequestOutput":
         """Create a RequestOutput from components, environment variables and project files."""
         if environment_variables is None:
@@ -183,5 +185,6 @@ def from_obj_list(
             build_config=BuildConfig(
                 environment_variables=environment_variables,
                 project_files=project_files,
+                options=options,
             ),
         )

cachi2/core/package_managers/rpm/main.py

@@ -5,7 +5,7 @@
 from configparser import ConfigParser
 from os import PathLike
 from pathlib import Path
-from typing import Any, Union, no_type_check
+from typing import Any, Dict, Optional, Union, no_type_check
 from urllib.parse import quote
 
 import yaml
@@ -74,14 +74,39 @@ def write(self, fileobject, space_around_delimiters=True) -> None:
 def fetch_rpm_source(request: Request) -> RequestOutput:
     """Process all the rpm source directories in a request."""
     components: list[Component] = []
+    options: Dict[str, Any] = {}
+    noptions = 0
+
     for package in request.rpm_packages:
         path = request.source_dir.join_within_root(package.path)
         components.extend(_resolve_rpm_project(path, request.output_dir))
 
+        # FIXME: this is only ever good enough for a PoC, but needs to be handled properly in the
+        # future.
+        # It's unlikely that a project would be split into multiple packages, i.e. supplying
+        # multiple rpms.lock.yaml files. We'd end up generating a single .repo file anyway,
+        # however, although trying to pass conflicting options to DNF for identical repoids via the
+        # input JSON doesn't make much sense from practical perspective (i.e. there's going to be a
+        # single .repo file) the CLI technically allows it in the input JSON.
+        # We're deliberately taking the easy route here by only assuming the "last" set of options
+        # we found in the input JSON instead of doing a deep merge of all the nested dicts.
+        # Nevertheless, we'll at least emit a warning at the end so that the user is informed
+        if package.options and package.options.dnf:
+            options = package.options.model_dump()
+            noptions += 1
+
+    if noptions > 1:
+        log.warning(
+            "Multiple sets of DNF options detected on the input: "
+            "Only one input RPM project package can specify extra DNF options, "
+            "the last one seen will take effect"
+        )
+
     return RequestOutput.from_obj_list(
         components=components,
         environment_variables=[],
         project_files=[],
+        options={"rpm": options} if options else None,
     )
 
 
@@ -272,7 +297,7 @@ def inject_files_post(from_output_dir: Path, for_output_dir: Path, **kwargs: Any
     """Run extra tasks for the RPM package manager (callback method) within `inject-files` cmd."""
     if Path.exists(from_output_dir.joinpath(DEFAULT_PACKAGE_DIR)):
         _generate_repos(from_output_dir)
-        _generate_repofiles(from_output_dir, for_output_dir)
+        _generate_repofiles(from_output_dir, for_output_dir, kwargs.get("options"))
 
 
 def _generate_repos(from_output_dir: Path) -> None:
@@ -298,7 +323,9 @@ def _createrepo(reponame: str, repodir: Path) -> None:
     log.debug(stdout)
 
 
-def _generate_repofiles(from_output_dir: Path, for_output_dir: Path) -> None:
+def _generate_repofiles(
+    from_output_dir: Path, for_output_dir: Path, options: Optional[Dict] = None
+) -> None:
     """
     Generate templates of repofiles for all arches.
 
@@ -308,6 +335,13 @@ def _generate_repofiles(from_output_dir: Path, for_output_dir: Path) -> None:
     Repofiles are not directly created from the templates in this method - templates are stored
     in the project file.
     """
+    dnf_options = None
+    dnf_options_repos = None
+
+    if options:
+        dnf_options = options.get("rpm", {}).get("dnf", {})
+        dnf_options_repos = dnf_options.keys()
+
     package_dir = from_output_dir.joinpath(DEFAULT_PACKAGE_DIR)
     for arch in package_dir.iterdir():
         if not arch.is_dir():
@@ -321,6 +355,12 @@ def _generate_repofiles(from_output_dir: Path, for_output_dir: Path) -> None:
             repoid = entry.name
             repofile[repoid] = {}
 
+            # TODO: purposefully ignoring the fact that options might be passed within the "main"
+            # context of DNF options which would mean we'd have to generate a dnf.conf since such
+            # options are global, skipping that for now
+            if dnf_options and dnf_options_repos and repoid in dnf_options_repos:
+                repofile[repoid] = dnf_options[repoid]
+
             localpath = for_output_dir.joinpath(DEFAULT_PACKAGE_DIR, arch.name, repoid)
             repofile[repoid]["baseurl"] = f"file://{localpath}"
 

cachi2/core/resolver.py

@@ -83,6 +83,7 @@ def _merge_outputs(outputs: Iterable[RequestOutput]) -> RequestOutput:
         components=components,
         environment_variables=env_vars,
         project_files=project_files,
+        options=output.build_config.options if output.build_config.options else None,
     )
 
 

cachi2/interface/cli.py

@@ -260,7 +260,7 @@ def combine_option_and_json_flags(json_flags: list[Flag]) -> list[str]:
 
     request.output_dir.path.mkdir(parents=True, exist_ok=True)
     request.output_dir.join_within_root(".build-config.json").path.write_text(
-        request_output.build_config.model_dump_json(indent=2)
+        request_output.build_config.model_dump_json(indent=2, exclude_none=True)
     )
 
     sbom = request_output.generate_sbom()
@@ -339,7 +339,11 @@ def inject_files(
         content = project_file.resolve_content(output_dir=for_output_dir)
         project_file.abspath.write_text(content)
 
-    inject_files_post(from_output_dir=from_output_dir, for_output_dir=for_output_dir)
+    inject_files_post(
+        from_output_dir=from_output_dir,
+        for_output_dir=for_output_dir,
+        options=fetch_deps_output.options,
+    )
 
 
 def _get_build_config(output_dir: Path) -> BuildConfig:

tests/unit/models/test_input.py

@@ -12,6 +12,7 @@
     PackageInput,
     PipPackageInput,
     Request,
+    RpmPackageInput,
     parse_user_input,
 )
 from cachi2.core.rooted_path import RootedPath
@@ -60,6 +61,35 @@ class TestPackageInput:
                     "allow_binary": True,
                 },
             ),
+            (
+                {"type": "rpm"},
+                {
+                    "type": "rpm",
+                    "path": Path("."),
+                    "options": None,
+                },
+            ),
+            (
+                {
+                    "type": "rpm",
+                    "options": {
+                        "dnf": {
+                            "main": {"best": True, "debuglevel": 2},
+                            "foorepo": {"arch": "x86_64", "enabled": True},
+                        }
+                    },
+                },
+                {
+                    "type": "rpm",
+                    "path": Path("."),
+                    "options": {
+                        "dnf": {
+                            "main": {"best": True, "debuglevel": 2},
+                            "foorepo": {"arch": "x86_64", "enabled": True},
+                        }
+                    },
+                },
+            ),
         ],
     )
     def test_valid_packages(self, input_data: dict[str, Any], expect_data: dict[str, Any]) -> None:
@@ -113,6 +143,26 @@ def test_valid_packages(self, input_data: dict[str, Any], expect_data: dict[str,
                 r"none is not an allowed value",
                 id="pip_no_requirements_build_files",
             ),
+            pytest.param(
+                {"type": "rpm", "options": "bad_type"},
+                r"Unexpected data type for 'options.bad_type' in input JSON",
+                id="rpm_bad_options_type",
+            ),
+            pytest.param(
+                {"type": "rpm", "options": {"unknown": "foo"}},
+                r"Missing required namespace attribute in '{\'unknown\': \'foo\'}': 'dnf'",
+                id="rpm_missing_required_namespace_dnf",
+            ),
+            pytest.param(
+                {"type": "rpm", "options": {"dnf": "bad_type"}},
+                r"Unexpected data type for 'options.dnf.bad_type' in input JSON",
+                id="rpm_bad_type_for_dnf_namespace",
+            ),
+            pytest.param(
+                {"type": "rpm", "options": {"dnf": {"repo": "bad_type"}}},
+                r"Unexpected data type for 'options.dnf.repo.bad_type' in input JSON",
+                id="rpm_bad_type_for_dnf_options",
+            ),
         ],
     )
     def test_invalid_packages(self, input_data: dict[str, Any], expect_error: str) -> None:
@@ -165,11 +215,12 @@ def test_valid_request(self, tmp_path: Path) -> None:
         assert isinstance(request.output_dir, RootedPath)
 
     def test_packages_properties(self, tmp_path: Path) -> None:
-        packages = [{"type": "gomod"}, {"type": "npm"}, {"type": "pip"}]
+        packages = [{"type": "gomod"}, {"type": "npm"}, {"type": "pip"}, {"type": "rpm"}]
         request = Request(source_dir=tmp_path, output_dir=tmp_path, packages=packages)
         assert request.gomod_packages == [GomodPackageInput(type="gomod")]
         assert request.npm_packages == [NpmPackageInput(type="npm")]
         assert request.pip_packages == [PipPackageInput(type="pip")]
+        assert request.rpm_packages == [RpmPackageInput(type="rpm")]
 
     @pytest.mark.parametrize("which_path", ["source_dir", "output_dir"])
     def test_path_not_absolute(self, which_path: str) -> None: