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
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
|
# Copyright (C) 2015 Intel Corporation
#
# Released under the MIT license (see COPYING.MIT)
# Provides functions to help with exporting binaries obtained from built targets
import os, re, glob as g, shutil as sh,sys
from time import sleep
from .commands import runCmd
from difflib import SequenceMatcher as SM
try:
import bb
except ImportError:
class my_log():
def __init__(self):
pass
def plain(self, msg):
if msg:
print(msg)
def warn(self, msg):
if msg:
print("WARNING: " + msg)
def fatal(self, msg):
if msg:
print("FATAL:" + msg)
sys.exit(1)
bb = my_log()
def determine_if_poky_env():
"""
used to determine if we are inside the poky env or not. Usefull for remote machine where poky is not present
"""
check_env = True if ("/scripts" and "/bitbake/bin") in os.getenv("PATH") else False
return check_env
def get_dest_folder(tune_features, folder_list):
"""
Function to determine what rpm deploy dir to choose for a given architecture based on TUNE_FEATURES
"""
features_list = tune_features.split(" ")
features_list.reverse()
features_list = "_".join(features_list)
match_rate = 0
best_match = None
for folder in folder_list:
curr_match_rate = SM(None, folder, features_list).ratio()
if curr_match_rate > match_rate:
match_rate = curr_match_rate
best_match = folder
return best_match
def process_binaries(d, params):
param_list = params
export_env = d.getVar("TEST_EXPORT_ONLY")
def extract_binary(pth_to_pkg, dest_pth=None):
cpio_command = runCmd("which cpio")
rpm2cpio_command = runCmd("ls /usr/bin/rpm2cpio")
if (cpio_command.status != 0) and (rpm2cpio_command.status != 0):
bb.fatal("Either \"rpm2cpio\" or \"cpio\" tools are not available on your system."
"All binaries extraction processes will not be available, crashing all related tests."
"Please install them according to your OS recommendations") # will exit here
if dest_pth:
os.chdir(dest_pth)
else:
os.chdir("%s" % os.sep)# this is for native package
extract_bin_command = runCmd("%s %s | %s -idm" % (rpm2cpio_command.output, pth_to_pkg, cpio_command.output)) # semi-hardcoded because of a bug on poky's rpm2cpio
return extract_bin_command
if determine_if_poky_env(): # machine with poky environment
exportpath = d.getVar("TEST_EXPORT_DIR", True) if export_env else d.getVar("DEPLOY_DIR", True)
rpm_deploy_dir = d.getVar("DEPLOY_DIR_RPM", True)
arch = get_dest_folder(d.getVar("TUNE_FEATURES", True), os.listdir(rpm_deploy_dir))
arch_rpm_dir = os.path.join(rpm_deploy_dir, arch)
extracted_bin_dir = os.path.join(exportpath,"binaries", arch, "extracted_binaries")
packaged_bin_dir = os.path.join(exportpath,"binaries", arch, "packaged_binaries")
# creating necessary directory structure in case testing is done in poky env.
if export_env == "0":
if not os.path.exists(extracted_bin_dir): bb.utils.mkdirhier(extracted_bin_dir)
if not os.path.exists(packaged_bin_dir): bb.utils.mkdirhier(packaged_bin_dir)
if param_list[3] == "native":
if export_env == "1": #this is a native package and we only need to copy it. no need for extraction
native_rpm_dir = os.path.join(rpm_deploy_dir, get_dest_folder("{} nativesdk".format(d.getVar("BUILD_SYS")), os.listdir(rpm_deploy_dir)))
native_rpm_file_list = [item for item in os.listdir(native_rpm_dir) if re.search("nativesdk-" + param_list[0] + "-([0-9]+\.*)", item)]
if not native_rpm_file_list:
bb.warn("Couldn't find any version of {} native package. Related tests will most probably fail.".format(param_list[0]))
return ""
for item in native_rpm_file_list:# will copy all versions of package. Used version will be selected on remote machine
bb.plain("Copying native package file: %s" % item)
sh.copy(os.path.join(rpm_deploy_dir, native_rpm_dir, item), os.path.join(d.getVar("TEST_EXPORT_DIR", True), "binaries", "native"))
else: # nothing to do here; running tests under bitbake, so we asume native binaries are in sysroots dir.
if param_list[1] or param_list[4]:
bb.warn("Native binary %s %s%s. Running tests under bitbake environment. Version can't be checked except when the test itself does it"
" and binary can't be removed."%(param_list[0],"has assigned ver. " + param_list[1] if param_list[1] else "",
", is marked for removal" if param_list[4] else ""))
else:# the package is target aka DUT intended and it is either required to be delivered in an extracted form or in a packaged version
target_rpm_file_list = [item for item in os.listdir(arch_rpm_dir) if re.search(param_list[0] + "-([0-9]+\.*)", item)]
if not target_rpm_file_list:
bb.warn("Couldn't find any version of target package %s. Please ensure it was built. "
"Related tests will probably fail." % param_list[0])
return ""
if param_list[2] == "rpm": # binary should be deployed as rpm; (other, .deb, .ipk? ; in the near future)
for item in target_rpm_file_list: # copying all related rpm packages. "Intuition" reasons, someone may need other versions too. Deciding later on version
bb.plain("Copying target specific packaged file: %s" % item)
sh.copy(os.path.join(arch_rpm_dir, item), packaged_bin_dir)
return "copied"
else: # it is required to extract the binary
if param_list[1]: # the package is versioned
for item in target_rpm_file_list:
if re.match(".*-{}-.*\.rpm".format(param_list[1]), item):
destination = os.path.join(extracted_bin_dir,param_list[0], param_list[1])
bb.utils.mkdirhier(destination)
extract_binary(os.path.join(arch_rpm_dir, item), destination)
break
else:
bb.warn("Couldn't find the desired version %s for target binary %s. Related test cases will probably fail." % (param_list[1], param_list[0]))
return ""
return "extracted"
else: # no version provided, just extract one binary
destination = os.path.join(extracted_bin_dir,param_list[0],
re.search(".*-([0-9]+\.[0-9]+)-.*rpm", target_rpm_file_list[0]).group(1))
bb.utils.mkdirhier(destination)
extract_binary(os.path.join(arch_rpm_dir, target_rpm_file_list[0]), destination)
return "extracted"
else: # remote machine
binaries_path = os.getenv("bin_dir")# in order to know where the binaries are, bin_dir is set as env. variable
if param_list[3] == "native": #need to extract the native pkg here
native_rpm_dir = os.path.join(binaries_path, "native")
native_rpm_file_list = os.listdir(native_rpm_dir)
for item in native_rpm_file_list:
if param_list[1] and re.match("nativesdk-{}-{}-.*\.rpm".format(param_list[0], param_list[1]), item): # native package has version
extract_binary(os.path.join(native_rpm_dir, item))
break
else:# just copy any related native binary
found_version = re.match("nativesdk-{}-([0-9]+\.[0-9]+)-".format(param_list[0]), item).group(1)
if found_version:
extract_binary(os.path.join(native_rpm_dir, item))
else:
bb.warn("Couldn't find native package %s%s. Related test cases will be influenced." %
(param_list[0], " with version " + param_list[1] if param_list[1] else ""))
return
else: # this is for target device
if param_list[2] == "rpm":
return "No need to extract, this is an .rpm file"
arch = get_dest_folder(d.getVar("TUNE_FEATURES", True), os.listdir(binaries_path))
extracted_bin_path = os.path.join(binaries_path, arch, "extracted_binaries")
extracted_bin_list = [item for item in os.listdir(extracted_bin_path)]
packaged_bin_path = os.path.join(binaries_path, arch, "packaged_binaries")
packaged_bin_file_list = os.listdir(packaged_bin_path)
# see if the package is already in the extracted ones; maybe it was deployed when exported the env.
if os.path.exists(os.path.join(extracted_bin_path, param_list[0], param_list[1] if param_list[1] else "")):
return "binary %s is already extracted" % param_list[0]
else: # we need to search for it in the packaged binaries directory. It may have been shipped after export
for item in packaged_bin_file_list:
if param_list[1]:
if re.match("%s-%s.*rpm" % (param_list[0], param_list[1]), item): # package with version
if not os.path.exists(os.path.join(extracted_bin_path, param_list[0],param_list[1])):
os.makedirs(os.path.join(extracted_bin_path, param_list[0], param_list[1]))
extract_binary(os.path.join(packaged_bin_path, item), os.path.join(extracted_bin_path, param_list[0],param_list[1]))
bb.plain("Using {} for {}".format(os.path.join(packaged_bin_path, item), param_list[0]))
break
else:
if re.match("%s-.*rpm" % param_list[0], item):
found_version = re.match(".*-([0-9]+\.[0-9]+)-", item).group(1)
if not os.path.exists(os.path.join(extracted_bin_path, param_list[0], found_version)):
os.makedirs(os.path.join(extracted_bin_path, param_list[0], found_version))
bb.plain("Used ver. %s for %s" % (found_version, param_list[0]))
extract_binary(os.path.join(packaged_bin_path, item), os.path.join(extracted_bin_path, param_list[0], found_version))
break
else:
bb.warn("Couldn't find target package %s%s. Please ensure it is available "
"in either of these directories: extracted_binaries or packaged_binaries. "
"Related tests will probably fail." % (param_list[0], " with version " + param_list[1] if param_list[1] else ""))
return
return "Binary %s extracted successfully." % param_list[0]
def files_to_copy(base_dir):
"""
Produces a list of files relative to the base dir path sent as param
:return: the list of relative path files
"""
files_list = []
dir_list = [base_dir]
count = 1
dir_count = 1
while (dir_count == 1 or dir_count != count):
count = dir_count
for dir in dir_list:
for item in os.listdir(dir):
if os.path.isdir(os.path.join(dir, item)) and os.path.join(dir, item) not in dir_list:
dir_list.append(os.path.join(dir, item))
dir_count = len(dir_list)
elif os.path.join(dir, item) not in files_list and os.path.isfile(os.path.join(dir, item)):
files_list.append(os.path.join(dir, item))
return files_list
def send_bin_to_DUT(d,params):
from oeqa.oetest import oeRuntimeTest
param_list = params
cleanup_list = list()
bins_dir = os.path.join(d.getVar("TEST_EXPORT_DIR", True), "binaries") if determine_if_poky_env() \
else os.getenv("bin_dir")
arch = get_dest_folder(d.getVar("TUNE_FEATURES", True), os.listdir(bins_dir))
arch_rpms_dir = os.path.join(bins_dir, arch, "packaged_binaries")
extracted_bin_dir = os.path.join(bins_dir, arch, "extracted_binaries", param_list[0])
def send_extracted_binary():
bin_local_dir = os.path.join(extracted_bin_dir, param_list[1] if param_list[1] else os.listdir(extracted_bin_dir)[0])
for item in files_to_copy(bin_local_dir):
split_path = item.split(bin_local_dir)[1]
path_on_DUT = split_path if split_path[0] is "/" else "/" + split_path # create the path as on DUT; eg. /usr/bin/bin_file
(status, output) = oeRuntimeTest.tc.target.copy_to(item, path_on_DUT)
if status != 0:
bb.warn("Failed to copy %s binary file %s on the remote target: %s" %
(param_list[0], "ver. " + param_list[1] if param_list[1] else "", d.getVar("MACHINE")))
return
if param_list[4] == "rm":
cleanup_list.append(path_on_DUT)
return cleanup_list
def send_rpm(remote_path): # if it is not required to have an extracted binary, but to send an .rpm file
rpm_to_send = ""
for item in os.listdir(arch_rpms_dir):
if param_list[1] and re.match("%s-%s-.*rpm"%(param_list[0], param_list[1]), item):
rpm_to_send = item
break
elif re.match("%s-[0-9]+\.[0-9]+-.*rpm" % param_list[0], item):
rpm_to_send = item
break
else:
bb.warn("No rpm package found for %s %s in .rpm files dir %s. Skipping deployment." %
(param_list[0], "ver. " + param_list[1] if param_list[1] else "", rpms_file_dir) )
return
(status, output) = oeRuntimeTest.tc.target.copy_to(os.path.join(arch_rpms_dir, rpm_to_send), remote_path)
if status != 0:
bb.warn("Failed to copy %s on the remote target: %s" %(param_list[0], d.getVar("MACHINE")))
return
if param_list[4] == "rm":
cleanup_list.append(os.path.join(remote_path, rpm_to_send))
return cleanup_list
if param_list[2] == "rpm": # send an .rpm file
return send_rpm("/home/root") # rpms will be sent on home dir of remote machine
else:
return send_extracted_binary()
def rm_bin(removal_list): # need to know both if the binary is sent archived and the path where it is sent if archived
from oeqa.oetest import oeRuntimeTest
for item in removal_list:
(status,output) = oeRuntimeTest.tc.target.run("rm " + item)
if status != 0:
bb.warn("Failed to remove: %s. Please ensure connection with the target device is up and running and "
"you have the needed rights." % item)
|
