1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
# Copyright (C) 2018 Linaro Limited
#
# Author: Remi Duraffort <remi.duraffort@linaro.org>
#
# This file is part of LAVA.
#
# LAVA is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# LAVA is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along
# with this program; if not, see <http://www.gnu.org/licenses>.
import re
import contextlib
import subprocess # nosec dpkg
from lava_common.exceptions import InfrastructureError
def binary_version(binary, flags="", pattern=""):
"""
Returns a string with the version of the binary by running it with
the provided flags. If the output from running the binary is verbose
and contains more than just the version number, a pattern can be
provided for the function to parse the output to pick out the
substring that contains the version number.
"""
# if binary is not absolute, fail.
msg = "Unable to retrieve version of %s" % binary
try:
ver_str = (
subprocess.check_output((binary, flags), stderr=subprocess.STDOUT)
.strip()
.decode("utf-8", errors="replace")
)
if not ver_str:
raise InfrastructureError(msg)
except (subprocess.CalledProcessError, FileNotFoundError):
raise InfrastructureError(msg)
if pattern != "":
p = re.compile(pattern)
result = p.search(ver_str)
if result is not None:
ver_str = result.group(1)
else:
raise InfrastructureError(msg)
return "%s, version %s" % (binary, ver_str)
def debian_package_arch(pkg):
"""
Relies on Debian Policy rules for the existence of the
changelog. Distributions not derived from Debian will
return an empty string.
"""
with contextlib.suppress(FileNotFoundError, subprocess.CalledProcessError):
return (
subprocess.check_output( # nosec dpkg-query
("dpkg-query", "-W", "-f=${Architecture}\n", "%s" % pkg),
stderr=subprocess.STDOUT,
)
.strip()
.decode("utf-8", errors="replace")
)
return ""
def debian_package_version(pkg):
"""
Use dpkg-query to retrieve the version of the given package.
Distributions not derived from Debian will return an empty string.
"""
with contextlib.suppress(FileNotFoundError, subprocess.CalledProcessError):
return (
subprocess.check_output( # nosec dpkg-query
("dpkg-query", "-W", "-f=${Version}\n", "%s" % pkg),
stderr=subprocess.STDOUT,
)
.strip()
.decode("utf-8", errors="replace")
)
return ""
def debian_filename_version(binary):
"""
Relies on Debian Policy rules for the existence of the
changelog. Distributions not derived from Debian will
return an empty string.
"""
# if binary is not absolute, fail.
msg = "Unable to retrieve version of %s" % binary
pkg_str = None
with contextlib.suppress(FileNotFoundError, subprocess.CalledProcessError):
pkg_str = (
subprocess.check_output( # nosec dpkg-query
("dpkg-query", "-S", binary), stderr=subprocess.STDOUT
)
.strip()
.decode("utf-8", errors="replace")
)
if not pkg_str:
raise InfrastructureError(msg)
pkg = pkg_str.split(":")[0]
pkg_ver = debian_package_version(pkg)
return "%s for <%s>, installed at version: %s" % (pkg, binary, pkg_ver)
|
