id_configurator/tests/test_apt.py

from id_configurator.config import Configuration
from id_configurator.context import Context
from id_configurator.modules.apt import AptModule

KEY_CONTENT = (
    "-----BEGIN PGP PUBLIC KEY BLOCK-----\n"
    "\n"
    "mQENBGqh08IBCAC0\n"
    "-----END PGP PUBLIC KEY BLOCK-----\n"
)

SOURCE_CONTENT = {
    "uuid": "12345678-1234-5678-1234-567812345678",
    "organisation": "InterDiode",
    "configurations": {
        "apt": {
            "sources": [
                {
                    "name": "interdiode",
                    "uris": ["http://example.invalid/debian"],
                    "suites": ["stable"],
                    "signed-by": KEY_CONTENT,
                }
            ],
        }
    },
}


def make_config() -> Configuration:
    return Configuration.model_validate(SOURCE_CONTENT)


def make_config_with(source_extra: dict, name: str = "interdiode") -> Configuration:
    data = {
        "uuid": "12345678-1234-5678-1234-567812345678",
        "organisation": "InterDiode",
        "configurations": {
            "apt": {
                "sources": [
                    {
                        "name": name,
                        "uris": ["http://example.invalid/debian"],
                        "suites": ["stable"],
                        **source_extra,
                    }
                ],
            }
        },
    }
    return Configuration.model_validate(data)


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


def test_is_relevant_false_without_apt_dir(tmp_path) -> None:
    src = tmp_path / "src"
    src.mkdir()
    context = Context(config=make_config(), dst=tmp_path / "dst", src=src)
    assert AptModule().is_relevant(context) is False


def test_is_relevant_true_with_apt_dir(tmp_path) -> None:
    src = tmp_path / "src"
    (src / "etc/apt").mkdir(parents=True)
    context = Context(config=make_config(), dst=tmp_path / "dst", src=src)
    assert AptModule().is_relevant(context) is True


def test_apply_copies_existing_sources_unchanged_without_uri_to_replace(
    tmp_path,
) -> None:
    src = tmp_path / "src"
    apt_dir = src / "etc/apt"
    sources_list_d = apt_dir / "sources.list.d"
    sources_list_d.mkdir(parents=True)

    (apt_dir / "sources.list").write_text(
        "# already disabled\ndeb http://example.invalid/legacy stable main\n\n",
        encoding="utf-8",
    )
    (sources_list_d / "extra.list").write_text(
        "deb http://example.invalid/extra stable main\n", encoding="utf-8"
    )
    (sources_list_d / "docker.sources").write_text(
        "Types: deb\nURIs: http://example.invalid/docker\nSuites: stable\nEnabled: yes\n"
        "\n"
        "Types: deb\nURIs: http://example.invalid/docker2\nSuites: stable\n"
        "\n\n",
        encoding="utf-8",
    )
    (sources_list_d / "ignored.txt").write_text("not a source file\n", encoding="utf-8")

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

    AptModule().apply(context)

    # No configured source declares a uri_to_replace, so none of the existing
    # entries are disabled: they are copied through unchanged.
    top_level = (dst / "etc/apt/sources.list").read_text(encoding="utf-8")
    assert (
        top_level
        == "# already disabled\ndeb http://example.invalid/legacy stable main\n\n"
    )

    extra_list = (dst / "etc/apt/sources.list.d/extra.list").read_text(encoding="utf-8")
    assert extra_list == "deb http://example.invalid/extra stable main\n"

    docker_sources = (dst / "etc/apt/sources.list.d/docker.sources").read_text(
        encoding="utf-8"
    )
    assert docker_sources == (
        "Types: deb\nURIs: http://example.invalid/docker\nSuites: stable\nEnabled: yes"
        "\n\n"
        "Types: deb\nURIs: http://example.invalid/docker2\nSuites: stable\n"
    )

    assert not (dst / "etc/apt/sources.list.d/ignored.txt").exists()

    configured_source = (dst / "etc/apt/sources.list.d/interdiode.sources").read_text(
        encoding="utf-8"
    )
    assert configured_source == (
        "Types: deb\nURIs: http://example.invalid/debian\nSuites: stable\n"
        "Signed-By:\n"
        " -----BEGIN PGP PUBLIC KEY BLOCK-----\n"
        " .\n"
        " mQENBGqh08IBCAC0\n"
        " -----END PGP PUBLIC KEY BLOCK-----\n"
        "Enabled: yes\n"
    )


def test_apply_disables_only_existing_sources_matching_uri_to_replace(
    tmp_path,
) -> None:
    src = tmp_path / "src"
    apt_dir = src / "etc/apt"
    sources_list_d = apt_dir / "sources.list.d"
    sources_list_d.mkdir(parents=True)

    (apt_dir / "sources.list").write_text(
        "deb http://example.invalid/legacy stable main\n", encoding="utf-8"
    )
    (sources_list_d / "extra.list").write_text(
        "deb http://example.invalid/extra stable main\n", encoding="utf-8"
    )
    (sources_list_d / "docker.sources").write_text(
        "Types: deb\nURIs: http://example.invalid/extra\nSuites: stable\nEnabled: yes\n"
        "\n"
        "Types: deb\nURIs: http://example.invalid/docker2\nSuites: stable\n"
        "\n\n",
        encoding="utf-8",
    )

    dst = tmp_path / "dst"
    context = Context(
        config=make_config_with({"uri-to-replace": "http://example.invalid/extra"}),
        dst=dst,
        src=src,
    )

    AptModule().apply(context)

    # The legacy source doesn't match the pattern, so it stays enabled...
    top_level = (dst / "etc/apt/sources.list").read_text(encoding="utf-8")
    assert top_level == "deb http://example.invalid/legacy stable main\n"

    # ...but every entry whose URI matches is disabled, regardless of format.
    extra_list = (dst / "etc/apt/sources.list.d/extra.list").read_text(encoding="utf-8")
    assert extra_list == "# deb http://example.invalid/extra stable main\n"

    docker_sources = (dst / "etc/apt/sources.list.d/docker.sources").read_text(
        encoding="utf-8"
    )
    assert docker_sources == (
        "Types: deb\nURIs: http://example.invalid/extra\nSuites: stable\nEnabled: no"
        "\n\n"
        "Types: deb\nURIs: http://example.invalid/docker2\nSuites: stable\n"
    )


def test_apply_without_existing_top_level_sources_list(tmp_path) -> None:
    src = tmp_path / "src"
    (src / "etc/apt").mkdir(parents=True)
    dst = tmp_path / "dst"
    context = Context(config=make_config(), dst=dst, src=src)

    AptModule().apply(context)

    assert not (dst / "etc/apt/sources.list").exists()
    assert (dst / "etc/apt/sources.list.d/interdiode.sources").exists()


def test_apply_uses_signed_by_file_when_no_key_content(tmp_path) -> None:
    dst = tmp_path / "dst"
    context = Context(
        config=make_config_with({"signed-by-file": "/etc/apt/keyrings/existing.gpg"}),
        dst=dst,
        src=None,
    )

    AptModule().apply(context)

    configured_source = (dst / "etc/apt/sources.list.d/interdiode.sources").read_text(
        encoding="utf-8"
    )
    assert configured_source == (
        "Types: deb\nURIs: http://example.invalid/debian\nSuites: stable\n"
        "Signed-By: /etc/apt/keyrings/existing.gpg\nEnabled: yes\n"
    )


def test_apply_signed_by_takes_precedence_over_signed_by_file(tmp_path) -> None:
    dst = tmp_path / "dst"
    context = Context(
        config=make_config_with(
            {
                "signed-by": KEY_CONTENT,
                "signed-by-file": "/etc/apt/keyrings/existing.gpg",
            }
        ),
        dst=dst,
        src=None,
    )

    AptModule().apply(context)

    configured_source = (dst / "etc/apt/sources.list.d/interdiode.sources").read_text(
        encoding="utf-8"
    )
    assert "Signed-By:\n -----BEGIN PGP PUBLIC KEY BLOCK-----" in configured_source
    assert "existing.gpg" not in configured_source


def test_apply_without_signed_by_writes_no_field(tmp_path) -> None:
    dst = tmp_path / "dst"
    context = Context(config=make_config_with({}), dst=dst, src=None)

    AptModule().apply(context)

    configured_source = (dst / "etc/apt/sources.list.d/interdiode.sources").read_text(
        encoding="utf-8"
    )
    assert configured_source == (
        "Types: deb\nURIs: http://example.invalid/debian\nSuites: stable\nEnabled: yes\n"
    )


def test_apply_skips_source_when_uri_to_replace_matches_nothing(tmp_path) -> None:
    src = tmp_path / "src"
    apt_dir = src / "etc/apt"
    apt_dir.mkdir(parents=True)
    (apt_dir / "sources.list").write_text(
        "deb http://example.invalid/legacy stable main\n", encoding="utf-8"
    )

    dst = tmp_path / "dst"
    context = Context(
        config=make_config_with({"uri-to-replace": "http://other.invalid/*"}),
        dst=dst,
        src=src,
    )

    AptModule().apply(context)

    assert not (dst / "etc/apt/sources.list.d/interdiode.sources").exists()


def test_apply_writes_source_when_uri_to_replace_matches_existing_uri(
    tmp_path,
) -> None:
    src = tmp_path / "src"
    apt_dir = src / "etc/apt"
    apt_dir.mkdir(parents=True)
    (apt_dir / "sources.list").write_text(
        "deb http://legacy.invalid/debian stable main\n", encoding="utf-8"
    )

    dst = tmp_path / "dst"
    context = Context(
        config=make_config_with({"uri-to-replace": "http://legacy.invalid/*"}),
        dst=dst,
        src=src,
    )

    AptModule().apply(context)

    assert (dst / "etc/apt/sources.list.d/interdiode.sources").exists()


def test_apply_skips_source_when_uri_to_replace_set_without_src(tmp_path) -> None:
    dst = tmp_path / "dst"
    context = Context(
        config=make_config_with({"uri-to-replace": "http://*"}), dst=dst, src=None
    )

    AptModule().apply(context)

    assert not (dst / "etc/apt/sources.list.d/interdiode.sources").exists()


def test_apply_matches_uri_to_replace_across_list_line_variants(tmp_path) -> None:
    src = tmp_path / "src"
    apt_dir = src / "etc/apt"
    sources_list_d = apt_dir / "sources.list.d"
    sources_list_d.mkdir(parents=True)

    (apt_dir / "sources.list").write_text(
        "# a comment\n"
        "\n"
        "garbage line here\n"
        "deb [unterminated bracket with no closing marker\n"
        "deb [arch=amd64 trusted=yes] http://example.invalid/extra stable main\n",
        encoding="utf-8",
    )
    (sources_list_d / "ignored.txt").write_text("not a source file\n", encoding="utf-8")

    data = {
        "uuid": "12345678-1234-5678-1234-567812345678",
        "organisation": "InterDiode",
        "configurations": {
            "apt": {
                "sources": [
                    {
                        "name": "first",
                        "uris": ["http://example.invalid/first"],
                        "suites": ["stable"],
                        "uri-to-replace": "http://example.invalid/extra",
                    },
                    {
                        "name": "second",
                        "uris": ["http://example.invalid/second"],
                        "suites": ["stable"],
                        "uri-to-replace": "http://example.invalid/extra",
                    },
                ],
            }
        },
    }
    dst = tmp_path / "dst"
    context = Context(config=Configuration.model_validate(data), dst=dst, src=src)

    AptModule().apply(context)

    assert (dst / "etc/apt/sources.list.d/first.sources").exists()
    assert (dst / "etc/apt/sources.list.d/second.sources").exists()


def test_apply_sanitizes_source_name_with_slash(tmp_path) -> None:
    dst = tmp_path / "dst"
    context = Context(
        config=make_config_with({"signed-by": KEY_CONTENT}, name="../../evil"),
        dst=dst,
        src=None,
    )

    AptModule().apply(context)

    source_file = dst / "etc/apt/sources.list.d/______evil.sources"
    assert source_file.exists()
    assert not (dst / "etc/apt/sources.list.d/evil.sources").exists()