How to use get_application_output method in avocado

Best Python code snippet using avocado_python Github


Full Screen

...73 bool : True if application is available,74 False if not available.75 """76 return which(name) is not None77def get_application_output(command, shell=False, timeout=None):78 """Run a command and get the output.79 Args:80 command (str, list): The command to run81 and it's arguments.82 shell (bool), optional: True if executing on83 shell, else False. Default is False.84 timeout (int, None), optional: Set a max85 execution time in seconds for the command.86 Returns:87 str: Command output if the command ran with88 a zero exit code. Returns a string containing89 the error reason in case the command failed.90 """91 try:92 return, shell=shell, check=True, universal_newlines=True,93 stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,94 timeout=timeout).stdout95 except subprocess.CalledProcessError:96 return "invalid"97 except subprocess.TimeoutExpired:98 return "timeout"99 except FileNotFoundError:100 return "unavailable"101def verify_interface(target_interface):102 """Verify if a wireless interface exists and is103 operational.104 Args:105 target_interface (str): The network interface to106 verify.107 Returns:108 None109 """110 cmd = "cat /sys/class/net/{0}/operstate".format(target_interface)111 check_interface = get_application_output(cmd, shell=True, timeout=10)112 if check_interface == "invalid":113 print("Interface {0} does not exist!".format(target_interface))114 exit(1)115 elif check_interface == "timeout":116 print("Unable to get interface {0} status".format(target_interface))117 exit(1)118 check_interface = check_interface.split("\n")[0]119 if check_interface != "up":120 print("Interface {0} is not ready.".format(target_interface))121 exit(1)122def process_iw(target_interface):123 """Get metrics from a wireless interface.124 Args:125 target_interface (str): The network interface to126 capture metrics from.127 Returns:128 dict: A dictionary containing the metrics and129 their values as corresponding (key, value) pairs.130 """131 verify_interface(target_interface)132 try:133 iw_info = get_application_output(134 ["iw {0} info".format(target_interface)],135 shell=True, timeout=10).replace("\t", " ")136 if iw_info == "invalid":137 print("The interface {0} is not a wireless interface".format(target_interface))138 exit(1)139 results = {}140 results["interface"] = re.findall(r"(?<=Interface )(.*)", iw_info)[0]141 results["interface_mac"] = re.findall(r"(?<=addr )(.*)", iw_info)[0]142 if not verify_mac(results["interface_mac"]):143 print("The interface {0} has an invalid MAC address".format(target_interface))144 exit(1)145 tmp = re.findall(r"(?<=channel )(.*?)(?=\,)", iw_info)[0].split(" ")146 results["channel"] = int(tmp[0])147 results["channel_frequency"] = int(tmp[1].replace("(", ""))148 results["ssid"] = re.findall(r"(?<=ssid )(.*)", iw_info)[0]149 iw_info = get_application_output(["iw {0} station dump".format(target_interface)],150 shell=True, timeout=10).replace("\t", " ")151 results["ssid_mac"] = re.findall(r"(?<=Station )(.*)(?= \()", iw_info)[0]152 if not verify_mac(results["ssid_mac"]):153 print("The station {0} has an invalid MAC address".format(results["ssid"]))154 exit(1)155 try:156 results["signal_strength"] = int(re.findall(r"(?<=signal avg: )(.*)", iw_info)[0].split(" ")[0])157 except IndexError:158 # Use fallback iwconfig command. This is not ideal. Need to rewrite this entire portion targeting nl80211159 iwconfig = get_application_output(["iwconfig {0}".format(target_interface)])160 results["signal_strength"] = int(re.findall(r"(?<=Signal level=)(.*)(?= dBm)", iwconfig)[0])161 except IndexError:162 raise ParseError("Unable to parse iw.") from None163 return results164def verify_mac(mac):165 """Verify if a MAC address is valid.166 Args:167 mac (str): The MAC address to verify.168 Returns:169 bool: True if the MAC address is valid,170 False if not.171 """172 if re.match(r"^([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})$", mac):173 return True174 else:175 return False176def verify_iperf(ip, port):177 """Create a socket and verify if iperf3 is running178 on the provided ip and port.179 Args:180 ip (str): The ip address of the iperf3 server.181 port (str): The port of the iperf3 server.182 Returns:183 bool: True if connection could be established.184 False if it failed to establish.185 """186 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)187 sock.settimeout(10)188 result = sock.connect_ex((ip, port))189 if result == 0:190 sock.close()191 return True192 else:193 sock.close()194 return False195@log_arguments196def run_iperf(ip, port, bind_address, download=True, protocol="tcp", retry=0):197 """Run iperf3 and return the json results.198 Args:199 ip (str): The ip address of the iperf3 server.200 port (str): The port of the iperf3 server.201 bind_address (str): The wireless interface ip202 address of the client which is being used to203 benchmark.204 download (bool), optional: True if testing download,205 False if testing upload. Defaults to True.206 protocol (str), optional: 'tcp' if testing using tcp207 protocol. 'udp' if testing using udp protocol.208 Defaults to 'tcp'209 retry (int), optional: The retry count.210 Returns:211 dict: Dictionary containing the iperf3 results.212 """213 client = iperf3.Client()214 client.server_hostname = ip215 client.port = port216 client.bind_address = bind_address217 client.reverse = False218 client.verbose = False219 if download:220 client.reverse = True221 client.protocol = protocol222 with suppress_stdout_stderr():223 iperf_result = iperf_result_json = iperf_result.json225 try:226 get_property_from(iperf_result_json, "start")227 get_property_from(iperf_result_json, "end")228 except ValueError:229 logging.error("Output from iperf3 : {0}".format(iperf_result_json))230 logging.exception("Unable to parse iperf3 result")231 if retry == 2:232 raise ExternalError("External Error generated from iperf3: {0}".format(iperf_result_json["error"])) from\233 None234 else:235 logging.warning("Rerunning iperf3 with retry count {0}".format(retry + 1))236 run_iperf(ip, port, bind_address, download, protocol, retry + 1)237 return iperf_result_json238@log_arguments239def run_speedtest(mode, bind_address, libre_speed_server_list=None, retry=0):240 """Run speedtest and return the json results.241 Args:242 mode (SpeedTestMode): The speedtest backend to use243 for benchmark.244 bind_address (str): The wireless interface ip245 address of the client which is being used to246 benchmark.247 libre_speed_server_list (str), optional: The248 path to the librespeed server json file.249 Default is None which forces librespeed to use250 global list.251 retry (int), optional: The retry count.252 Returns:253 dict: Dictionary containing the speedtest results.254 """255 try:256 if mode == SpeedTestMode.OOKLA:257 try:258 speedtest_result = json.loads(get_application_output(["speedtest", "-f", "json", "-i", bind_address],259 timeout=120))260 except ValueError:261 raise ParseError("Unable to decode output from Speedtest Ookla") from None262 return speedtest_result263 elif mode == SpeedTestMode.SIVEL:264 try:265 speedtest_result = json.loads(get_application_output(["speedtest", "--json", "--source", bind_address],266 timeout=120))267 except ValueError:268 raise ParseError("Unable to decode output from Speedtest Sivel") from None269 return speedtest_result270 elif mode == SpeedTestMode.LIBRESPEED:271 libre_args = ["librespeed-cli", "--json", "--source", bind_address, "--mebibytes"]272 if libre_speed_server_list is not None:273 if not os.path.isfile((libre_speed_server_list)):274 raise OSError("Invalid server list specified for libre office")275 libre_speed_server_list = os.path.abspath(libre_speed_server_list)276 libre_args += ["--local-json", libre_speed_server_list]277 logging.debug("Libre Args: {0}".format(libre_args))278 try:279 librespeed_result = json.loads(get_application_output(libre_args, timeout=120))280 except ValueError:281 raise ParseError("Unable to decode output from Librespeed CLI") from None282 return librespeed_result283 except ParseError as err:284 logging.exception("Parse Error has occured.")285 if retry == 2:286 raise err287 else:288 logging.warning("Rerunning Speedtest with retry count {0}".format(retry + 1))289 run_speedtest(mode, bind_address, libre_speed_server_list, retry + 1)290@log_arguments291def test_libre_speed(bind_address, libre_speed_server_list=None):292 """Test user provided server list for librespeed.293 Args:294 bind_address (str): The wireless interface ip295 address of the client which is being used to296 benchmark.297 libre_speed_server_list (str): The298 path to the librespeed server json file.299 Returns:300 bool: True if librespeed works with the301 user provided server list, otherwise False.302 """303 libre_args = ["librespeed-cli", "--json", "--source", bind_address, "--no-download", "--no-upload", "--no-icmp"]304 if not os.path.isfile((libre_speed_server_list)):305 raise OSError("Invalid server list specified for libre office")306 libre_speed_server_list = os.path.abspath(libre_speed_server_list)307 libre_args += ["--local-json", libre_speed_server_list]308 logging.debug("Libre Args: {0}".format(libre_args))309 try:310 json.loads(get_application_output(libre_args, timeout=120))311 except ValueError:312 return False313 return True314def check_speedtest():315 """Detect the speedtest installed and available.316 Args:317 None318 Returns:319 class (SpeedTestMode)320 """321 speedtest_result = get_application_output(["speedtest", "--version"], timeout=10)322 if "Speedtest by Ookla" in speedtest_result:323 return SpeedTestMode.OOKLA324 elif importlib.util.find_spec("speedtest") is not None:325 return SpeedTestMode.SIVEL326 return SpeedTestMode.UNKNOWN327def save_json(file_path, data):328 """Save a json dictionary to disk.329 Args:330 file_path (str): Path to the json331 file.332 data (dict): json dictionary to be saved.333 Returns:334 bool: True if json dictionary was saved,335 False otherwise.336 """337 try:338 with open(file_path, "w") as f:339 json.dump(data, f, indent=4)340 return True341 except:342 return False343def load_json(file_path):344 """Read a json dictionary from disk.345 Args:346 file_path (str): Path to the json347 file.348 Returns:349 dict or bool: json dictionary350 if file was read successfully.351 False if it failed to read.352 """353 try:354 with open(file_path, "r") as f:355 return json.load(f)356 except:357 return False358def get_property_from(dict, key):359 """Get a key value from a dictionary.360 Args:361 dict (dict): Dictionary to read362 from.363 key (str): key in the dictionary364 to get the value for.365 Returns:366 object: Containing the value.367 Raises:368 ValueError: When key does not369 exist in the dictionary.370 """371 try:372 return dict[key]373 except KeyError:374 raise ValueError("Could not retrieve property {0}".format(key)) from None375def bytes_to_human_readable(bytes, ndigits=2, limit=None):376 """Convert bytes to human readable format.377 Args:378 bytes (int): Size in bytes.379 ndigits (int), optional: Number of decimal380 places to round the human readable size to.381 Defaults to 2.382 limit (int), optional: Limit to a predefined383 unit size.384 Returns:385 tuple: Tuple containing the readable bytes386 in float, unit size in float, unit suffix387 in str.388 """389 if limit is None:390 for limit, suffix in HUMAN_BYTE_SIZE:391 if bytes >= limit:392 break393 if limit == 1 and bytes > 1:394 suffix += "s"395 readable_bytes = round((bytes / limit), ndigits)396 return (readable_bytes, limit, suffix)397def get_ip_address_from_interface(interface):398 """Get the IPv4 address of an interface.399 Args:400 target_interface (str): The network interface to401 get the ip address for.402 Returns:403 str or None: Returns the IPv4 address of the404 interface, returns None if no IPv4 address405 exists for that interface.406 """407 data = json.loads(get_application_output(["ip", "-br", "--json", "addr", "show"]))408 for datum in data:409 if datum["ifname"] == interface:410 for address in datum["addr_info"]:411 ip_addr = address["local"]412 if validate_ipv4(ip_addr):413 return ip_addr414 return None415def validate_ipv4(ip_address):416 """Validate an IPv4 address.417 Args:418 ip_address (str): The IPv4 address to check.419 Returns:420 bool: True if a valid IPv4 address, False421 otherwise....

Full Screen

Full Screen Github


Full Screen

...68 return AwaitableGetApplicationResult(69,70 get_application_output(name: Optional[pulumi.Input[str]] = None,73 opts: Optional[pulumi.InvokeOptions] = None) -> pulumi.Output[GetApplicationResult]:74 """75 ## # Application Resource76 [Applications API]( ## Example Usage78 ```python79 import pulumi80 import pulumi_fusionauth as fusionauth81 fusion_auth = fusionauth.get_application(name="FusionAuth")82 ```83 :param str name: The name of the Application.84 """...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:


You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run avocado automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?