Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_application_output method in avocado

Best Python code snippet using avocado_python

misc.py

Source:misc.py

...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 subprocess.run(command, 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 = client.run()224    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....

get_application.py

Source:get_application.py

...68    return AwaitableGetApplicationResult(69        id=__ret__.id,70        name=__ret__.name)71@_utilities.lift_output_func(get_application)72def get_application_output(name: Optional[pulumi.Input[str]] = None,73                           opts: Optional[pulumi.InvokeOptions] = None) -> pulumi.Output[GetApplicationResult]:74    """75    ## # Application Resource76    [Applications API](https://fusionauth.io/docs/v1/tech/apis/applications)77    ## 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    """...

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.