| # 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) |
| |