xref: /openbmc/openbmc/poky/meta/lib/oeqa/utils/package_manager.py (revision c342db35)

#
# SPDX-License-Identifier: MIT
#

import os
import json
import shutil

from oeqa.core.utils.test import getCaseFile, getCaseMethod

def get_package_manager(d, root_path):
    """
    Returns an OE package manager that can install packages in root_path.
    """
    from oe.package_manager import RpmPM, OpkgPM, DpkgPM

    pkg_class = d.getVar("IMAGE_PKGTYPE")
    if pkg_class == "rpm":
        pm = RpmPM(d,
                   root_path,
                   d.getVar('TARGET_VENDOR'),
                   filterbydependencies=False)
        pm.create_configs()

    elif pkg_class == "ipk":
        pm = OpkgPM(d,
                    root_path,
                    d.getVar("IPKGCONF_TARGET"),
                    d.getVar("ALL_MULTILIB_PACKAGE_ARCHS"),
                    filterbydependencies=False)

    elif pkg_class == "deb":
        pm = DpkgPM(d,
                    root_path,
                    d.getVar('PACKAGE_ARCHS'),
                    d.getVar('DPKG_ARCH'),
                    filterbydependencies=False)

    pm.write_index()
    pm.update()

    return pm

def find_packages_to_extract(test_suite):
    """
    Returns packages to extract required by runtime tests.
    """
    from oeqa.core.utils.test import getSuiteCasesFiles

    needed_packages = {}
    files = getSuiteCasesFiles(test_suite)

    for f in set(files):
        json_file = _get_json_file(f)
        if json_file:
            needed_packages.update(_get_needed_packages(json_file))

    return needed_packages

def _get_json_file(module_path):
    """
    Returns the path of the JSON file for a module, empty if doesn't exitst.
    """

    json_file = '%s.json' % module_path.rsplit('.', 1)[0]
    if os.path.isfile(module_path) and os.path.isfile(json_file):
        return json_file
    else:
        return ''

def _get_needed_packages(json_file, test=None):
    """
    Returns a dict with needed packages based on a JSON file.

    If a test is specified it will return the dict just for that test.
    """
    needed_packages = {}

    with open(json_file) as f:
        test_packages = json.load(f)
    for key,value in test_packages.items():
        needed_packages[key] = value

    if test:
        if test in needed_packages:
            needed_packages = needed_packages[test]
        else:
            needed_packages = {}

    return needed_packages

def extract_packages(d, needed_packages):
    """
    Extract packages that will be needed during runtime.
    """

    import bb
    import oe.path

    extracted_path = d.getVar('TEST_EXTRACTED_DIR')

    for key,value in needed_packages.items():
        packages = ()
        if isinstance(value, dict):
            packages = (value, )
        elif isinstance(value, list):
            packages = value
        else:
            bb.fatal('Failed to process needed packages for %s; '
                     'Value must be a dict or list' % key)

        for package in packages:
            pkg = package['pkg']
            rm = package.get('rm', False)
            extract = package.get('extract', True)

            if extract:
                #logger.debug(1, 'Extracting %s' % pkg)
                dst_dir = os.path.join(extracted_path, pkg)
                # Same package used for more than one test,
                # don't need to extract again.
                if os.path.exists(dst_dir):
                    continue

                # Extract package and copy it to TEST_EXTRACTED_DIR
                pkg_dir = _extract_in_tmpdir(d, pkg)
                oe.path.copytree(pkg_dir, dst_dir)
                shutil.rmtree(pkg_dir)

            else:
                #logger.debug(1, 'Copying %s' % pkg)
                _copy_package(d, pkg)

def _extract_in_tmpdir(d, pkg):
    """"
    Returns path to a temp directory where the package was
    extracted without dependencies.
    """

    from oeqa.utils.package_manager import get_package_manager

    pkg_path = os.path.join(d.getVar('TEST_INSTALL_TMP_DIR'), pkg)
    pm = get_package_manager(d, pkg_path)
    extract_dir = pm.extract(pkg)
    shutil.rmtree(pkg_path)

    return extract_dir

def _copy_package(d, pkg):
    """
    Copy the RPM, DEB or IPK package to dst_dir
    """

    from oeqa.utils.package_manager import get_package_manager

    pkg_path = os.path.join(d.getVar('TEST_INSTALL_TMP_DIR'), pkg)
    dst_dir = d.getVar('TEST_PACKAGED_DIR')
    pm = get_package_manager(d, pkg_path)
    pkg_info = pm.package_info(pkg)
    file_path = pkg_info[pkg]['filepath']
    shutil.copy2(file_path, dst_dir)
    shutil.rmtree(pkg_path)

def install_package(test_case):
    """
    Installs package in DUT if required.
    """
    needed_packages = test_needs_package(test_case)
    if needed_packages:
        _install_uninstall_packages(needed_packages, test_case, True)

def uninstall_package(test_case):
    """
    Uninstalls package in DUT if required.
    """
    needed_packages = test_needs_package(test_case)
    if needed_packages:
        _install_uninstall_packages(needed_packages, test_case, False)

def test_needs_package(test_case):
    """
    Checks if a test case requires to install/uninstall packages.
    """
    test_file = getCaseFile(test_case)
    json_file = _get_json_file(test_file)

    if json_file:
        test_method = getCaseMethod(test_case)
        needed_packages = _get_needed_packages(json_file, test_method)
        if needed_packages:
            return needed_packages

    return None

def _install_uninstall_packages(needed_packages, test_case, install=True):
    """
    Install/Uninstall packages in the DUT without using a package manager
    """

    if isinstance(needed_packages, dict):
        packages = [needed_packages]
    elif isinstance(needed_packages, list):
        packages = needed_packages

    for package in packages:
        pkg = package['pkg']
        rm = package.get('rm', False)
        extract = package.get('extract', True)
        src_dir = os.path.join(test_case.tc.extract_dir, pkg)

        # Install package
        if install and extract:
            test_case.tc.target.copyDirTo(src_dir, '/')

        # Uninstall package
        elif not install and rm:
            test_case.tc.target.deleteDirStructure(src_dir, '/')