package_managers: rpm: Generate the .repo file through ConfigParser

Repo files follow the INI syntax for which we can use ConfigParser.
The motivation behind using a library abstraction to generate the INI
repo files stems from the fact that we already have some options
rendered conditionally with a fair chance of extending this in the
future with more custom user-supplied DNF repo options which means even
more conditionals which would stretch the capabilities of a plain
triple string (or a template for that matter) a bit too far.
Therefore, this patch proposes using ConfigParser instance for creating
and dumping the repo files in the inject_files post action.

Note that to make things a tiny bit more ergonomic, we subclass the
ConfigParser class to define a couple utility methods that can tell us
whether the ConfigParser instance is effectively empty (because an
empty instance of ConfigParser still evaluates to True) and thus can
be dumped to a file as well as some extra handling of default options,
e.g. defaulting to 'gpgcheck=1' explicitly in the generated .repo file.

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

cachi2/core/package_managers/rpm/main.py

@@ -2,9 +2,10 @@
 import hashlib
 import logging
 import shlex
+from configparser import ConfigParser
 from os import PathLike
 from pathlib import Path
-from typing import Any, Union
+from typing import Any, Union, no_type_check
 from urllib.parse import quote
 
 import yaml
@@ -30,6 +31,46 @@
 READ_CHUNK = 1048576
 
 
+class _Repofile(ConfigParser):
+    def _apply_defaults(self) -> None:
+        defaults = self.defaults()
+
+        if not defaults:
+            return
+
+        # apply defaults per-section
+        for s in self.sections():
+            section = self[s]
+
+            # ConfigParser's section is of the Mapping abstract type rather than a dictionary.
+            # That means that when queried for options the results will include the defaults.
+            # However, those defaults are referenced from a different map which on its own is good
+            # until one tries to dump the ConfigParser instance to a file which will create a
+            # dedicated section for the defaults -> [DEFAULTS] which we don't want.
+            # This hackish line will make sure that by converting both the defaults and existing
+            # section options to dictionaries and merging those back to the section object will
+            # bake the defaults into each section rather than referencing them from a different
+            # map, allowing us to rid of the defaults right before we dump the contents to a file
+            section.update(dict(defaults) | dict(section))
+
+        # defaults applied, clear the default section to prevent it from being formatted to the
+        # output as
+        #  [DEFAULTS]
+        #  default1=val'
+        self[self.default_section].clear()
+
+    @property
+    def empty(self) -> bool:
+        return not bool(self.sections())
+
+    # typeshed uses a private protocol type for the file-like object:
+    # https://github.com/python/typeshed/blob/0445d74489d7a0b04a8c64a5a349ada1408718a9/stdlib/configparser.pyi#L197
+    @no_type_check
+    def write(self, fileobject, space_around_delimiters=True) -> None:
+        self._apply_defaults()
+        return super().write(fileobject, space_around_delimiters)
+
+
 def fetch_rpm_source(request: Request) -> RequestOutput:
     """Process all the rpm source directories in a request."""
     components: list[Component] = []
@@ -272,26 +313,27 @@ def _generate_repofiles(from_output_dir: Path, for_output_dir: Path) -> None:
         if not arch.is_dir():
             continue
         log.debug(f"Preparing repofile content for arch '{arch.name}'")
-        content = ""
+        repofile = _Repofile(defaults={"gpgcheck": "1"})
         for entry in sorted(arch.iterdir()):
             if not entry.is_dir() or entry.name == "repos.d":
                 continue
             repoid = entry.name
+            repofile[repoid] = {}
+
             localpath = for_output_dir.joinpath(DEFAULT_PACKAGE_DIR, arch.name, repoid)
-            content += f"[{repoid}]\n"
-            content += f"baseurl=file://{localpath}\n"
-            content += "gpgcheck=1\n"
+            repofile[repoid]["baseurl"] = f"file://{localpath}"
+
             # repoid directory matches the internal repoid
             if repoid.startswith("cachi2-"):
-                content += "Packages unaffiliated with an official repository\n"
+                repofile[repoid]["name"] = "Packages unaffiliated with an official repository"
 
-        if content:
+        if not repofile.empty:
             repo_file_path = arch.joinpath("repos.d", "cachi2.repo")
             if repo_file_path.exists():
                 log.warning(f"Overwriting {repo_file_path}")
             else:
                 Path.mkdir(arch.joinpath("repos.d"), parents=True, exist_ok=True)
                 log.info(f"Creating {repo_file_path}")
 
-            with open(repo_file_path, "w") as repo_file:
-                repo_file.write(content)
+            with open(repo_file_path, "w") as f:
+                repofile.write(f)

tests/unit/package_managers/test_rpm.py

@@ -1,4 +1,6 @@
+from configparser import ConfigParser
 from pathlib import Path
+from typing import Any, Dict, Optional
 from unittest import mock
 from urllib.parse import quote
 
@@ -16,6 +18,7 @@
     _generate_repofiles,
     _generate_repos,
     _generate_sbom_components,
+    _Repofile,
     _resolve_rpm_project,
     _verify_downloaded,
 )
@@ -310,21 +313,38 @@ def test_generate_repos(mock_createrepo: mock.Mock, rooted_tmp_path: RootedPath)
     mock_createrepo.assert_called_once_with("repo1", arch_dir.joinpath("repo1"))
 
 
-def test_generate_repofiles(rooted_tmp_path: RootedPath) -> None:
+@pytest.mark.parametrize(
+    "expected_repofile",
+    [
+        pytest.param(
+            """
+            [repo1]
+            baseurl=file://{output_dir}/repo1
+            gpgcheck=1
+
+            [cachi2-repo]
+            baseurl=file://{output_dir}/cachi2-repo
+            gpgcheck=1
+            name=Packages unaffiliated with an official repository
+            """,
+            id="no_repo_options",
+        ),
+    ],
+)
+def test_generate_repofiles(rooted_tmp_path: RootedPath, expected_repofile: str) -> None:
     package_dir = rooted_tmp_path.join_within_root(DEFAULT_PACKAGE_DIR)
-    arch_dir = package_dir.path.joinpath("x86_64")
-    arch_dir.joinpath("repo1").mkdir(parents=True)
-    arch_dir.joinpath("cachi2-repo2").mkdir(parents=True)
-    arch_dir.joinpath("repos.d").mkdir(parents=True)
-    repopath = arch_dir.joinpath("repos.d", "cachi2.repo")
-    output_dir = f"{package_dir}/x86_64"
-    name = "Packages unaffiliated with an official repository"
+    arch_dir = Path(package_dir.path, "x86_64")
+    for dir_ in ["repo1", "cachi2-repo", "repos.d"]:
+        Path(arch_dir, dir_).mkdir(parents=True)
 
-    # repo items need to be sorted to match with the repofile
-    template = f"[cachi2-repo2]\nbaseurl=file://{output_dir}/cachi2-repo2\ngpgcheck=1\n{name}\n[repo1]\nbaseurl=file://{output_dir}/repo1\ngpgcheck=1\n"
     _generate_repofiles(rooted_tmp_path.path, rooted_tmp_path.path)
+    repopath = arch_dir.joinpath("repos.d", "cachi2.repo")
     with open(repopath) as f:
-        assert template == f.read()
+        actual = ConfigParser()
+        expected = ConfigParser()
+        actual.read_file(f)
+        expected.read_string(expected_repofile.format(output_dir=arch_dir.as_posix()))
+        assert expected == actual
 
 
 @mock.patch("cachi2.core.package_managers.rpm.main.run_cmd")
@@ -478,3 +498,63 @@ def test_uuid(self, raw_content: dict) -> None:
         lock = RedhatRpmsLock.model_validate(raw_content)
         uuid = lock._uuid
         assert len(uuid) == 6
+
+
+class TestRepofile:
+    @pytest.mark.parametrize(
+        "defaults, data, expected",
+        [
+            pytest.param(None, {}, True, id="no_defaults_no_sections"),
+            pytest.param({"foo": "bar"}, {}, True, id="just_defaults_no_sections"),
+            pytest.param({"fake": {"foo": "bar"}}, {}, True, id="complex_defaults_no_sections"),
+            pytest.param(None, {"section": {"foo": "bar"}}, False, id="with_data"),
+        ],
+    )
+    def test_empty(
+        self, data: Dict[str, Any], defaults: Optional[Dict[str, Any]], expected: bool
+    ) -> None:
+        actual = _Repofile(defaults)
+        actual.read_dict(data)
+        assert actual.empty == expected
+
+    @pytest.mark.parametrize(
+        "defaults, data, expected",
+        [
+            pytest.param(
+                None, {"section": {"foo": "bar"}}, {"section": {"foo": "bar"}}, id="no_defaults"
+            ),
+            pytest.param(
+                {"default": "baz"},
+                {"section": {"foo": "bar"}},
+                {"section": {"foo": "bar", "default": "baz"}},
+                id="defaults_no_value_conflict",
+            ),
+            pytest.param(
+                {"foo": "baz"},
+                {"section1": {"foo": "bar"}, "section2": {"foo2": "bar2"}},
+                {"section1": {"foo": "bar"}, "section2": {"foo2": "bar2", "foo": "baz"}},
+                id="defaults_value_conflict",
+            ),
+        ],
+    )
+    def test_apply_defaults(
+        self, data: Dict[str, Any], defaults: Optional[Dict[str, Any]], expected: Dict[str, Any]
+    ) -> None:
+        expected_r = _Repofile()
+        expected_r.read_dict(expected)
+        actual = _Repofile(defaults)
+        actual.read_dict(data)
+        actual._apply_defaults()
+        assert actual == expected_r
+
+    @mock.patch("cachi2.core.package_managers.rpm.main._Repofile._apply_defaults")
+    @mock.patch("cachi2.core.package_managers.rpm.main.ConfigParser.write")
+    def test_write(
+        self, mock_superclass_write: mock.Mock, mock_apply_defaults: mock.Mock, tmp_path: Path
+    ) -> None:
+        mock_superclass_write.return_value = None
+
+        with open(tmp_path / "test.repo", "w") as f:
+            _Repofile({"foo": "bar"}).write(f)
+
+        mock_apply_defaults.assert_called_once()