"Fossies" - the Fresh Open Source Software Archive

Member "salt-3002.2/tests/integration/modules/test_pkg.py" (18 Nov 2020, 18524 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_pkg.py": 3002.1_vs_3002.2.

    1 import os
    2 import pprint
    3 
    4 import pytest
    5 import salt.utils.path
    6 import salt.utils.pkg
    7 import salt.utils.platform
    8 from tests.support.case import ModuleCase
    9 from tests.support.helpers import (
   10     destructiveTest,
   11     requires_network,
   12     requires_salt_modules,
   13     requires_salt_states,
   14     requires_system_grains,
   15     skip_if_not_root,
   16     slowTest,
   17 )
   18 from tests.support.mixins import SaltReturnAssertsMixin
   19 from tests.support.unit import skipIf
   20 
   21 
   22 @pytest.mark.windows_whitelisted
   23 class PkgModuleTest(ModuleCase, SaltReturnAssertsMixin):
   24     """
   25     Validate the pkg module
   26     """
   27 
   28     @classmethod
   29     @requires_system_grains
   30     def setUpClass(cls, grains):  # pylint: disable=arguments-differ
   31         cls.ctx = {}
   32         cls.pkg = "figlet"
   33         if salt.utils.platform.is_windows():
   34             cls.pkg = "putty"
   35         elif grains["os_family"] == "RedHat":
   36             cls.pkg = "units"
   37 
   38     @skip_if_not_root
   39     @requires_salt_modules("pkg.refresh_db")
   40     def setUp(self):
   41         if "refresh" not in self.ctx:
   42             self.run_function("pkg.refresh_db")
   43             self.ctx["refresh"] = True
   44 
   45     @requires_salt_modules("pkg.list_pkgs")
   46     @slowTest
   47     def test_list(self):
   48         """
   49         verify that packages are installed
   50         """
   51         ret = self.run_function("pkg.list_pkgs")
   52         self.assertNotEqual(len(ret.keys()), 0)
   53 
   54     @requires_salt_modules("pkg.version_cmp")
   55     @requires_system_grains
   56     @slowTest
   57     def test_version_cmp(self, grains):
   58         """
   59         test package version comparison on supported platforms
   60         """
   61         func = "pkg.version_cmp"
   62         if grains["os_family"] == "Debian":
   63             lt = ["0.2.4-0ubuntu1", "0.2.4.1-0ubuntu1"]
   64             eq = ["0.2.4-0ubuntu1", "0.2.4-0ubuntu1"]
   65             gt = ["0.2.4.1-0ubuntu1", "0.2.4-0ubuntu1"]
   66         elif grains["os_family"] == "Suse":
   67             lt = ["2.3.0-1", "2.3.1-15.1"]
   68             eq = ["2.3.1-15.1", "2.3.1-15.1"]
   69             gt = ["2.3.2-15.1", "2.3.1-15.1"]
   70         else:
   71             lt = ["2.3.0", "2.3.1"]
   72             eq = ["2.3.1", "2.3.1"]
   73             gt = ["2.3.2", "2.3.1"]
   74 
   75         self.assertEqual(self.run_function(func, lt), -1)
   76         self.assertEqual(self.run_function(func, eq), 0)
   77         self.assertEqual(self.run_function(func, gt), 1)
   78 
   79     @destructiveTest
   80     @requires_salt_modules("pkg.mod_repo", "pkg.del_repo", "pkg.get_repo")
   81     @requires_network()
   82     @requires_system_grains
   83     @slowTest
   84     def test_mod_del_repo(self, grains):
   85         """
   86         test modifying and deleting a software repository
   87         """
   88         repo = None
   89 
   90         try:
   91             if grains["os"] == "Ubuntu":
   92                 repo = "ppa:otto-kesselgulasch/gimp-edge"
   93                 uri = "http://ppa.launchpad.net/otto-kesselgulasch/gimp-edge/ubuntu"
   94                 ret = self.run_function("pkg.mod_repo", [repo, "comps=main"])
   95                 self.assertNotEqual(ret, {})
   96                 ret = self.run_function("pkg.get_repo", [repo])
   97 
   98                 self.assertIsInstance(
   99                     ret,
  100                     dict,
  101                     "The 'pkg.get_repo' command did not return the excepted dictionary. "
  102                     "Output:\n{}".format(ret),
  103                 )
  104                 self.assertEqual(
  105                     ret["uri"],
  106                     uri,
  107                     msg="The URI did not match. Full return:\n{}".format(
  108                         pprint.pformat(ret)
  109                     ),
  110                 )
  111             elif grains["os_family"] == "RedHat":
  112                 repo = "saltstack"
  113                 name = "SaltStack repo for RHEL/CentOS {}".format(
  114                     grains["osmajorrelease"]
  115                 )
  116                 baseurl = "http://repo.saltstack.com/yum/redhat/{}/x86_64/latest/".format(
  117                     grains["osmajorrelease"]
  118                 )
  119                 gpgkey = "https://repo.saltstack.com/yum/rhel{}/SALTSTACK-GPG-KEY.pub".format(
  120                     grains["osmajorrelease"]
  121                 )
  122                 gpgcheck = 1
  123                 enabled = 1
  124                 ret = self.run_function(
  125                     "pkg.mod_repo",
  126                     [repo],
  127                     name=name,
  128                     baseurl=baseurl,
  129                     gpgkey=gpgkey,
  130                     gpgcheck=gpgcheck,
  131                     enabled=enabled,
  132                 )
  133                 # return data from pkg.mod_repo contains the file modified at
  134                 # the top level, so use next(iter(ret)) to get that key
  135                 self.assertNotEqual(ret, {})
  136                 repo_info = ret[next(iter(ret))]
  137                 self.assertIn(repo, repo_info)
  138                 self.assertEqual(repo_info[repo]["baseurl"], baseurl)
  139                 ret = self.run_function("pkg.get_repo", [repo])
  140                 self.assertEqual(ret["baseurl"], baseurl)
  141         finally:
  142             if repo is not None:
  143                 self.run_function("pkg.del_repo", [repo])
  144 
  145     @slowTest
  146     def test_mod_del_repo_multiline_values(self):
  147         """
  148         test modifying and deleting a software repository defined with multiline values
  149         """
  150         os_grain = self.run_function("grains.item", ["os"])["os"]
  151         repo = None
  152         try:
  153             if os_grain in ["CentOS", "RedHat"]:
  154                 my_baseurl = (
  155                     "http://my.fake.repo/foo/bar/\n http://my.fake.repo.alt/foo/bar/"
  156                 )
  157                 expected_get_repo_baseurl = (
  158                     "http://my.fake.repo/foo/bar/\nhttp://my.fake.repo.alt/foo/bar/"
  159                 )
  160                 major_release = int(
  161                     self.run_function("grains.item", ["osmajorrelease"])[
  162                         "osmajorrelease"
  163                     ]
  164                 )
  165                 repo = "fakerepo"
  166                 name = "Fake repo for RHEL/CentOS/SUSE"
  167                 baseurl = my_baseurl
  168                 gpgkey = "https://my.fake.repo/foo/bar/MY-GPG-KEY.pub"
  169                 failovermethod = "priority"
  170                 gpgcheck = 1
  171                 enabled = 1
  172                 ret = self.run_function(
  173                     "pkg.mod_repo",
  174                     [repo],
  175                     name=name,
  176                     baseurl=baseurl,
  177                     gpgkey=gpgkey,
  178                     gpgcheck=gpgcheck,
  179                     enabled=enabled,
  180                     failovermethod=failovermethod,
  181                 )
  182                 # return data from pkg.mod_repo contains the file modified at
  183                 # the top level, so use next(iter(ret)) to get that key
  184                 self.assertNotEqual(ret, {})
  185                 repo_info = ret[next(iter(ret))]
  186                 self.assertIn(repo, repo_info)
  187                 self.assertEqual(repo_info[repo]["baseurl"], my_baseurl)
  188                 ret = self.run_function("pkg.get_repo", [repo])
  189                 self.assertEqual(ret["baseurl"], expected_get_repo_baseurl)
  190                 self.run_function("pkg.mod_repo", [repo])
  191                 ret = self.run_function("pkg.get_repo", [repo])
  192                 self.assertEqual(ret["baseurl"], expected_get_repo_baseurl)
  193         finally:
  194             if repo is not None:
  195                 self.run_function("pkg.del_repo", [repo])
  196 
  197     @requires_salt_modules("pkg.owner")
  198     def test_owner(self):
  199         """
  200         test finding the package owning a file
  201         """
  202         func = "pkg.owner"
  203         ret = self.run_function(func, ["/bin/ls"])
  204         self.assertNotEqual(len(ret), 0)
  205 
  206     # Similar to pkg.owner, but for FreeBSD's pkgng
  207     @requires_salt_modules("pkg.which")
  208     def test_which(self):
  209         """
  210         test finding the package owning a file
  211         """
  212         func = "pkg.which"
  213         ret = self.run_function(func, ["/usr/local/bin/salt-call"])
  214         self.assertNotEqual(len(ret), 0)
  215 
  216     @destructiveTest
  217     @requires_salt_modules("pkg.version", "pkg.install", "pkg.remove")
  218     @requires_network()
  219     @slowTest
  220     def test_install_remove(self):
  221         """
  222         successfully install and uninstall a package
  223         """
  224         version = self.run_function("pkg.version", [self.pkg])
  225 
  226         def test_install():
  227             install_ret = self.run_function("pkg.install", [self.pkg])
  228             self.assertIn(self.pkg, install_ret)
  229 
  230         def test_remove():
  231             remove_ret = self.run_function("pkg.remove", [self.pkg])
  232             self.assertIn(self.pkg, remove_ret)
  233 
  234         if version and isinstance(version, dict):
  235             version = version[self.pkg]
  236 
  237         if version:
  238             test_remove()
  239             test_install()
  240         else:
  241             test_install()
  242             test_remove()
  243 
  244     @destructiveTest
  245     @requires_salt_modules(
  246         "pkg.hold",
  247         "pkg.unhold",
  248         "pkg.install",
  249         "pkg.version",
  250         "pkg.remove",
  251         "pkg.list_pkgs",
  252     )
  253     @requires_salt_states("pkg.installed")
  254     @requires_network()
  255     @requires_system_grains
  256     @slowTest
  257     def test_hold_unhold(self, grains):
  258         """
  259         test holding and unholding a package
  260         """
  261         versionlock_pkg = None
  262         if grains["os_family"] == "RedHat":
  263             pkgs = {
  264                 p for p in self.run_function("pkg.list_pkgs") if "-versionlock" in p
  265             }
  266             if not pkgs:
  267                 self.skipTest("No versionlock package found in repositories")
  268             for versionlock_pkg in pkgs:
  269                 ret = self.run_state(
  270                     "pkg.installed", name=versionlock_pkg, refresh=False
  271                 )
  272                 # Exit loop if a versionlock package installed correctly
  273                 try:
  274                     self.assertSaltTrueReturn(ret)
  275                     break
  276                 except AssertionError:
  277                     pass
  278             else:
  279                 self.fail("Could not install versionlock package from {}".format(pkgs))
  280 
  281         self.run_function("pkg.install", [self.pkg])
  282 
  283         try:
  284             hold_ret = self.run_function("pkg.hold", [self.pkg])
  285             if versionlock_pkg and "-versionlock is not installed" in str(hold_ret):
  286                 self.skipTest("{}  `{}` is installed".format(hold_ret, versionlock_pkg))
  287             self.assertIn(self.pkg, hold_ret)
  288             self.assertTrue(hold_ret[self.pkg]["result"])
  289 
  290             unhold_ret = self.run_function("pkg.unhold", [self.pkg])
  291             self.assertIn(self.pkg, unhold_ret)
  292             self.assertTrue(unhold_ret[self.pkg]["result"])
  293             self.run_function("pkg.remove", [self.pkg])
  294         finally:
  295             if versionlock_pkg:
  296                 ret = self.run_state("pkg.removed", name=versionlock_pkg)
  297                 self.assertSaltTrueReturn(ret)
  298 
  299     @destructiveTest
  300     @requires_salt_modules("pkg.refresh_db")
  301     @requires_network()
  302     @requires_system_grains
  303     @slowTest
  304     def test_refresh_db(self, grains):
  305         """
  306         test refreshing the package database
  307         """
  308         func = "pkg.refresh_db"
  309 
  310         rtag = salt.utils.pkg.rtag(self.minion_opts)
  311         salt.utils.pkg.write_rtag(self.minion_opts)
  312         self.assertTrue(os.path.isfile(rtag))
  313 
  314         ret = self.run_function(func)
  315         if not isinstance(ret, dict):
  316             self.skipTest(
  317                 "Upstream repo did not return coherent results: {}".format(ret)
  318             )
  319 
  320         if grains["os_family"] == "RedHat":
  321             self.assertIn(ret, (True, None))
  322         elif grains["os_family"] == "Suse":
  323             if not isinstance(ret, dict):
  324                 self.skipTest(
  325                     "Upstream repo did not return coherent results. Skipping test."
  326                 )
  327             self.assertNotEqual(ret, {})
  328             for source, state in ret.items():
  329                 self.assertIn(state, (True, False, None))
  330 
  331         self.assertFalse(os.path.isfile(rtag))
  332 
  333     @requires_salt_modules("pkg.info_installed")
  334     @requires_system_grains
  335     @slowTest
  336     def test_pkg_info(self, grains):
  337         """
  338         Test returning useful information on Ubuntu systems.
  339         """
  340         func = "pkg.info_installed"
  341 
  342         if grains["os_family"] == "Debian":
  343             ret = self.run_function(func, ["bash", "dpkg"])
  344             keys = ret.keys()
  345             self.assertIn("bash", keys)
  346             self.assertIn("dpkg", keys)
  347         elif grains["os_family"] == "RedHat":
  348             ret = self.run_function(func, ["rpm", "bash"])
  349             keys = ret.keys()
  350             self.assertIn("rpm", keys)
  351             self.assertIn("bash", keys)
  352         elif grains["os_family"] == "Suse":
  353             ret = self.run_function(func, ["less", "zypper"])
  354             keys = ret.keys()
  355             self.assertIn("less", keys)
  356             self.assertIn("zypper", keys)
  357         else:
  358             ret = self.run_function(func, [self.pkg])
  359             keys = ret.keys()
  360             self.assertIn(self.pkg, keys)
  361 
  362     @skipIf(True, "Temporary Skip - Causes centos 8 test to fail")
  363     @destructiveTest
  364     @requires_network()
  365     @requires_salt_modules(
  366         "pkg.refresh_db",
  367         "pkg.upgrade",
  368         "pkg.install",
  369         "pkg.list_repo_pkgs",
  370         "pkg.list_upgrades",
  371     )
  372     @requires_system_grains
  373     @slowTest
  374     def test_pkg_upgrade_has_pending_upgrades(self, grains):
  375         """
  376         Test running a system upgrade when there are packages that need upgrading
  377         """
  378         if grains["os"] == "Arch":
  379             self.skipTest("Arch moved to Python 3.8 and we're not ready for it yet")
  380 
  381         func = "pkg.upgrade"
  382 
  383         # First make sure that an up-to-date copy of the package db is available
  384         self.run_function("pkg.refresh_db")
  385 
  386         if grains["os_family"] == "Suse":
  387             # This test assumes that there are multiple possible versions of a
  388             # package available. That makes it brittle if you pick just one
  389             # target, as changes in the available packages will break the test.
  390             # Therefore, we'll choose from several packages to make sure we get
  391             # one that is suitable for this test.
  392             packages = ("hwinfo", "avrdude", "diffoscope", "vim")
  393             available = self.run_function("pkg.list_repo_pkgs", packages)
  394 
  395             for package in packages:
  396                 try:
  397                     new, old = available[package][:2]
  398                 except (KeyError, ValueError):
  399                     # Package not available, or less than 2 versions
  400                     # available. This is not a suitable target.
  401                     continue
  402                 else:
  403                     target = package
  404                     break
  405             else:
  406                 # None of the packages have more than one version available, so
  407                 # we need to find new package(s). pkg.list_repo_pkgs can be
  408                 # used to get an overview of the available packages. We should
  409                 # try to find packages with few dependencies and small download
  410                 # sizes, to keep this test from taking longer than necessary.
  411                 self.fail("No suitable package found for this test")
  412 
  413             # Make sure we have the 2nd-oldest available version installed
  414             ret = self.run_function("pkg.install", [target], version=old)
  415             if not isinstance(ret, dict):
  416                 if ret.startswith("ERROR"):
  417                     self.skipTest(
  418                         "Could not install older {} to complete " "test.".format(target)
  419                     )
  420 
  421             # Run a system upgrade, which should catch the fact that the
  422             # targeted package needs upgrading, and upgrade it.
  423             ret = self.run_function(func)
  424 
  425             # The changes dictionary should not be empty.
  426             if "changes" in ret:
  427                 self.assertIn(target, ret["changes"])
  428             else:
  429                 self.assertIn(target, ret)
  430         else:
  431             ret = self.run_function("pkg.list_upgrades")
  432             if ret == "" or ret == {}:
  433                 self.skipTest(
  434                     "No updates available for this machine.  Skipping pkg.upgrade test."
  435                 )
  436             else:
  437                 args = []
  438                 if grains["os_family"] == "Debian":
  439                     args = ["dist_upgrade=True"]
  440                 ret = self.run_function(func, args)
  441                 self.assertNotEqual(ret, {})
  442 
  443     @destructiveTest
  444     @skipIf(
  445         salt.utils.platform.is_darwin(),
  446         "The jenkins user is equivalent to root on mac, causing the test to be unrunnable",
  447     )
  448     @requires_salt_modules("pkg.remove", "pkg.latest_version")
  449     @requires_salt_states("pkg.removed")
  450     @requires_system_grains
  451     @slowTest
  452     def test_pkg_latest_version(self, grains):
  453         """
  454         Check that pkg.latest_version returns the latest version of the uninstalled package.
  455         The package is not installed. Only the package version is checked.
  456         """
  457         self.run_state("pkg.removed", name=self.pkg)
  458 
  459         cmd_pkg = []
  460         if grains["os_family"] == "RedHat":
  461             cmd_pkg = self.run_function("cmd.run", ["yum list {}".format(self.pkg)])
  462         elif salt.utils.platform.is_windows():
  463             cmd_pkg = self.run_function("pkg.list_available", [self.pkg])
  464         elif grains["os_family"] == "Debian":
  465             cmd_pkg = self.run_function("cmd.run", ["apt list {}".format(self.pkg)])
  466         elif grains["os_family"] == "Arch":
  467             cmd_pkg = self.run_function("cmd.run", ["pacman -Si {}".format(self.pkg)])
  468         elif grains["os_family"] == "FreeBSD":
  469             cmd_pkg = self.run_function(
  470                 "cmd.run", ["pkg search -S name -qQ version -e {}".format(self.pkg)]
  471             )
  472         elif grains["os_family"] == "Suse":
  473             cmd_pkg = self.run_function("cmd.run", ["zypper info {}".format(self.pkg)])
  474         elif grains["os_family"] == "MacOS":
  475             brew_bin = salt.utils.path.which("brew")
  476             mac_user = self.run_function("file.get_user", [brew_bin])
  477             if mac_user == "root":
  478                 self.skipTest(
  479                     "brew cannot run as root, try a user in {}".format(
  480                         os.listdir("/Users/")
  481                     )
  482                 )
  483             cmd_pkg = self.run_function(
  484                 "cmd.run", ["brew info {}".format(self.pkg)], run_as=mac_user
  485             )
  486         else:
  487             self.skipTest(
  488                 "TODO: test not configured for {}".format(grains["os_family"])
  489             )
  490         pkg_latest = self.run_function("pkg.latest_version", [self.pkg])
  491         self.assertIn(pkg_latest, cmd_pkg)