xref: /openbmc/openbmc/poky/meta/lib/oe/package_manager/common_deb_ipk.py (revision edff49234e31f23dc79f823473c9e286a21596c1)
1#
2# Copyright OpenEmbedded Contributors
3#
4# SPDX-License-Identifier: GPL-2.0-only
5#
6
7import glob
8import os
9import subprocess
10import tempfile
11
12import bb
13
14from oe.package_manager import opkg_query, PackageManager
15
16class OpkgDpkgPM(PackageManager):
17    def __init__(self, d, target_rootfs):
18        """
19        This is an abstract class. Do not instantiate this directly.
20        """
21        super(OpkgDpkgPM, self).__init__(d, target_rootfs)
22
23    def package_info(self, pkg):
24        """
25        Returns a dictionary with the package info.
26        """
27        raise NotImplementedError
28
29    def _common_package_info(self, cmd):
30        """
31       "Returns a dictionary with the package info.
32
33        This method extracts the common parts for Opkg and Dpkg
34        """
35
36        proc = subprocess.run(cmd, capture_output=True, encoding="utf-8", shell=True)
37        if proc.returncode:
38            bb.fatal("Unable to list available packages. Command '%s' "
39                     "returned %d:\n%s" % (cmd, proc.returncode, proc.stderr))
40        elif proc.stderr:
41            bb.note("Command '%s' returned stderr: %s" % (cmd, proc.stderr))
42
43        return opkg_query(proc.stdout)
44
45    def extract(self, pkg):
46        """
47        Returns the path to a tmpdir where resides the contents of a package.
48
49        Deleting the tmpdir is responsability of the caller.
50        """
51        pkg_info = self.package_info(pkg)
52        if not pkg_info:
53            bb.fatal("Unable to get information for package '%s' while "
54                     "trying to extract the package."  % pkg)
55
56        ar_cmd = bb.utils.which(os.getenv("PATH"), "ar")
57        tar_cmd = bb.utils.which(os.getenv("PATH"), "tar")
58        pkg_path = pkg_info[pkg]["filepath"]
59
60        if not os.path.isfile(pkg_path):
61            bb.fatal("Unable to extract package for '%s'."
62                     "File %s doesn't exists" % (pkg, pkg_path))
63
64        tmp_dir = tempfile.mkdtemp()
65        current_dir = os.getcwd()
66        os.chdir(tmp_dir)
67
68        try:
69            cmd = [ar_cmd, 'x', pkg_path]
70            output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
71            data_tar = glob.glob("data.tar.*")
72            if len(data_tar) != 1:
73                bb.fatal("Unable to extract %s package. Failed to identify "
74                         "data tarball (found tarballs '%s').",
75                         pkg_path, data_tar)
76            data_tar = data_tar[0]
77            cmd = [tar_cmd, 'xf', data_tar]
78            output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
79        except subprocess.CalledProcessError as e:
80            bb.utils.remove(tmp_dir, recurse=True)
81            bb.fatal("Unable to extract %s package. Command '%s' "
82                     "returned %d:\n%s" % (pkg_path, ' '.join(cmd), e.returncode, e.output.decode("utf-8")))
83        except OSError as e:
84            bb.utils.remove(tmp_dir, recurse=True)
85            bb.fatal("Unable to extract %s package. Command '%s' "
86                     "returned %d:\n%s at %s" % (pkg_path, ' '.join(cmd), e.errno, e.strerror, e.filename))
87
88        bb.note("Extracted %s to %s" % (pkg_path, tmp_dir))
89        bb.utils.remove(os.path.join(tmp_dir, "debian-binary"))
90        bb.utils.remove(os.path.join(tmp_dir, "control.tar.gz"))
91        bb.utils.remove(os.path.join(tmp_dir, data_tar))
92        os.chdir(current_dir)
93
94        return tmp_dir
95
96    def _handle_intercept_failure(self, registered_pkgs):
97        self.mark_packages("unpacked", registered_pkgs.split())
98