summaryrefslogtreecommitdiffstats
path: root/meta/lib/oeqa/utils/package_manager.py
blob: 2d358f7172eee7083501e5d8d0d0b71a010351b7 (plain)
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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
#
# 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, '/')