Generate RPM repo mapping

Added a new phase to generate compose/metadata/rpm-repo-mapping.yaml

JIRA: ROK-1541
Signed-off-by: Haibo Lin <hlin@redhat.com>

doc/_static/phases.svg

@@ -27,6 +27,20 @@
          transform="matrix(0,1,1,0,0,0)" />
       <text style="font-size:13px;font-family:sans-serif;" x="51.554729" y="970.26605">Pkgset</text>
     </g>
+    <g transform="translate(98.243246,-80.817124)">
+      <rect
+         style="fill:#4e9a06;fill-rule:evenodd;stroke:none;"
+         width="26.295755"
+         height="49.214859"
+         x="985.682984"
+         y="49.250374"
+         transform="matrix(0,1,1,0,0,0)" />
+      <text style="font-size:8px;font-family:sans-serif;fill:#ffffff;" x="51.554729" y="993.683">
+        <tspan x="51.554729">RPM</tspan>
+        <tspan x="51.554729" dy="8">Repo</tspan>
+        <tspan x="51.554729" dy="8">Mapping</tspan>
+      </text>
+    </g>
     <g transform="translate(141.04531,-80.817124)">
       <rect
          y="553.98242"

pungi/checks.py

@@ -791,6 +791,17 @@ def make_schema():
                 "patternProperties": {".+": {"$ref": "#/definitions/strings"}},
                 "additionalProperties": False,
             },
+            "rpm_repo_mapping": {
+                "type": "object",
+                "properties": {
+                    "repos": {
+                        "type": "object",
+                        "patternProperties": {".+": {"type": "string"}},
+                        "additionalProperties": False,
+                    },
+                },
+                "additionalProperties": False,
+            },
             "module_obsoletes_dir": {"$ref": "#/definitions/str_or_scm_dict"},
             "create_optional_isos": {"type": "boolean", "default": False},
             "symlink_isos_to": {"type": "string"},

pungi/phases/__init__.py

@@ -40,6 +40,7 @@ from .ostree_container import OSTreeContainerPhase  # noqa
 from .osbs import OSBSPhase  # noqa
 from .phases_metadata import gather_phases_metadata  # noqa
 from .trigger_konflux_pipeline import TriggerKonfluxPipelinePhase  # noqa
+from .rpm_repo_mapping import RpmRepoMappingPhase  # noqa
 
 this_module = sys.modules[__name__]
 PHASES_NAMES = gather_phases_metadata(this_module)

pungi/phases/gather/methods/method_deps.py

@@ -120,6 +120,7 @@ def write_pungi_config(
     prepopulate=None,
     source_name=None,
     package_sets=None,
+    rpm_repo_mapping=None,
 ):
     """write pungi config (kickstart) for arch/variant"""
     pungi_wrapper = PungiWrapper()
@@ -135,6 +136,8 @@ def write_pungi_config(
     for i, pkgset in enumerate(package_sets or []):
         if not variant.pkgsets or pkgset.name in variant.pkgsets:
             repos["pungi-repo-%d" % i] = pkgset.paths[arch]
+    if rpm_repo_mapping:
+        repos["rpm-repo-mapping"] = rpm_repo_mapping
     if compose.has_comps:
         repos["comps-repo"] = compose.paths.work.comps_repo(arch=arch, variant=variant)
     if variant.type == "optional":

pungi/phases/rpm_repo_mapping.py

@@ -0,0 +1,201 @@
+# -*- coding: utf-8 -*-
+
+
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 2 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Library General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, see <https://gnu.org/licenses/>.
+
+
+import os
+
+from kobo.rpmlib import parse_nvra
+from ruamel.yaml import YAML
+
+from ..compose import get_ordered_variant_uids
+from ..phases.gather.methods.method_deps import resolve_deps, write_pungi_config
+from ..phases.gather.phase import get_variant_packages
+from ..phases.gather.utils import get_prepopulate_packages
+from .base import ConfigGuardedPhase
+
+
+class RpmRepoMappingPhase(ConfigGuardedPhase):
+    """Generate RPM repo mapping file"""
+
+    name = "rpm_repo_mapping"
+
+    def run(self):
+        lookaside = {}
+        for dest, lookaside_variant_uid in self.compose.conf.get(
+            "variant_as_lookaside", []
+        ):
+            if dest not in lookaside:
+                lookaside[dest] = set()
+            lookaside[dest].add(lookaside_variant_uid)
+
+        mapping = {}
+        for variant_uid in get_ordered_variant_uids(self.compose):
+            variant = self.compose.all_variants[variant_uid]
+            for arch in variant.arches:
+                generate_mapping(
+                    mapping,
+                    self.compose,
+                    arch,
+                    variant,
+                    *get_variant_packages(self.compose, arch, variant, "comps"),
+                    lookaside,
+                )
+        write_rpm_repo_mapping(mapping, self.compose)
+
+
+def generate_mapping(
+    mapping, compose, arch, variant, packages, groups, filter_packages, lookaside
+):
+    """
+    Generate RPM repository mapping for a given architecture and variant.
+
+    This function processes packages and creates a mapping structure that tracks
+    which repository architectures contain each package name and architecture
+    combination. The mapping is organized as:
+        variant_uid -> package_name -> package_arch -> [repo_archs]
+
+    Args:
+        mapping (dict): The global mapping dictionary to update with new entries
+        compose: The pungi.compose.Compose object
+        arch (str): The repository architecture being processed
+        variant: The variant object containing uid and other metadata
+        packages (list): List of package names to include in the mapping
+        groups (list): List of package groups to include in the mapping
+        filter_packages (list): List of packages to filter out from the mapping
+
+    Returns:
+        None: The function modifies the mapping dictionary in-place
+    """
+
+    source_name = "rpm-repo-mapping"
+    repo = compose.conf.get("rpm_repo_mapping", {}).get("repos", {}).get(arch, "")
+    if repo:
+        compose.log_debug(
+            "Generating rpm repo mapping (arch: %s, variant: %s)" % (arch, variant)
+        )
+    else:
+        compose.log_debug(
+            "Skip rpm repo mapping generation (arch: %s, variant: %s) "
+            "as missing configuration" % (arch, variant)
+        )
+        return
+
+    write_pungi_config(
+        compose,
+        arch,
+        variant,
+        packages,
+        groups,
+        filter_packages,
+        multilib_whitelist=None,
+        multilib_blacklist=None,
+        fulltree_excludes=None,
+        prepopulate=get_prepopulate_packages(compose, arch, variant),
+        source_name=source_name,
+        package_sets=None,
+        rpm_repo_mapping=repo,
+    )
+    pkg_map, _ = resolve_deps(compose, arch, variant, source_name=source_name)
+    # Iterate through repository architectures
+    if variant.uid not in mapping:
+        mapping[variant.uid] = {}
+
+    # Process each package type (rpm, srpm, debuginfo)
+    for pkg_type, packages in pkg_map.items():
+        for pkg in packages:
+            pkg_path = pkg["path"]
+            pkg_filename = os.path.basename(pkg_path)
+
+            # Skip non-RPM files
+            if not pkg_filename.endswith(".rpm"):
+                continue
+
+            nvra = parse_nvra(pkg_filename)
+            pkg_name = nvra["name"]
+
+            # Determine package architecture
+            if pkg_type == "srpm":
+                pkg_arch = "src"
+            else:
+                pkg_arch = nvra["arch"]
+
+            # Check lookaside repo
+            if variant.uid in lookaside:
+                found = False
+                for required in lookaside[variant.uid]:
+                    if (
+                        pkg_name in mapping[required]
+                        and pkg_arch in mapping[required][pkg_name]
+                        and arch in mapping[required][pkg_name][pkg_arch]
+                    ):
+                        found = True
+                        break
+                if found:
+                    continue
+
+            # Initialize nested structure
+            if pkg_name not in mapping[variant.uid]:
+                mapping[variant.uid][pkg_name] = {}
+
+            if pkg_arch not in mapping[variant.uid][pkg_name]:
+                mapping[variant.uid][pkg_name][pkg_arch] = []
+
+            # Add arch if not already present
+            if arch not in mapping[variant.uid][pkg_name][pkg_arch]:
+                mapping[variant.uid][pkg_name][pkg_arch].append(arch)
+
+
+def _sort_mapping(data):
+    """Recursively sorts a rpm repo mapping dictionary."""
+    if isinstance(data, dict):
+        return {k: _sort_mapping(v) for k, v in sorted(data.items())}
+    elif isinstance(data, list):
+        return sorted(data)
+    else:
+        return data
+
+
+def write_rpm_repo_mapping(mapping, compose):
+    """
+    Write the RPM repository mapping to a YAML file and push to Git repository.
+
+    This function serializes the complete RPM repository mapping structure to
+    a YAML file in the compose metadata directory and pushes it to the specified
+    Git repository.
+    The mapping contains information about which repository architectures
+    contain each package.
+
+    Args:
+        mapping (dict): The complete mapping dictionary containing:
+                       variant_uid -> package_name -> package_arch -> [repo_archs]
+        compose: The pungi.compose.Compose object
+
+    Returns:
+        None
+
+    File Format:
+        The output YAML file contains a nested structure where:
+        - Top level keys are variant UIDs
+        - Second level keys are package names
+        - Third level keys are package architectures
+        - Values are lists of repository architectures containing the package
+    """
+    mapping_file = compose.paths.compose.metadata("rpm-repo-mapping.yaml")
+    compose.log_debug(f"Writing rpm repo mapping: {mapping_file}")
+    with open(mapping_file, "w") as f:
+        yaml_handler = YAML()
+        yaml_handler.default_flow_style = False
+        yaml_handler.indent(mapping=2, offset=2)
+        yaml_handler.dump(_sort_mapping(mapping), f)

pungi/scripts/pungi_koji.py

@@ -407,6 +407,7 @@ def run_compose(
 
     # initialize all phases
     init_phase = pungi.phases.InitPhase(compose)
+    rpm_repo_mapping_phase = pungi.phases.RpmRepoMappingPhase(compose)
     pkgset_phase = pungi.phases.PkgsetPhase(compose)
     buildinstall_phase = pungi.phases.BuildinstallPhase(compose, pkgset_phase)
     gather_phase = pungi.phases.GatherPhase(compose, pkgset_phase)
@@ -434,6 +435,7 @@ def run_compose(
     # check if all config options are set
     for phase in (
         init_phase,
+        rpm_repo_mapping_phase,
         pkgset_phase,
         createrepo_phase,
         buildinstall_phase,
@@ -471,8 +473,10 @@ def run_compose(
     init_phase.start()
     init_phase.stop()
 
-    pkgset_phase.start()
-    pkgset_phase.stop()
+    setup_schema = (pkgset_phase, rpm_repo_mapping_phase)
+    setup_phase = pungi.phases.WeaverPhase(compose, setup_schema)
+    setup_phase.start()
+    setup_phase.stop()
 
     # WEAVER phase - launches other phases which can safely run in parallel
     essentials_schema = (

requirements.txt

@@ -11,3 +11,4 @@ ordered_set
 productmd
 pykickstart
 python-multilib
+ruamel.yaml

tests/test_rpm_repo_mapping.py

@@ -0,0 +1,474 @@
+# -*- coding: utf-8 -*-
+
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 2 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Library General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, see <https://gnu.org/licenses/>.
+
+import os
+import unittest
+from unittest import mock
+
+from pungi.phases.rpm_repo_mapping import (
+    RpmRepoMappingPhase,
+    _sort_mapping,
+    generate_mapping,
+    write_rpm_repo_mapping,
+)
+from tests import helpers
+
+
+class TestRpmRepoMappingPhaseSkip(helpers.PungiTestCase):
+    def test_skip_when_no_config(self):
+        compose = helpers.DummyCompose(self.topdir, {})
+        compose.just_phases = None
+        compose.skip_phases = []
+        phase = RpmRepoMappingPhase(compose)
+        self.assertTrue(phase.skip())
+
+    def test_no_skip_when_config_present(self):
+        compose = helpers.DummyCompose(
+            self.topdir,
+            {"rpm_repo_mapping": {"repos": {"x86_64": "http://example.com/repo"}}},
+        )
+        compose.just_phases = None
+        compose.skip_phases = []
+        phase = RpmRepoMappingPhase(compose)
+        self.assertFalse(phase.skip())
+
+
+class TestRpmRepoMappingPhaseRun(helpers.PungiTestCase):
+    @mock.patch("pungi.phases.rpm_repo_mapping.write_rpm_repo_mapping")
+    @mock.patch("pungi.phases.rpm_repo_mapping.generate_mapping")
+    @mock.patch("pungi.phases.rpm_repo_mapping.get_variant_packages")
+    @mock.patch("pungi.phases.rpm_repo_mapping.get_ordered_variant_uids")
+    def test_run_calls_generate_for_each_variant_arch(
+        self,
+        mock_get_ordered,
+        mock_get_variant_packages,
+        mock_generate,
+        mock_write,
+    ):
+        compose = helpers.DummyCompose(
+            self.topdir,
+            {"rpm_repo_mapping": {"repos": {"x86_64": "http://example.com/repo"}}},
+        )
+        server = compose.all_variants["Server"]
+        server.arches = ["x86_64"]
+        mock_get_ordered.return_value = ["Server"]
+        mock_get_variant_packages.return_value = (["pkg1"], ["grp1"], ["filter1"])
+
+        phase = RpmRepoMappingPhase(compose)
+        phase.run()
+
+        mock_get_ordered.assert_called_once_with(compose)
+        mock_get_variant_packages.assert_called_once_with(
+            compose, "x86_64", server, "comps"
+        )
+        mock_generate.assert_called_once_with(
+            {},
+            compose,
+            "x86_64",
+            server,
+            ["pkg1"],
+            ["grp1"],
+            ["filter1"],
+            {},
+        )
+        mock_write.assert_called_once_with({}, compose)
+
+
+class TestGenerateMapping(helpers.PungiTestCase):
+    def setUp(self):
+        super().setUp()
+        self.compose = helpers.DummyCompose(
+            self.topdir,
+            {
+                "rpm_repo_mapping": {
+                    "repos": {"x86_64": "http://example.com/x86_64/repo"}
+                }
+            },
+        )
+        self.variant = helpers.MockVariant(
+            uid="BaseOS", arches=["x86_64"], type="variant"
+        )
+
+    @mock.patch("pungi.phases.rpm_repo_mapping.get_prepopulate_packages")
+    @mock.patch("pungi.phases.rpm_repo_mapping.resolve_deps")
+    @mock.patch("pungi.phases.rpm_repo_mapping.write_pungi_config")
+    def test_skip_when_no_repo_config_for_arch(
+        self, mock_write_cfg, mock_resolve, mock_prepopulate
+    ):
+        mapping = {}
+        generate_mapping(
+            mapping,
+            self.compose,
+            "aarch64",  # not in repos config
+            self.variant,
+            [],
+            [],
+            [],
+            {},
+        )
+        mock_write_cfg.assert_not_called()
+        mock_resolve.assert_not_called()
+        self.assertEqual(mapping, {})
+
+    @mock.patch("pungi.phases.rpm_repo_mapping.get_prepopulate_packages")
+    @mock.patch("pungi.phases.rpm_repo_mapping.resolve_deps")
+    @mock.patch("pungi.phases.rpm_repo_mapping.write_pungi_config")
+    def test_adds_rpm_package_to_mapping(
+        self, mock_write_cfg, mock_resolve, mock_prepopulate
+    ):
+        mock_prepopulate.return_value = set()
+        mock_resolve.return_value = (
+            {
+                "rpm": [{"path": "/repo/foo-1.0-1.x86_64.rpm"}],
+                "srpm": [],
+                "debuginfo": [],
+            },
+            None,
+        )
+
+        mapping = {}
+        generate_mapping(
+            mapping,
+            self.compose,
+            "x86_64",
+            self.variant,
+            ["foo"],
+            [],
+            [],
+            {},
+        )
+
+        self.assertIn("BaseOS", mapping)
+        self.assertIn("foo", mapping["BaseOS"])
+        self.assertIn("x86_64", mapping["BaseOS"]["foo"])
+        self.assertEqual(mapping["BaseOS"]["foo"]["x86_64"], ["x86_64"])
+
+    @mock.patch("pungi.phases.rpm_repo_mapping.get_prepopulate_packages")
+    @mock.patch("pungi.phases.rpm_repo_mapping.resolve_deps")
+    @mock.patch("pungi.phases.rpm_repo_mapping.write_pungi_config")
+    def test_adds_srpm_with_src_arch(
+        self, mock_write_cfg, mock_resolve, mock_prepopulate
+    ):
+        mock_prepopulate.return_value = set()
+        mock_resolve.return_value = (
+            {
+                "rpm": [],
+                "srpm": [{"path": "/repo/foo-1.0-1.src.rpm"}],
+                "debuginfo": [],
+            },
+            None,
+        )
+
+        mapping = {}
+        generate_mapping(
+            mapping,
+            self.compose,
+            "x86_64",
+            self.variant,
+            ["foo"],
+            [],
+            [],
+            {},
+        )
+
+        self.assertIn("foo", mapping["BaseOS"])
+        self.assertIn("src", mapping["BaseOS"]["foo"])
+        self.assertEqual(mapping["BaseOS"]["foo"]["src"], ["x86_64"])
+
+    @mock.patch("pungi.phases.rpm_repo_mapping.get_prepopulate_packages")
+    @mock.patch("pungi.phases.rpm_repo_mapping.resolve_deps")
+    @mock.patch("pungi.phases.rpm_repo_mapping.write_pungi_config")
+    def test_skips_non_rpm_files(self, mock_write_cfg, mock_resolve, mock_prepopulate):
+        mock_prepopulate.return_value = set()
+        mock_resolve.return_value = (
+            {
+                "rpm": [{"path": "/repo/some-file.tar.gz"}],
+                "srpm": [],
+                "debuginfo": [],
+            },
+            None,
+        )
+
+        mapping = {}
+        generate_mapping(
+            mapping,
+            self.compose,
+            "x86_64",
+            self.variant,
+            [],
+            [],
+            [],
+            {},
+        )
+
+        self.assertEqual(mapping["BaseOS"], {})
+
+    @mock.patch("pungi.phases.rpm_repo_mapping.get_prepopulate_packages")
+    @mock.patch("pungi.phases.rpm_repo_mapping.resolve_deps")
+    @mock.patch("pungi.phases.rpm_repo_mapping.write_pungi_config")
+    def test_does_not_duplicate_arch(
+        self, mock_write_cfg, mock_resolve, mock_prepopulate
+    ):
+        mock_prepopulate.return_value = set()
+        mock_resolve.return_value = (
+            {
+                "rpm": [
+                    {"path": "/repo/foo-1.0-1.x86_64.rpm"},
+                    {"path": "/repo/foo-1.0-1.x86_64.rpm"},
+                ],
+                "srpm": [],
+                "debuginfo": [],
+            },
+            None,
+        )
+
+        mapping = {}
+        generate_mapping(
+            mapping,
+            self.compose,
+            "x86_64",
+            self.variant,
+            ["foo"],
+            [],
+            [],
+            {},
+        )
+
+        self.assertEqual(mapping["BaseOS"]["foo"]["x86_64"], ["x86_64"])
+
+    @mock.patch("pungi.phases.rpm_repo_mapping.get_prepopulate_packages")
+    @mock.patch("pungi.phases.rpm_repo_mapping.resolve_deps")
+    @mock.patch("pungi.phases.rpm_repo_mapping.write_pungi_config")
+    def test_skips_package_present_in_lookaside(
+        self, mock_write_cfg, mock_resolve, mock_prepopulate
+    ):
+        mock_prepopulate.return_value = set()
+        mock_resolve.return_value = (
+            {
+                "rpm": [{"path": "/repo/foo-1.0-1.x86_64.rpm"}],
+                "srpm": [],
+                "debuginfo": [],
+            },
+            None,
+        )
+
+        # Pre-populate mapping with the package in a lookaside variant
+        mapping = {
+            "BaseOS": {"foo": {"x86_64": ["x86_64"]}},
+            "AppStream": {},
+        }
+        everything_variant = helpers.MockVariant(
+            uid="AppStream", arches=["x86_64"], type="variant"
+        )
+        lookaside = {"AppStream": {"BaseOS"}}
+
+        generate_mapping(
+            mapping,
+            self.compose,
+            "x86_64",
+            everything_variant,
+            ["foo"],
+            [],
+            [],
+            lookaside,
+        )
+
+        # foo should not be added to Everything since it's in lookaside Server
+        self.assertEqual(mapping["AppStream"], {})
+
+    @mock.patch("pungi.phases.rpm_repo_mapping.get_prepopulate_packages")
+    @mock.patch("pungi.phases.rpm_repo_mapping.resolve_deps")
+    @mock.patch("pungi.phases.rpm_repo_mapping.write_pungi_config")
+    def test_adds_package_not_in_lookaside(
+        self, mock_write_cfg, mock_resolve, mock_prepopulate
+    ):
+        mock_prepopulate.return_value = set()
+        mock_resolve.return_value = (
+            {
+                "rpm": [{"path": "/repo/bar-1.0-1.x86_64.rpm"}],
+                "srpm": [],
+                "debuginfo": [],
+            },
+            None,
+        )
+
+        # Lookaside has a different package
+        mapping = {
+            "BaseOS": {"foo": {"x86_64": ["x86_64"]}},
+            "AppStream": {},
+        }
+        everything_variant = helpers.MockVariant(
+            uid="AppStream", arches=["x86_64"], type="variant"
+        )
+        lookaside = {"AppStream": {"BaseOS"}}
+
+        generate_mapping(
+            mapping,
+            self.compose,
+            "x86_64",
+            everything_variant,
+            ["bar"],
+            [],
+            [],
+            lookaside,
+        )
+
+        # bar is not in Server, so it should be added to Everything
+        self.assertIn("bar", mapping["AppStream"])
+        self.assertEqual(mapping["AppStream"]["bar"]["x86_64"], ["x86_64"])
+
+    @mock.patch("pungi.phases.rpm_repo_mapping.get_prepopulate_packages")
+    @mock.patch("pungi.phases.rpm_repo_mapping.resolve_deps")
+    @mock.patch("pungi.phases.rpm_repo_mapping.write_pungi_config")
+    def test_accumulates_multiple_repo_arches(
+        self, mock_write_cfg, mock_resolve, mock_prepopulate
+    ):
+        compose = helpers.DummyCompose(
+            self.topdir,
+            {
+                "rpm_repo_mapping": {
+                    "repos": {
+                        "x86_64": "http://example.com/x86_64/repo",
+                        "aarch64": "http://example.com/aarch64/repo",
+                    }
+                }
+            },
+        )
+        mock_prepopulate.return_value = set()
+        mock_resolve.return_value = (
+            {
+                "rpm": [{"path": "/repo/foo-1.0-1.x86_64.rpm"}],
+                "srpm": [],
+                "debuginfo": [],
+            },
+            None,
+        )
+
+        mapping = {}
+        generate_mapping(mapping, compose, "x86_64", self.variant, ["foo"], [], [], {})
+
+        mock_resolve.return_value = (
+            {
+                "rpm": [{"path": "/repo/foo-1.0-1.aarch64.rpm"}],
+                "srpm": [],
+                "debuginfo": [],
+            },
+            None,
+        )
+        generate_mapping(mapping, compose, "aarch64", self.variant, ["foo"], [], [], {})
+
+        self.assertCountEqual(mapping["BaseOS"]["foo"]["x86_64"], ["x86_64"])
+        self.assertCountEqual(mapping["BaseOS"]["foo"]["aarch64"], ["aarch64"])
+
+    @mock.patch("pungi.phases.rpm_repo_mapping.get_prepopulate_packages")
+    @mock.patch("pungi.phases.rpm_repo_mapping.resolve_deps")
+    @mock.patch("pungi.phases.rpm_repo_mapping.write_pungi_config")
+    def test_write_pungi_config_called_with_repo(
+        self, mock_write_cfg, mock_resolve, mock_prepopulate
+    ):
+        mock_prepopulate.return_value = set()
+        mock_resolve.return_value = (
+            {"rpm": [], "srpm": [], "debuginfo": []},
+            None,
+        )
+
+        generate_mapping(
+            {},
+            self.compose,
+            "x86_64",
+            self.variant,
+            ["pkg1"],
+            ["grp1"],
+            ["filter1"],
+            {},
+        )
+
+        mock_write_cfg.assert_called_once_with(
+            self.compose,
+            "x86_64",
+            self.variant,
+            ["pkg1"],
+            ["grp1"],
+            ["filter1"],
+            multilib_whitelist=None,
+            multilib_blacklist=None,
+            fulltree_excludes=None,
+            prepopulate=mock_prepopulate.return_value,
+            source_name="rpm-repo-mapping",
+            package_sets=None,
+            rpm_repo_mapping="http://example.com/x86_64/repo",
+        )
+
+
+class TestSortMapping(unittest.TestCase):
+    def test_sorts_dict_keys(self):
+        data = {"b": 1, "a": 2}
+        result = _sort_mapping(data)
+        self.assertEqual(list(result.keys()), ["a", "b"])
+
+    def test_sorts_nested_dict(self):
+        data = {"z": {"b": [3, 1], "a": [2]}, "a": {}}
+        result = _sort_mapping(data)
+        self.assertEqual(list(result.keys()), ["a", "z"])
+        self.assertEqual(list(result["z"].keys()), ["a", "b"])
+
+    def test_sorts_lists(self):
+        data = {"pkg": {"x86_64": ["aarch64", "x86_64", "ppc64le"]}}
+        result = _sort_mapping(data)
+        self.assertEqual(result["pkg"]["x86_64"], ["aarch64", "ppc64le", "x86_64"])
+
+    def test_passthrough_scalar(self):
+        self.assertEqual(_sort_mapping("hello"), "hello")
+        self.assertEqual(_sort_mapping(42), 42)
+
+
+class TestWriteRpmRepoMapping(helpers.PungiTestCase):
+    def test_writes_yaml_file(self):
+        compose = helpers.DummyCompose(self.topdir, {})
+        mapping = {
+            "BaseOS": {
+                "foo": {"x86_64": ["x86_64"]},
+            }
+        }
+
+        mapping_file = compose.paths.compose.metadata("rpm-repo-mapping.yaml")
+        os.makedirs(os.path.dirname(mapping_file), exist_ok=True)
+
+        write_rpm_repo_mapping(mapping, compose)
+
+        self.assertTrue(os.path.exists(mapping_file))
+        with open(mapping_file) as f:
+            content = f.read()
+        self.assertIn("BaseOS", content)
+        self.assertIn("foo", content)
+        self.assertIn("x86_64", content)
+
+    def test_writes_sorted_yaml(self):
+        compose = helpers.DummyCompose(self.topdir, {})
+        mapping = {
+            "BaseOS": {
+                "zzz": {"x86_64": ["x86_64"]},
+                "aaa": {"x86_64": ["x86_64"]},
+            }
+        }
+
+        mapping_file = compose.paths.compose.metadata("rpm-repo-mapping.yaml")
+        os.makedirs(os.path.dirname(mapping_file), exist_ok=True)
+
+        write_rpm_repo_mapping(mapping, compose)
+
+        with open(mapping_file) as f:
+            content = f.read()
+        # aaa should appear before zzz in sorted output
+        self.assertLess(content.index("aaa"), content.index("zzz"))