id_configurator/tests/test_yum.py

import subprocess
from configparser import ConfigParser
from unittest.mock import MagicMock

import pytest

from id_configurator.config import Configuration
from id_configurator.context import Context
from id_configurator.modules.yum import YumModule

KEY_CONTENT = "fake-rpm-gpg-key-bytes"

CONFIG_WITH_REPO_AND_KEY = {
    "uuid": "12345678-1234-5678-1234-567812345678",
    "organisation": "InterDiode",
    "configurations": {
        "yum": {
            "repositories": [
                {
                    "name": "interdiode",
                    "index-url": "https://example.invalid/yum",
                    "gpg-url": "https://example.invalid/RPM-GPG-KEY-interdiode",
                }
            ],
            "keys": [{"name": "interdiode", "content": KEY_CONTENT}],
        }
    },
}

CONFIG_WITH_UNSIGNED_REPO = {
    "uuid": "12345678-1234-5678-1234-567812345678",
    "organisation": "InterDiode",
    "configurations": {
        "yum": {
            "repositories": [
                {"name": "unsigned", "index-url": "https://example.invalid/yum"}
            ]
        }
    },
}

EMPTY_CONFIG = {
    "uuid": "12345678-1234-5678-1234-567812345678",
    "organisation": "InterDiode",
}


def make_config(data: dict) -> Configuration:
    return Configuration.model_validate(data)


def test_is_relevant_false_when_empty(tmp_path) -> None:
    context = Context(config=make_config(EMPTY_CONFIG), dst=tmp_path / "dst", src=None)
    assert YumModule().is_relevant(context) is False


def test_is_relevant_true_with_repository(tmp_path) -> None:
    context = Context(
        config=make_config(CONFIG_WITH_UNSIGNED_REPO), dst=tmp_path / "dst", src=None
    )
    assert YumModule().is_relevant(context) is True


def test_is_relevant_true_with_existing_repo_file(tmp_path) -> None:
    src = tmp_path / "src"
    repos_dir = src / "etc/yum.repos.d"
    repos_dir.mkdir(parents=True)
    (repos_dir / "old.repo").write_text(
        "[old]\nname=old\nbaseurl=https://old.invalid\nenabled=1\n", encoding="utf-8"
    )
    context = Context(config=make_config(EMPTY_CONFIG), dst=tmp_path / "dst", src=src)
    assert YumModule().is_relevant(context) is True


def test_is_relevant_false_with_only_unrelated_src_files(tmp_path) -> None:
    src = tmp_path / "src"
    repos_dir = src / "etc/yum.repos.d"
    repos_dir.mkdir(parents=True)
    (repos_dir / "README").write_text("not a repo file\n", encoding="utf-8")
    context = Context(config=make_config(EMPTY_CONFIG), dst=tmp_path / "dst", src=src)
    assert YumModule().is_relevant(context) is False


def _mock_gpg_success(monkeypatch) -> MagicMock:
    mock_run = MagicMock(
        return_value=subprocess.CompletedProcess(
            args=[], returncode=0, stdout=b"", stderr=b""
        )
    )
    monkeypatch.setattr("id_configurator.modules.yum.subprocess.run", mock_run)
    return mock_run


def test_apply_writes_signed_repo_and_imports_key(tmp_path, monkeypatch) -> None:
    mock_run = _mock_gpg_success(monkeypatch)
    dst = tmp_path / "dst"
    context = Context(config=make_config(CONFIG_WITH_REPO_AND_KEY), dst=dst, src=None)

    YumModule().apply(context)

    repo_content = (dst / "etc/yum.repos.d/interdiode.repo").read_text(encoding="utf-8")
    assert repo_content == (
        "[interdiode]\n"
        "name=interdiode\n"
        "baseurl=https://example.invalid/yum\n"
        "enabled=1\n"
        "gpgcheck=1\n"
        "gpgkey=https://example.invalid/RPM-GPG-KEY-interdiode\n"
    )

    key_content = (dst / "etc/pki/rpm-gpg/RPM-GPG-KEY-interdiode").read_bytes()
    assert key_content == b"fake-rpm-gpg-key-bytes"
    mock_run.assert_called_once()
    assert mock_run.call_args.kwargs["input"] == b"fake-rpm-gpg-key-bytes"


def test_apply_unsigned_repo_disables_gpgcheck(tmp_path) -> None:
    dst = tmp_path / "dst"
    context = Context(config=make_config(CONFIG_WITH_UNSIGNED_REPO), dst=dst, src=None)

    YumModule().apply(context)

    repo_content = (dst / "etc/yum.repos.d/unsigned.repo").read_text(encoding="utf-8")
    assert repo_content == (
        "[unsigned]\nname=unsigned\nbaseurl=https://example.invalid/yum\nenabled=1\ngpgcheck=0\n"
    )


def test_apply_copies_existing_repos_unchanged_without_index_url_to_replace(
    tmp_path,
) -> None:
    src = tmp_path / "src"
    repos_dir = src / "etc/yum.repos.d"
    repos_dir.mkdir(parents=True)
    (repos_dir / "multi.repo").write_text(
        "[old-one]\nname=old-one\nbaseurl=https://old-one.invalid\nenabled=1\ngpgcheck=1\n"
        "\n"
        "[old-two]\nname=old-two\nbaseurl=https://old-two.invalid\nenabled=1\n",
        encoding="utf-8",
    )
    (repos_dir / "ignored.txt").write_text("not a repo file\n", encoding="utf-8")
    # A same-named existing file must end up enabled again once the configured repo overwrites it.
    (repos_dir / "unsigned.repo").write_text(
        "[unsigned]\nname=unsigned\nbaseurl=https://stale.invalid\nenabled=1\n",
        encoding="utf-8",
    )

    dst = tmp_path / "dst"
    context = Context(config=make_config(CONFIG_WITH_UNSIGNED_REPO), dst=dst, src=src)

    YumModule().apply(context)

    # No configured repository declares an index_url_to_replace, so the
    # existing sections are copied through unchanged, still enabled.
    copied_multi = (dst / "etc/yum.repos.d/multi.repo").read_text(encoding="utf-8")
    parser = ConfigParser(interpolation=None)
    parser.read_string(copied_multi)
    assert parser["old-one"]["enabled"] == "1"
    assert parser["old-two"]["enabled"] == "1"

    assert not (dst / "etc/yum.repos.d/ignored.txt").exists()

    overridden = (dst / "etc/yum.repos.d/unsigned.repo").read_text(encoding="utf-8")
    assert (
        overridden
        == "[unsigned]\nname=unsigned\nbaseurl=https://example.invalid/yum\nenabled=1\ngpgcheck=0\n"
    )


def test_apply_disables_only_existing_sections_matching_index_url_to_replace(
    tmp_path,
) -> None:
    src = tmp_path / "src"
    repos_dir = src / "etc/yum.repos.d"
    repos_dir.mkdir(parents=True)
    (repos_dir / "multi.repo").write_text(
        "[old-one]\nname=old-one\nbaseurl=https://old-one.invalid\nenabled=1\ngpgcheck=1\n"
        "\n"
        "[old-two]\nname=old-two\nbaseurl=https://old-two.invalid\nenabled=1\n",
        encoding="utf-8",
    )

    data = {
        "uuid": "12345678-1234-5678-1234-567812345678",
        "organisation": "InterDiode",
        "configurations": {
            "yum": {
                "repositories": [
                    {
                        "name": "unsigned",
                        "index-url": "https://example.invalid/yum",
                        "index-url-to-replace": "https://old-one.invalid",
                    }
                ]
            }
        },
    }
    dst = tmp_path / "dst"
    context = Context(config=make_config(data), dst=dst, src=src)

    YumModule().apply(context)

    disabled_multi = (dst / "etc/yum.repos.d/multi.repo").read_text(encoding="utf-8")
    parser = ConfigParser(interpolation=None)
    parser.read_string(disabled_multi)
    assert parser["old-one"]["enabled"] == "0"
    assert parser["old-two"]["enabled"] == "1"


def test_apply_skips_repository_when_index_url_to_replace_matches_nothing(
    tmp_path,
) -> None:
    src = tmp_path / "src"
    repos_dir = src / "etc/yum.repos.d"
    repos_dir.mkdir(parents=True)
    (repos_dir / "old.repo").write_text(
        "[old]\nname=old\nbaseurl=https://old.invalid\nenabled=1\n", encoding="utf-8"
    )
    data = {
        "uuid": "12345678-1234-5678-1234-567812345678",
        "organisation": "InterDiode",
        "configurations": {
            "yum": {
                "repositories": [
                    {
                        "name": "unsigned",
                        "index-url": "https://example.invalid/yum",
                        "index-url-to-replace": "https://other.invalid/*",
                    }
                ]
            }
        },
    }
    dst = tmp_path / "dst"
    context = Context(config=make_config(data), dst=dst, src=src)

    YumModule().apply(context)

    assert not (dst / "etc/yum.repos.d/unsigned.repo").exists()


def test_apply_writes_repository_when_index_url_to_replace_matches_existing(
    tmp_path,
) -> None:
    src = tmp_path / "src"
    repos_dir = src / "etc/yum.repos.d"
    repos_dir.mkdir(parents=True)
    (repos_dir / "old.repo").write_text(
        "[old]\nname=old\nbaseurl=https://old.invalid/yum\nenabled=1\n",
        encoding="utf-8",
    )
    data = {
        "uuid": "12345678-1234-5678-1234-567812345678",
        "organisation": "InterDiode",
        "configurations": {
            "yum": {
                "repositories": [
                    {
                        "name": "unsigned",
                        "index-url": "https://example.invalid/yum",
                        "index-url-to-replace": "https://old.invalid/*",
                    }
                ]
            }
        },
    }
    dst = tmp_path / "dst"
    context = Context(config=make_config(data), dst=dst, src=src)

    YumModule().apply(context)

    assert (dst / "etc/yum.repos.d/unsigned.repo").exists()


def test_apply_matches_index_url_to_replace_with_unrelated_files_and_multiple_repos(
    tmp_path,
) -> None:
    src = tmp_path / "src"
    repos_dir = src / "etc/yum.repos.d"
    repos_dir.mkdir(parents=True)
    (repos_dir / "old.repo").write_text(
        "[old]\nname=old\nbaseurl=https://old.invalid/yum\nenabled=1\n",
        encoding="utf-8",
    )
    (repos_dir / "ignored.txt").write_text("not a repo file\n", encoding="utf-8")

    data = {
        "uuid": "12345678-1234-5678-1234-567812345678",
        "organisation": "InterDiode",
        "configurations": {
            "yum": {
                "repositories": [
                    {
                        "name": "first",
                        "index-url": "https://example.invalid/first",
                        "index-url-to-replace": "https://old.invalid/*",
                    },
                    {
                        "name": "second",
                        "index-url": "https://example.invalid/second",
                        "index-url-to-replace": "https://old.invalid/*",
                    },
                ]
            }
        },
    }
    dst = tmp_path / "dst"
    context = Context(config=make_config(data), dst=dst, src=src)

    YumModule().apply(context)

    assert (dst / "etc/yum.repos.d/first.repo").exists()
    assert (dst / "etc/yum.repos.d/second.repo").exists()


def test_apply_sanitizes_repository_and_key_names_with_slash(
    tmp_path, monkeypatch
) -> None:
    mock_run = _mock_gpg_success(monkeypatch)
    data = {
        "uuid": "12345678-1234-5678-1234-567812345678",
        "organisation": "InterDiode",
        "configurations": {
            "yum": {
                "repositories": [
                    {
                        "name": "../../evil-repo",
                        "index-url": "https://example.invalid/yum",
                    }
                ],
                "keys": [{"name": "../../evil-key", "content": KEY_CONTENT}],
            }
        },
    }
    dst = tmp_path / "dst"
    context = Context(config=make_config(data), dst=dst, src=None)

    YumModule().apply(context)

    assert (dst / "etc/yum.repos.d/.._.._evil-repo.repo").exists()
    key_content = (dst / "etc/pki/rpm-gpg/RPM-GPG-KEY-.._.._evil-key").read_bytes()
    assert key_content == b"fake-rpm-gpg-key-bytes"
    mock_run.assert_called_once()


def test_import_key_raises_when_gpg_binary_missing(tmp_path, monkeypatch) -> None:
    monkeypatch.setattr(
        "id_configurator.modules.yum.subprocess.run",
        MagicMock(side_effect=FileNotFoundError("invalid GPG key")),
    )
    context = Context(
        config=make_config(CONFIG_WITH_REPO_AND_KEY), dst=tmp_path / "dst", src=None
    )

    with pytest.raises(FileNotFoundError, match="invalid GPG key"):
        YumModule().apply(context)


def test_import_key_raises_when_gpg_not_on_path(tmp_path, monkeypatch) -> None:
    monkeypatch.setattr("id_configurator.modules.yum.shutil.which", lambda _: None)
    context = Context(
        config=make_config(CONFIG_WITH_REPO_AND_KEY), dst=tmp_path / "dst", src=None
    )

    with pytest.raises(RuntimeError, match="gpg is not available"):
        YumModule().apply(context)


def test_import_key_raises_on_invalid_key(tmp_path, monkeypatch) -> None:
    mock_run = MagicMock(
        return_value=subprocess.CompletedProcess(
            args=[], returncode=2, stdout=b"", stderr=b"gpg: no valid data found"
        )
    )
    monkeypatch.setattr("id_configurator.modules.yum.subprocess.run", mock_run)
    context = Context(
        config=make_config(CONFIG_WITH_REPO_AND_KEY), dst=tmp_path / "dst", src=None
    )

    with pytest.raises(RuntimeError, match="invalid GPG key interdiode"):
        YumModule().apply(context)