xref: /openbmc/qemu/tests/functional/qemu_test/archive.py (revision c055f1d26fbcc18de02e59569f3e2a224595835a)
1# SPDX-License-Identifier: GPL-2.0-or-later
2#
3# Utilities for python-based QEMU tests
4#
5# Copyright 2024 Red Hat, Inc.
6#
7# Authors:
8#  Thomas Huth <thuth@redhat.com>
9
10import os
11from subprocess import check_call, run, DEVNULL
12import tarfile
13import zipfile
14
15from .cmd import run_cmd
16
17
18def tar_extract(archive, dest_dir, member=None):
19    with tarfile.open(archive) as tf:
20        if hasattr(tarfile, 'data_filter'):
21            tf.extraction_filter = getattr(tarfile, 'data_filter',
22                                           (lambda member, path: member))
23        if member:
24            tf.extract(member=member, path=dest_dir)
25        else:
26            tf.extractall(path=dest_dir)
27
28def cpio_extract(archive, output_path):
29    cwd = os.getcwd()
30    os.chdir(output_path)
31    # Not passing 'check=True' as cpio exits with non-zero
32    # status if the archive contains any device nodes :-(
33    if type(archive) == str:
34        run(['cpio', '-i', '-F', archive],
35            stdout=DEVNULL, stderr=DEVNULL)
36    else:
37        run(['cpio', '-i'],
38            input=archive.read(),
39            stdout=DEVNULL, stderr=DEVNULL)
40    os.chdir(cwd)
41
42def zip_extract(archive, dest_dir, member=None):
43    with zipfile.ZipFile(archive, 'r') as zf:
44        if member:
45            zf.extract(member=member, path=dest_dir)
46        else:
47            zf.extractall(path=dest_dir)
48
49def deb_extract(archive, dest_dir, member=None):
50    cwd = os.getcwd()
51    os.chdir(dest_dir)
52    try:
53        (stdout, stderr, ret) = run_cmd(['ar', 't', archive])
54        file_path = stdout.split()[2]
55        run_cmd(['ar', 'x', archive, file_path])
56        tar_extract(file_path, dest_dir, member)
57    finally:
58        os.chdir(cwd)
59