blob: be2a2110fc62820729d606986be26d5c49bb5e25 [file] [log] [blame]
Patrick Williamsd8c66bc2016-06-20 12:57:21 -05001# Copyright (C) 2015 Intel Corporation
2#
3# Released under the MIT license (see COPYING.MIT)
4
5# Provides functions to help with exporting binaries obtained from built targets
6
7import os, re, glob as g, shutil as sh,sys
8from time import sleep
Patrick Williamsc0f7c042017-02-23 20:41:17 -06009from .commands import runCmd
Patrick Williamsd8c66bc2016-06-20 12:57:21 -050010from difflib import SequenceMatcher as SM
11
12try:
13 import bb
14except ImportError:
15 class my_log():
16 def __init__(self):
17 pass
18 def plain(self, msg):
19 if msg:
Patrick Williamsc0f7c042017-02-23 20:41:17 -060020 print(msg)
Patrick Williamsd8c66bc2016-06-20 12:57:21 -050021 def warn(self, msg):
22 if msg:
Patrick Williamsc0f7c042017-02-23 20:41:17 -060023 print("WARNING: " + msg)
Patrick Williamsd8c66bc2016-06-20 12:57:21 -050024 def fatal(self, msg):
25 if msg:
Patrick Williamsc0f7c042017-02-23 20:41:17 -060026 print("FATAL:" + msg)
Patrick Williamsd8c66bc2016-06-20 12:57:21 -050027 sys.exit(1)
28 bb = my_log()
29
30
31def determine_if_poky_env():
32 """
33 used to determine if we are inside the poky env or not. Usefull for remote machine where poky is not present
34 """
35 check_env = True if ("/scripts" and "/bitbake/bin") in os.getenv("PATH") else False
36 return check_env
37
38
39def get_dest_folder(tune_features, folder_list):
40 """
41 Function to determine what rpm deploy dir to choose for a given architecture based on TUNE_FEATURES
42 """
43 features_list = tune_features.split(" ")
44 features_list.reverse()
45 features_list = "_".join(features_list)
46 match_rate = 0
47 best_match = None
48 for folder in folder_list:
49 curr_match_rate = SM(None, folder, features_list).ratio()
50 if curr_match_rate > match_rate:
51 match_rate = curr_match_rate
52 best_match = folder
53 return best_match
54
55
56def process_binaries(d, params):
57 param_list = params
58 export_env = d.getVar("TEST_EXPORT_ONLY")
59
60 def extract_binary(pth_to_pkg, dest_pth=None):
61 cpio_command = runCmd("which cpio")
62 rpm2cpio_command = runCmd("ls /usr/bin/rpm2cpio")
63 if (cpio_command.status != 0) and (rpm2cpio_command.status != 0):
64 bb.fatal("Either \"rpm2cpio\" or \"cpio\" tools are not available on your system."
65 "All binaries extraction processes will not be available, crashing all related tests."
66 "Please install them according to your OS recommendations") # will exit here
67 if dest_pth:
68 os.chdir(dest_pth)
69 else:
70 os.chdir("%s" % os.sep)# this is for native package
71 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
72 return extract_bin_command
73
74 if determine_if_poky_env(): # machine with poky environment
Brad Bishop6e60e8b2018-02-01 10:27:11 -050075 exportpath = d.getVar("TEST_EXPORT_DIR") if export_env else d.getVar("DEPLOY_DIR")
76 rpm_deploy_dir = d.getVar("DEPLOY_DIR_RPM")
77 arch = get_dest_folder(d.getVar("TUNE_FEATURES"), os.listdir(rpm_deploy_dir))
Patrick Williamsd8c66bc2016-06-20 12:57:21 -050078 arch_rpm_dir = os.path.join(rpm_deploy_dir, arch)
79 extracted_bin_dir = os.path.join(exportpath,"binaries", arch, "extracted_binaries")
80 packaged_bin_dir = os.path.join(exportpath,"binaries", arch, "packaged_binaries")
81 # creating necessary directory structure in case testing is done in poky env.
82 if export_env == "0":
83 if not os.path.exists(extracted_bin_dir): bb.utils.mkdirhier(extracted_bin_dir)
84 if not os.path.exists(packaged_bin_dir): bb.utils.mkdirhier(packaged_bin_dir)
85
86 if param_list[3] == "native":
87 if export_env == "1": #this is a native package and we only need to copy it. no need for extraction
88 native_rpm_dir = os.path.join(rpm_deploy_dir, get_dest_folder("{} nativesdk".format(d.getVar("BUILD_SYS")), os.listdir(rpm_deploy_dir)))
89 native_rpm_file_list = [item for item in os.listdir(native_rpm_dir) if re.search("nativesdk-" + param_list[0] + "-([0-9]+\.*)", item)]
90 if not native_rpm_file_list:
91 bb.warn("Couldn't find any version of {} native package. Related tests will most probably fail.".format(param_list[0]))
92 return ""
93 for item in native_rpm_file_list:# will copy all versions of package. Used version will be selected on remote machine
94 bb.plain("Copying native package file: %s" % item)
Brad Bishop6e60e8b2018-02-01 10:27:11 -050095 sh.copy(os.path.join(rpm_deploy_dir, native_rpm_dir, item), os.path.join(d.getVar("TEST_EXPORT_DIR"), "binaries", "native"))
Patrick Williamsd8c66bc2016-06-20 12:57:21 -050096 else: # nothing to do here; running tests under bitbake, so we asume native binaries are in sysroots dir.
97 if param_list[1] or param_list[4]:
98 bb.warn("Native binary %s %s%s. Running tests under bitbake environment. Version can't be checked except when the test itself does it"
99 " and binary can't be removed."%(param_list[0],"has assigned ver. " + param_list[1] if param_list[1] else "",
100 ", is marked for removal" if param_list[4] else ""))
101 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
102 target_rpm_file_list = [item for item in os.listdir(arch_rpm_dir) if re.search(param_list[0] + "-([0-9]+\.*)", item)]
103 if not target_rpm_file_list:
104 bb.warn("Couldn't find any version of target package %s. Please ensure it was built. "
105 "Related tests will probably fail." % param_list[0])
106 return ""
107 if param_list[2] == "rpm": # binary should be deployed as rpm; (other, .deb, .ipk? ; in the near future)
108 for item in target_rpm_file_list: # copying all related rpm packages. "Intuition" reasons, someone may need other versions too. Deciding later on version
109 bb.plain("Copying target specific packaged file: %s" % item)
110 sh.copy(os.path.join(arch_rpm_dir, item), packaged_bin_dir)
111 return "copied"
112 else: # it is required to extract the binary
113 if param_list[1]: # the package is versioned
114 for item in target_rpm_file_list:
115 if re.match(".*-{}-.*\.rpm".format(param_list[1]), item):
116 destination = os.path.join(extracted_bin_dir,param_list[0], param_list[1])
117 bb.utils.mkdirhier(destination)
118 extract_binary(os.path.join(arch_rpm_dir, item), destination)
119 break
120 else:
121 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]))
122 return ""
123 return "extracted"
124 else: # no version provided, just extract one binary
125 destination = os.path.join(extracted_bin_dir,param_list[0],
126 re.search(".*-([0-9]+\.[0-9]+)-.*rpm", target_rpm_file_list[0]).group(1))
127 bb.utils.mkdirhier(destination)
128 extract_binary(os.path.join(arch_rpm_dir, target_rpm_file_list[0]), destination)
129 return "extracted"
130 else: # remote machine
131 binaries_path = os.getenv("bin_dir")# in order to know where the binaries are, bin_dir is set as env. variable
132 if param_list[3] == "native": #need to extract the native pkg here
133 native_rpm_dir = os.path.join(binaries_path, "native")
134 native_rpm_file_list = os.listdir(native_rpm_dir)
135 for item in native_rpm_file_list:
136 if param_list[1] and re.match("nativesdk-{}-{}-.*\.rpm".format(param_list[0], param_list[1]), item): # native package has version
137 extract_binary(os.path.join(native_rpm_dir, item))
138 break
139 else:# just copy any related native binary
140 found_version = re.match("nativesdk-{}-([0-9]+\.[0-9]+)-".format(param_list[0]), item).group(1)
141 if found_version:
142 extract_binary(os.path.join(native_rpm_dir, item))
143 else:
144 bb.warn("Couldn't find native package %s%s. Related test cases will be influenced." %
145 (param_list[0], " with version " + param_list[1] if param_list[1] else ""))
146 return
147
148 else: # this is for target device
149 if param_list[2] == "rpm":
150 return "No need to extract, this is an .rpm file"
Brad Bishop6e60e8b2018-02-01 10:27:11 -0500151 arch = get_dest_folder(d.getVar("TUNE_FEATURES"), os.listdir(binaries_path))
Patrick Williamsd8c66bc2016-06-20 12:57:21 -0500152 extracted_bin_path = os.path.join(binaries_path, arch, "extracted_binaries")
153 extracted_bin_list = [item for item in os.listdir(extracted_bin_path)]
154 packaged_bin_path = os.path.join(binaries_path, arch, "packaged_binaries")
155 packaged_bin_file_list = os.listdir(packaged_bin_path)
156 # see if the package is already in the extracted ones; maybe it was deployed when exported the env.
157 if os.path.exists(os.path.join(extracted_bin_path, param_list[0], param_list[1] if param_list[1] else "")):
158 return "binary %s is already extracted" % param_list[0]
159 else: # we need to search for it in the packaged binaries directory. It may have been shipped after export
160 for item in packaged_bin_file_list:
161 if param_list[1]:
162 if re.match("%s-%s.*rpm" % (param_list[0], param_list[1]), item): # package with version
163 if not os.path.exists(os.path.join(extracted_bin_path, param_list[0],param_list[1])):
164 os.makedirs(os.path.join(extracted_bin_path, param_list[0], param_list[1]))
165 extract_binary(os.path.join(packaged_bin_path, item), os.path.join(extracted_bin_path, param_list[0],param_list[1]))
166 bb.plain("Using {} for {}".format(os.path.join(packaged_bin_path, item), param_list[0]))
167 break
168 else:
169 if re.match("%s-.*rpm" % param_list[0], item):
170 found_version = re.match(".*-([0-9]+\.[0-9]+)-", item).group(1)
171 if not os.path.exists(os.path.join(extracted_bin_path, param_list[0], found_version)):
172 os.makedirs(os.path.join(extracted_bin_path, param_list[0], found_version))
173 bb.plain("Used ver. %s for %s" % (found_version, param_list[0]))
174 extract_binary(os.path.join(packaged_bin_path, item), os.path.join(extracted_bin_path, param_list[0], found_version))
175 break
176 else:
177 bb.warn("Couldn't find target package %s%s. Please ensure it is available "
178 "in either of these directories: extracted_binaries or packaged_binaries. "
179 "Related tests will probably fail." % (param_list[0], " with version " + param_list[1] if param_list[1] else ""))
180 return
181 return "Binary %s extracted successfully." % param_list[0]
182
183
184def files_to_copy(base_dir):
185 """
186 Produces a list of files relative to the base dir path sent as param
187 :return: the list of relative path files
188 """
189 files_list = []
190 dir_list = [base_dir]
191 count = 1
192 dir_count = 1
193 while (dir_count == 1 or dir_count != count):
194 count = dir_count
195 for dir in dir_list:
196 for item in os.listdir(dir):
197 if os.path.isdir(os.path.join(dir, item)) and os.path.join(dir, item) not in dir_list:
198 dir_list.append(os.path.join(dir, item))
199 dir_count = len(dir_list)
200 elif os.path.join(dir, item) not in files_list and os.path.isfile(os.path.join(dir, item)):
201 files_list.append(os.path.join(dir, item))
202 return files_list
203
204
205def send_bin_to_DUT(d,params):
206 from oeqa.oetest import oeRuntimeTest
207 param_list = params
208 cleanup_list = list()
Brad Bishop6e60e8b2018-02-01 10:27:11 -0500209 bins_dir = os.path.join(d.getVar("TEST_EXPORT_DIR"), "binaries") if determine_if_poky_env() \
Patrick Williamsd8c66bc2016-06-20 12:57:21 -0500210 else os.getenv("bin_dir")
Brad Bishop6e60e8b2018-02-01 10:27:11 -0500211 arch = get_dest_folder(d.getVar("TUNE_FEATURES"), os.listdir(bins_dir))
Patrick Williamsd8c66bc2016-06-20 12:57:21 -0500212 arch_rpms_dir = os.path.join(bins_dir, arch, "packaged_binaries")
213 extracted_bin_dir = os.path.join(bins_dir, arch, "extracted_binaries", param_list[0])
214
215 def send_extracted_binary():
216 bin_local_dir = os.path.join(extracted_bin_dir, param_list[1] if param_list[1] else os.listdir(extracted_bin_dir)[0])
217 for item in files_to_copy(bin_local_dir):
218 split_path = item.split(bin_local_dir)[1]
219 path_on_DUT = split_path if split_path[0] is "/" else "/" + split_path # create the path as on DUT; eg. /usr/bin/bin_file
220 (status, output) = oeRuntimeTest.tc.target.copy_to(item, path_on_DUT)
221 if status != 0:
222 bb.warn("Failed to copy %s binary file %s on the remote target: %s" %
223 (param_list[0], "ver. " + param_list[1] if param_list[1] else "", d.getVar("MACHINE")))
224 return
225 if param_list[4] == "rm":
226 cleanup_list.append(path_on_DUT)
227 return cleanup_list
228
229 def send_rpm(remote_path): # if it is not required to have an extracted binary, but to send an .rpm file
230 rpm_to_send = ""
231 for item in os.listdir(arch_rpms_dir):
232 if param_list[1] and re.match("%s-%s-.*rpm"%(param_list[0], param_list[1]), item):
233 rpm_to_send = item
234 break
235 elif re.match("%s-[0-9]+\.[0-9]+-.*rpm" % param_list[0], item):
236 rpm_to_send = item
237 break
238 else:
239 bb.warn("No rpm package found for %s %s in .rpm files dir %s. Skipping deployment." %
240 (param_list[0], "ver. " + param_list[1] if param_list[1] else "", rpms_file_dir) )
241 return
242 (status, output) = oeRuntimeTest.tc.target.copy_to(os.path.join(arch_rpms_dir, rpm_to_send), remote_path)
243 if status != 0:
244 bb.warn("Failed to copy %s on the remote target: %s" %(param_list[0], d.getVar("MACHINE")))
245 return
246 if param_list[4] == "rm":
247 cleanup_list.append(os.path.join(remote_path, rpm_to_send))
248 return cleanup_list
249
250 if param_list[2] == "rpm": # send an .rpm file
251 return send_rpm("/home/root") # rpms will be sent on home dir of remote machine
252 else:
253 return send_extracted_binary()
254
255
256def rm_bin(removal_list): # need to know both if the binary is sent archived and the path where it is sent if archived
257 from oeqa.oetest import oeRuntimeTest
258 for item in removal_list:
259 (status,output) = oeRuntimeTest.tc.target.run("rm " + item)
260 if status != 0:
261 bb.warn("Failed to remove: %s. Please ensure connection with the target device is up and running and "
262 "you have the needed rights." % item)
263