"Fossies" - the Fresh Open Source Software Archive

Member "salt-3002.2/tests/integration/states/test_pkgrepo.py" (18 Nov 2020, 13985 Bytes) of package /linux/misc/salt-3002.2.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_pkgrepo.py": 3002.1_vs_3002.2.

    1 """
    2 tests for pkgrepo states
    3 """
    4 
    5 
    6 import os
    7 
    8 import salt.utils.files
    9 import salt.utils.platform
   10 from tests.support.case import ModuleCase
   11 from tests.support.helpers import (
   12     destructiveTest,
   13     requires_salt_modules,
   14     requires_salt_states,
   15     requires_system_grains,
   16     runs_on,
   17     slowTest,
   18 )
   19 from tests.support.mixins import SaltReturnAssertsMixin
   20 from tests.support.pytest.helpers import temp_state_file
   21 from tests.support.unit import skipIf
   22 
   23 
   24 @destructiveTest
   25 @skipIf(salt.utils.platform.is_windows(), "minion is windows")
   26 class PkgrepoTest(ModuleCase, SaltReturnAssertsMixin):
   27     """
   28     pkgrepo state tests
   29     """
   30 
   31     @requires_salt_modules("pkgrepo.managed")
   32     @requires_system_grains
   33     def test_pkgrepo_01_managed(self, grains):
   34         """
   35         Test adding a repo
   36         """
   37         if grains["os"] == "Ubuntu" and grains["osrelease_info"] >= (15, 10):
   38             self.skipTest(
   39                 "The PPA used for this test does not exist for Ubuntu Wily"
   40                 " (15.10) and later."
   41             )
   42 
   43         if grains["os_family"] == "Debian":
   44             try:
   45                 from aptsources import sourceslist  # pylint: disable=unused-import
   46             except ImportError:
   47                 self.skipTest("aptsources.sourceslist python module not found")
   48         ret = self.run_function("state.sls", mods="pkgrepo.managed", timeout=120)
   49         # If the below assert fails then no states were run, and the SLS in
   50         # tests/integration/files/file/base/pkgrepo/managed.sls needs to be
   51         # corrected.
   52         self.assertReturnNonEmptySaltType(ret)
   53         for state_id, state_result in ret.items():
   54             self.assertSaltTrueReturn(dict([(state_id, state_result)]))
   55 
   56     @requires_salt_modules("pkgrepo.absent")
   57     @requires_system_grains
   58     def test_pkgrepo_02_absent(self, grains):
   59         """
   60         Test removing the repo from the above test
   61         """
   62         if grains["os"] == "Ubuntu" and grains["osrelease_info"] >= (15, 10):
   63             self.skipTest(
   64                 "The PPA used for this test does not exist for Ubuntu Wily"
   65                 " (15.10) and later."
   66             )
   67 
   68         ret = self.run_function("state.sls", mods="pkgrepo.absent", timeout=120)
   69         # If the below assert fails then no states were run, and the SLS in
   70         # tests/integration/files/file/base/pkgrepo/absent.sls needs to be
   71         # corrected.
   72         self.assertReturnNonEmptySaltType(ret)
   73         for state_id, state_result in ret.items():
   74             self.assertSaltTrueReturn(dict([(state_id, state_result)]))
   75 
   76     @requires_salt_states("pkgrepo.absent", "pkgrepo.managed")
   77     @requires_system_grains
   78     @slowTest
   79     def test_pkgrepo_03_with_comments(self, grains):
   80         """
   81         Test adding a repo with comments
   82         """
   83         kwargs = {}
   84         if grains["os_family"] == "RedHat":
   85             kwargs = {
   86                 "name": "examplerepo",
   87                 "baseurl": "http://example.com/repo",
   88                 "enabled": False,
   89                 "comments": ["This is a comment"],
   90             }
   91         else:
   92             self.skipTest(
   93                 "{}/{} test case needed".format(grains["os_family"], grains["os"])
   94             )
   95 
   96         try:
   97             # Run the state to add the repo
   98             ret = self.run_state("pkgrepo.managed", **kwargs)
   99             self.assertSaltTrueReturn(ret)
  100 
  101             # Run again with modified comments
  102             kwargs["comments"].append("This is another comment")
  103             ret = self.run_state("pkgrepo.managed", **kwargs)
  104             self.assertSaltTrueReturn(ret)
  105             ret = ret[next(iter(ret))]
  106             self.assertEqual(
  107                 ret["changes"],
  108                 {
  109                     "comments": {
  110                         "old": ["This is a comment"],
  111                         "new": ["This is a comment", "This is another comment"],
  112                     }
  113                 },
  114             )
  115 
  116             # Run a third time, no changes should be made
  117             ret = self.run_state("pkgrepo.managed", **kwargs)
  118             self.assertSaltTrueReturn(ret)
  119             ret = ret[next(iter(ret))]
  120             self.assertFalse(ret["changes"])
  121             self.assertEqual(
  122                 ret["comment"],
  123                 "Package repo '{}' already configured".format(kwargs["name"]),
  124             )
  125         finally:
  126             # Clean up
  127             self.run_state("pkgrepo.absent", name=kwargs["name"])
  128 
  129     @requires_salt_states("pkgrepo.managed")
  130     @requires_system_grains
  131     @slowTest
  132     def test_pkgrepo_04_apt_with_architectures(self, grains):
  133         """
  134         Test managing a repo with architectures specified
  135         """
  136         if grains["os_family"].lower() != "debian":
  137             self.skipTest("APT-only test")
  138 
  139         name = "deb {{arch}}http://foo.com/bar/latest {oscodename} main".format(
  140             oscodename=grains["oscodename"]
  141         )
  142 
  143         def _get_arch(arch):
  144             return "[arch={}] ".format(arch) if arch else ""
  145 
  146         def _run(arch="", test=False):
  147             ret = self.run_state(
  148                 "pkgrepo.managed",
  149                 name=name.format(arch=_get_arch(arch)),
  150                 file=fn_,
  151                 refresh=False,
  152                 test=test,
  153             )
  154             return ret[next(iter(ret))]
  155 
  156         fn_ = salt.utils.files.mkstemp(dir="/etc/apt/sources.list.d", suffix=".list")
  157 
  158         try:
  159             # Run with test=True
  160             ret = _run(test=True)
  161             assert ret["changes"] == {"repo": name.format(arch="")}, ret["changes"]
  162             assert "would be" in ret["comment"], ret["comment"]
  163             assert ret["result"] is None, ret["result"]
  164 
  165             # Run for real
  166             ret = _run()
  167             assert ret["changes"] == {"repo": name.format(arch="")}, ret["changes"]
  168             assert ret["comment"].startswith("Configured"), ret["comment"]
  169             assert ret["result"] is True, ret["result"]
  170 
  171             # Run again with test=True, should exit with no changes and a True
  172             # result.
  173             ret = _run(test=True)
  174             assert not ret["changes"], ret["changes"]
  175             assert "already" in ret["comment"], ret["comment"]
  176             assert ret["result"] is True, ret["result"]
  177 
  178             # Run for real again, results should be the same as above (i.e. we
  179             # should never get to the point where we exit with a None result).
  180             ret = _run()
  181             assert not ret["changes"], ret["changes"]
  182             assert "already" in ret["comment"], ret["comment"]
  183             assert ret["result"] is True, ret["result"]
  184 
  185             expected_changes = {
  186                 "line": {
  187                     "new": name.format(arch=_get_arch("amd64")),
  188                     "old": name.format(arch=""),
  189                 },
  190                 "architectures": {"new": ["amd64"], "old": []},
  191             }
  192 
  193             # Run with test=True and the architecture set. We should get a None
  194             # result with some expected changes.
  195             ret = _run(arch="amd64", test=True)
  196             assert ret["changes"] == expected_changes, ret["changes"]
  197             assert "would be" in ret["comment"], ret["comment"]
  198             assert ret["result"] is None, ret["result"]
  199 
  200             # Run for real, with the architecture set. We should get a True
  201             # result with the same changes.
  202             ret = _run(arch="amd64")
  203             assert ret["changes"] == expected_changes, ret["changes"]
  204             assert ret["comment"].startswith("Configured"), ret["comment"]
  205             assert ret["result"] is True, ret["result"]
  206 
  207             # Run again with test=True, should exit with no changes and a True
  208             # result.
  209             ret = _run(arch="amd64", test=True)
  210             assert not ret["changes"], ret["changes"]
  211             assert "already" in ret["comment"], ret["comment"]
  212             assert ret["result"] is True, ret["result"]
  213 
  214             # Run for real again, results should be the same as above (i.e. we
  215             # should never get to the point where we exit with a None result).
  216             ret = _run(arch="amd64")
  217             assert not ret["changes"], ret["changes"]
  218             assert "already" in ret["comment"], ret["comment"]
  219             assert ret["result"] is True, ret["result"]
  220 
  221             expected_changes = {
  222                 "line": {
  223                     "new": name.format(arch=""),
  224                     "old": name.format(arch=_get_arch("amd64")),
  225                 },
  226                 "architectures": {"new": [], "old": ["amd64"]},
  227             }
  228 
  229             # Run with test=True and the architecture set back to the original
  230             # value. We should get a None result with some expected changes.
  231             ret = _run(test=True)
  232             assert ret["changes"] == expected_changes, ret["changes"]
  233             assert "would be" in ret["comment"], ret["comment"]
  234             assert ret["result"] is None, ret["result"]
  235 
  236             # Run for real, with the architecture set. We should get a True
  237             # result with the same changes.
  238             ret = _run()
  239             assert ret["changes"] == expected_changes, ret["changes"]
  240             assert ret["comment"].startswith("Configured"), ret["comment"]
  241             assert ret["result"] is True, ret["result"]
  242 
  243             # Run again with test=True, should exit with no changes and a True
  244             # result.
  245             ret = _run(test=True)
  246             assert not ret["changes"], ret["changes"]
  247             assert "already" in ret["comment"], ret["comment"]
  248             assert ret["result"] is True, ret["result"]
  249 
  250             # Run for real again, results should be the same as above (i.e. we
  251             # should never get to the point where we exit with a None result).
  252             ret = _run()
  253             assert not ret["changes"], ret["changes"]
  254             assert "already" in ret["comment"], ret["comment"]
  255             assert ret["result"] is True, ret["result"]
  256         finally:
  257             try:
  258                 os.remove(fn_)
  259             except OSError:
  260                 pass
  261 
  262     @requires_salt_states("pkgrepo.absent", "pkgrepo.managed")
  263     @requires_system_grains
  264     @slowTest
  265     def test_pkgrepo_05_copr_with_comments(self, grains):
  266         """
  267         Test copr
  268         """
  269         kwargs = {}
  270         if grains["os_family"] == "RedHat":
  271             if (
  272                 grains["osfinger"] == "CentOS Linux-7"
  273                 or grains["osfinger"] == "Amazon Linux-2"
  274             ):
  275                 self.skipTest("copr plugin not installed on Centos 7 CI")
  276             kwargs = {
  277                 "name": "hello-copr",
  278                 "copr": "mymindstorm/hello",
  279                 "enabled": False,
  280                 "comments": ["This is a comment"],
  281             }
  282         else:
  283             self.skipTest(
  284                 "{}/{} test case needed".format(grains["os_family"], grains["os"])
  285             )
  286 
  287         try:
  288             # Run the state to add the repo
  289             ret = self.run_state("pkgrepo.managed", **kwargs)
  290             self.assertSaltTrueReturn(ret)
  291 
  292             # Run again with modified comments
  293             kwargs["comments"].append("This is another comment")
  294             ret = self.run_state("pkgrepo.managed", **kwargs)
  295             self.assertSaltTrueReturn(ret)
  296             ret = ret[next(iter(ret))]
  297             self.assertEqual(
  298                 ret["changes"],
  299                 {
  300                     "comments": {
  301                         "old": ["This is a comment"],
  302                         "new": ["This is a comment", "This is another comment"],
  303                     }
  304                 },
  305             )
  306 
  307             # Run a third time, no changes should be made
  308             ret = self.run_state("pkgrepo.managed", **kwargs)
  309             self.assertSaltTrueReturn(ret)
  310             ret = ret[next(iter(ret))]
  311             self.assertFalse(ret["changes"])
  312             self.assertEqual(
  313                 ret["comment"],
  314                 "Package repo '{}' already configured".format(kwargs["name"]),
  315             )
  316         finally:
  317             # Clean up
  318             self.run_state("pkgrepo.absent", copr=kwargs["copr"])
  319 
  320     @runs_on(kernel="linux", os="Ubuntu")
  321     def test_managed_multiple_comps(self):
  322         state_file = """
  323         ubuntu-backports:
  324           pkgrepo.managed:
  325             - name: 'deb http://fi.archive.ubuntu.com/ubuntu focal-backports'
  326             - comps: main, restricted, universe, multiverse
  327             - refresh: false
  328             - disabled: false
  329             - clean_file: true
  330             - file: /etc/apt/sources.list.d/99-salt-archive-ubuntu-focal-backports.list
  331             - require_in:
  332               - pkgrepo: canonical-ubuntu
  333 
  334         canonical-ubuntu:
  335           pkgrepo.managed:
  336             - name: 'deb http://archive.canonical.com/ubuntu {{ salt['grains.get']('oscodename') }}'
  337             - comps: partner
  338             - refresh: false
  339             - disabled: false
  340             - clean_file: true
  341             - file: /etc/apt/sources.list.d/99-salt-canonical-ubuntu.list
  342         """
  343 
  344         def remove_apt_list_file(path):
  345             if os.path.exists(path):
  346                 os.unlink(path)
  347 
  348         self.addCleanup(
  349             remove_apt_list_file,
  350             "/etc/apt/sources.list.d/99-salt-canonical-ubuntu.list",
  351         )
  352         self.addCleanup(
  353             remove_apt_list_file,
  354             "/etc/apt/sources.list.d/99-salt-archive-ubuntu-focal-backports.list",
  355         )
  356         with temp_state_file("multiple-comps-repos.sls", state_file):
  357             ret = self.run_function("state.sls", ["multiple-comps-repos"])
  358             for state_run in ret.values():
  359                 # On the first run, we must have changes
  360                 assert state_run["changes"]
  361             ret = self.run_function("state.sls", ["multiple-comps-repos"])
  362             for state_run in ret.values():
  363                 # On the second run though, we shouldn't have changes made
  364                 assert not state_run["changes"]