File: utils.py

package info (click to toggle)
lava 2024.09-1.1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 29,616 kB
  • sloc: python: 79,650; javascript: 16,875; sh: 1,364; makefile: 336
file content (102 lines) | stat: -rw-r--r-- 3,330 bytes parent folder | download
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
# Copyright (C) 2018 Linaro Limited
#
# Author: Remi Duraffort <remi.duraffort@linaro.org>
#
# SPDX-License-Identifier: GPL-2.0-or-later

import contextlib
import re
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)