salt/tests/integration/states/test_pkgrepo.py

"""
tests for pkgrepo states
"""


import os

import salt.utils.files
import salt.utils.platform
from tests.support.case import ModuleCase
from tests.support.helpers import (
    destructiveTest,
    requires_salt_modules,
    requires_salt_states,
    requires_system_grains,
    runs_on,
    slowTest,
)
from tests.support.mixins import SaltReturnAssertsMixin
from tests.support.pytest.helpers import temp_state_file
from tests.support.unit import skipIf


@destructiveTest
@skipIf(salt.utils.platform.is_windows(), "minion is windows")
class PkgrepoTest(ModuleCase, SaltReturnAssertsMixin):
    """
    pkgrepo state tests
    """

    @requires_salt_modules("pkgrepo.managed")
    @requires_system_grains
    def test_pkgrepo_01_managed(self, grains):
        """
        Test adding a repo
        """
        if grains["os"] == "Ubuntu" and grains["osrelease_info"] >= (15, 10):
            self.skipTest(
                "The PPA used for this test does not exist for Ubuntu Wily"
                " (15.10) and later."
            )

        if grains["os_family"] == "Debian":
            try:
                from aptsources import sourceslist  # pylint: disable=unused-import
            except ImportError:
                self.skipTest("aptsources.sourceslist python module not found")
        ret = self.run_function("state.sls", mods="pkgrepo.managed", timeout=120)
        # If the below assert fails then no states were run, and the SLS in
        # tests/integration/files/file/base/pkgrepo/managed.sls needs to be
        # corrected.
        self.assertReturnNonEmptySaltType(ret)
        for state_id, state_result in ret.items():
            self.assertSaltTrueReturn(dict([(state_id, state_result)]))

    @requires_salt_modules("pkgrepo.absent")
    @requires_system_grains
    def test_pkgrepo_02_absent(self, grains):
        """
        Test removing the repo from the above test
        """
        if grains["os"] == "Ubuntu" and grains["osrelease_info"] >= (15, 10):
            self.skipTest(
                "The PPA used for this test does not exist for Ubuntu Wily"
                " (15.10) and later."
            )

        ret = self.run_function("state.sls", mods="pkgrepo.absent", timeout=120)
        # If the below assert fails then no states were run, and the SLS in
        # tests/integration/files/file/base/pkgrepo/absent.sls needs to be
        # corrected.
        self.assertReturnNonEmptySaltType(ret)
        for state_id, state_result in ret.items():
            self.assertSaltTrueReturn(dict([(state_id, state_result)]))

    @requires_salt_states("pkgrepo.absent", "pkgrepo.managed")
    @requires_system_grains
    @slowTest
    def test_pkgrepo_03_with_comments(self, grains):
        """
        Test adding a repo with comments
        """
        kwargs = {}
        if grains["os_family"] == "RedHat":
            kwargs = {
                "name": "examplerepo",
                "baseurl": "http://example.com/repo",
                "enabled": False,
                "comments": ["This is a comment"],
            }
        else:
            self.skipTest(
                "{}/{} test case needed".format(grains["os_family"], grains["os"])
            )

        try:
            # Run the state to add the repo
            ret = self.run_state("pkgrepo.managed", **kwargs)
            self.assertSaltTrueReturn(ret)

            # Run again with modified comments
            kwargs["comments"].append("This is another comment")
            ret = self.run_state("pkgrepo.managed", **kwargs)
            self.assertSaltTrueReturn(ret)
            ret = ret[next(iter(ret))]
            self.assertEqual(
                ret["changes"],
                {
                    "comments": {
                        "old": ["This is a comment"],
                        "new": ["This is a comment", "This is another comment"],
                    }
                },
            )

            # Run a third time, no changes should be made
            ret = self.run_state("pkgrepo.managed", **kwargs)
            self.assertSaltTrueReturn(ret)
            ret = ret[next(iter(ret))]
            self.assertFalse(ret["changes"])
            self.assertEqual(
                ret["comment"],
                "Package repo '{}' already configured".format(kwargs["name"]),
            )
        finally:
            # Clean up
            self.run_state("pkgrepo.absent", name=kwargs["name"])

    @requires_salt_states("pkgrepo.managed")
    @requires_system_grains
    @slowTest
    def test_pkgrepo_04_apt_with_architectures(self, grains):
        """
        Test managing a repo with architectures specified
        """
        if grains["os_family"].lower() != "debian":
            self.skipTest("APT-only test")

        name = "deb {{arch}}http://foo.com/bar/latest {oscodename} main".format(
            oscodename=grains["oscodename"]
        )

        def _get_arch(arch):
            return "[arch={}] ".format(arch) if arch else ""

        def _run(arch="", test=False):
            ret = self.run_state(
                "pkgrepo.managed",
                name=name.format(arch=_get_arch(arch)),
                file=fn_,
                refresh=False,
                test=test,
            )
            return ret[next(iter(ret))]

        fn_ = salt.utils.files.mkstemp(dir="/etc/apt/sources.list.d", suffix=".list")

        try:
            # Run with test=True
            ret = _run(test=True)
            assert ret["changes"] == {"repo": name.format(arch="")}, ret["changes"]
            assert "would be" in ret["comment"], ret["comment"]
            assert ret["result"] is None, ret["result"]

            # Run for real
            ret = _run()
            assert ret["changes"] == {"repo": name.format(arch="")}, ret["changes"]
            assert ret["comment"].startswith("Configured"), ret["comment"]
            assert ret["result"] is True, ret["result"]

            # Run again with test=True, should exit with no changes and a True
            # result.
            ret = _run(test=True)
            assert not ret["changes"], ret["changes"]
            assert "already" in ret["comment"], ret["comment"]
            assert ret["result"] is True, ret["result"]

            # Run for real again, results should be the same as above (i.e. we
            # should never get to the point where we exit with a None result).
            ret = _run()
            assert not ret["changes"], ret["changes"]
            assert "already" in ret["comment"], ret["comment"]
            assert ret["result"] is True, ret["result"]

            expected_changes = {
                "line": {
                    "new": name.format(arch=_get_arch("amd64")),
                    "old": name.format(arch=""),
                },
                "architectures": {"new": ["amd64"], "old": []},
            }

            # Run with test=True and the architecture set. We should get a None
            # result with some expected changes.
            ret = _run(arch="amd64", test=True)
            assert ret["changes"] == expected_changes, ret["changes"]
            assert "would be" in ret["comment"], ret["comment"]
            assert ret["result"] is None, ret["result"]

            # Run for real, with the architecture set. We should get a True
            # result with the same changes.
            ret = _run(arch="amd64")
            assert ret["changes"] == expected_changes, ret["changes"]
            assert ret["comment"].startswith("Configured"), ret["comment"]
            assert ret["result"] is True, ret["result"]

            # Run again with test=True, should exit with no changes and a True
            # result.
            ret = _run(arch="amd64", test=True)
            assert not ret["changes"], ret["changes"]
            assert "already" in ret["comment"], ret["comment"]
            assert ret["result"] is True, ret["result"]

            # Run for real again, results should be the same as above (i.e. we
            # should never get to the point where we exit with a None result).
            ret = _run(arch="amd64")
            assert not ret["changes"], ret["changes"]
            assert "already" in ret["comment"], ret["comment"]
            assert ret["result"] is True, ret["result"]

            expected_changes = {
                "line": {
                    "new": name.format(arch=""),
                    "old": name.format(arch=_get_arch("amd64")),
                },
                "architectures": {"new": [], "old": ["amd64"]},
            }

            # Run with test=True and the architecture set back to the original
            # value. We should get a None result with some expected changes.
            ret = _run(test=True)
            assert ret["changes"] == expected_changes, ret["changes"]
            assert "would be" in ret["comment"], ret["comment"]
            assert ret["result"] is None, ret["result"]

            # Run for real, with the architecture set. We should get a True
            # result with the same changes.
            ret = _run()
            assert ret["changes"] == expected_changes, ret["changes"]
            assert ret["comment"].startswith("Configured"), ret["comment"]
            assert ret["result"] is True, ret["result"]

            # Run again with test=True, should exit with no changes and a True
            # result.
            ret = _run(test=True)
            assert not ret["changes"], ret["changes"]
            assert "already" in ret["comment"], ret["comment"]
            assert ret["result"] is True, ret["result"]

            # Run for real again, results should be the same as above (i.e. we
            # should never get to the point where we exit with a None result).
            ret = _run()
            assert not ret["changes"], ret["changes"]
            assert "already" in ret["comment"], ret["comment"]
            assert ret["result"] is True, ret["result"]
        finally:
            try:
                os.remove(fn_)
            except OSError:
                pass

    @requires_salt_states("pkgrepo.absent", "pkgrepo.managed")
    @requires_system_grains
    @slowTest
    def test_pkgrepo_05_copr_with_comments(self, grains):
        """
        Test copr
        """
        kwargs = {}
        if grains["os_family"] == "RedHat":
            if (
                grains["osfinger"] == "CentOS Linux-7"
                or grains["osfinger"] == "Amazon Linux-2"
            ):
                self.skipTest("copr plugin not installed on Centos 7 CI")
            kwargs = {
                "name": "hello-copr",
                "copr": "mymindstorm/hello",
                "enabled": False,
                "comments": ["This is a comment"],
            }
        else:
            self.skipTest(
                "{}/{} test case needed".format(grains["os_family"], grains["os"])
            )

        try:
            # Run the state to add the repo
            ret = self.run_state("pkgrepo.managed", **kwargs)
            self.assertSaltTrueReturn(ret)

            # Run again with modified comments
            kwargs["comments"].append("This is another comment")
            ret = self.run_state("pkgrepo.managed", **kwargs)
            self.assertSaltTrueReturn(ret)
            ret = ret[next(iter(ret))]
            self.assertEqual(
                ret["changes"],
                {
                    "comments": {
                        "old": ["This is a comment"],
                        "new": ["This is a comment", "This is another comment"],
                    }
                },
            )

            # Run a third time, no changes should be made
            ret = self.run_state("pkgrepo.managed", **kwargs)
            self.assertSaltTrueReturn(ret)
            ret = ret[next(iter(ret))]
            self.assertFalse(ret["changes"])
            self.assertEqual(
                ret["comment"],
                "Package repo '{}' already configured".format(kwargs["name"]),
            )
        finally:
            # Clean up
            self.run_state("pkgrepo.absent", copr=kwargs["copr"])

    @runs_on(kernel="linux", os="Ubuntu")
    def test_managed_multiple_comps(self):
        state_file = """
        ubuntu-backports:
          pkgrepo.managed:
            - name: 'deb http://fi.archive.ubuntu.com/ubuntu focal-backports'
            - comps: main, restricted, universe, multiverse
            - refresh: false
            - disabled: false
            - clean_file: true
            - file: /etc/apt/sources.list.d/99-salt-archive-ubuntu-focal-backports.list
            - require_in:
              - pkgrepo: canonical-ubuntu

        canonical-ubuntu:
          pkgrepo.managed:
            - name: 'deb http://archive.canonical.com/ubuntu {{ salt['grains.get']('oscodename') }}'
            - comps: partner
            - refresh: false
            - disabled: false
            - clean_file: true
            - file: /etc/apt/sources.list.d/99-salt-canonical-ubuntu.list
        """

        def remove_apt_list_file(path):
            if os.path.exists(path):
                os.unlink(path)

        self.addCleanup(
            remove_apt_list_file,
            "/etc/apt/sources.list.d/99-salt-canonical-ubuntu.list",
        )
        self.addCleanup(
            remove_apt_list_file,
            "/etc/apt/sources.list.d/99-salt-archive-ubuntu-focal-backports.list",
        )
        with temp_state_file("multiple-comps-repos.sls", state_file):
            ret = self.run_function("state.sls", ["multiple-comps-repos"])
            for state_run in ret.values():
                # On the first run, we must have changes
                assert state_run["changes"]
            ret = self.run_function("state.sls", ["multiple-comps-repos"])
            for state_run in ret.values():
                # On the second run though, we shouldn't have changes made
                assert not state_run["changes"]