"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)