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

How to use ping_check method in avocado

Best Python code snippet using avocado_python

connectivity_logger.py

Source:connectivity_logger.py

1#!/usr/bin/env python2'''3Short script for logging your internet connectivity history in a CSV file. The intention4is to gather data to tell allow you to tell objectively how reliable your ISP really is5and allow you to compare that data to others.6Intended to be run as a cron job.7Only ping tests are supported at the moment. HTTPS connection tests are planned next.8Copyright Â© 2021 Guido Winkelmann9Licensed under the terms of the GNU General Public License version 210@package connectivity_logger11'''12import socket13import re14import time15import os16from datetime import datetime, timezone17from configparser import ConfigParser18from enum import Enum19from subprocess import Popen, PIPE, TimeoutExpired20from ipaddress import ip_address21CheckResult = Enum("CheckResult",22                   ["OK",23                    "UNREACHABLE",24                    "UNRESOLVABLE",25                    "PACKET_LOSS",26                    "UNROUTABLE_IP",27                    "BAD_CERTIFICATE",28                    "UNKNOWN"])29ping_summary_regex = re.compile("(\d+) packets transmitted, (\d+) received, (\d+)")30class PingCheck:31    def __init__(self, name, configuration_section, num_pings, interval):32        self.name = name33        self.protocol = configuration_section.get("protocol", "any")34        self.hostname = configuration_section["hostname"]35        self.non_global_okay = configuration_section.getboolean("non_global_okay", False)36        self.num_pings = num_pings37        self.interval = interval38        39        self.used_address = ""40        self.used_protocol = ""41    def start_check(self):42        try:43            addresses = socket.getaddrinfo(self.hostname, None)44        except socket.gaierror:45            return46        for address in addresses:47            family, sock_type, proto, _, sockaddr = address48            current_address, *_ = sockaddr49            if (sock_type == socket.SocketKind.SOCK_RAW and50                not hasattr(self, "process")):51                    if (family == socket.AddressFamily.AF_INET and 52                        self.protocol in ["ipv4", "any"]):53                            self.process = Popen(54                                ["ping", "-c", str(self.num_pings), "-i", str(self.interval), current_address],55                                stdout=PIPE,56                                env={"LC_ALL": "C", "LANG": "C"})57                            self.used_address = ip_address(current_address)58                            self.used_protocol = "ipv4"59                    if (family == socket.AddressFamily.AF_INET6 and 60                        self.protocol in ["ipv6", "any"]):61                            self.process = Popen(62                                ["ping6", "-c", str(self.num_pings), "-i", str(self.interval), current_address],63                                stdout=PIPE,64                                env={"LC_ALL": "C", "LANG": "C"})65                            self.used_address = ip_address(current_address)66                            self.used_protocol = "ipv6"67    def get_check_results(self, timeout):68        if not hasattr(self, "process"):69            return CheckResult.UNRESOLVABLE70        71        try:72            self.process.wait(timeout=timeout)73        except TimeoutExpired:74            print("%s timed out" % self.name)75            self.process.kill()76            return CheckResult.UNREACHABLE77        unroutable_ip = False78        if not self.used_address.is_global and not self.non_global_okay:79            unroutable_ip = True80        for line in self.process.stdout:81            if b"packets transmitted" in line:82                match = ping_summary_regex.search(line.decode(encoding="ascii"))83                self.sent = int(match.group(1))84                self.received = int(match.group(2))85                self.packet_loss = int(match.group(3))86                if unroutable_ip:87                    return CheckResult.UNROUTABLE_IP88                elif self.received == self.sent:89                    return CheckResult.OK90                elif self.received > 0:91                    return CheckResult.PACKET_LOSS92                else:93                    return CheckResult.UNREACHABLE94        return CheckResult.UNKNOWN # Only reached if we did not see a ping summary line95def get_configuration(paths):96    configuration = ConfigParser()97    configuration_read = False98    paths_iter = iter(paths)99    while not configuration_read:100        try:101            with open(next(paths_iter), "r") as file:102                configuration.read_file(file)103                configuration_read = True104        except FileNotFoundError:105            pass106    return configuration107if __name__ == '__main__':108    configuration = get_configuration([109        "connectivity_logger.cfg",110        os.path.expanduser("~/.connectivity_logger.cfg"),111        "/etc/connectivity_logger.cfg",112        os.path.join(os.path.dirname(os.path.realpath(__file__)), "connectivity_logger.cfg")]113        )114    # We assume this script will be run once per minute, so we set ourselves a deadline115    # for all checks for 45 seconds from now. This should give us enough time to finish116    # up before the next start.117    # TODO Also use a file lock118    deadline = time.time() + 45119    check_time = datetime.now(tz=timezone.utc)120    num_pings = configuration["connectivity_logger"].getint("pings", 5)121    ping_interval = configuration["connectivity_logger"].getfloat("ping_interval", 1.0)122    ping_checks = []123    for section in configuration:124        if section not in ['DEFAULT', 'connectivity_logger']:125            if configuration[section].get("protocol", "any") == "both":126                ping_check_v4 = PingCheck(section, configuration[section],127                                          num_pings=num_pings,128                                          interval=ping_interval)129                ping_check_v4.protocol = "ipv4"130                ping_check_v4.start_check()131                ping_checks.append(ping_check_v4)132                ping_check_v6 = PingCheck(section, configuration[section],133                                          num_pings=num_pings,134                                          interval=ping_interval)135                ping_check_v6.protocol = "ipv6"136                ping_check_v6.start_check()137                ping_checks.append(ping_check_v6)138            else:139                ping_check = PingCheck(section, configuration[section],140                                       num_pings=num_pings,141                                       interval=ping_interval)142                ping_check.start_check()143                ping_checks.append(ping_check)144    145    with open(configuration["connectivity_logger"].get("logfile", "connectivity_log.csv"), "a") as logfile:146        for ping_check in ping_checks:147            result = ping_check.get_check_results(timeout=deadline - time.time())148            logfile.write("{};{};{};{};{};{};{}\n".format(149                check_time.strftime("%Y-%m-%d %H:%M UTC"),150                ping_check.name, # TODO Escape name for CSV format151                ping_check.used_protocol,152                ping_check.used_address,153                result.name,154                getattr(ping_check, "sent", ""),...

conntest.py

Source:conntest.py

...5class ConnectionTester:6    @staticmethod7    def conn_check(host, port):8        print("socket good" if ConnectionTester.check_socket(host, port, 1) == 0 else "socket failure")9        print("ping good" if ConnectionTester.ping_check(host) == 0 else "ping failure for host")10        print("dns good" if ConnectionTester.ping_check("www.google.com") == 0 else "dns failure")11        print("network connection good" if ConnectionTester.ping_check("8.8.8.8") == 0 else "network failure")12        print("loopback good" if ConnectionTester.ping_check("127.0.0.1") == 0 else "loopback failure")13        print("\n")14    @staticmethod15    def check_socket(host, port, timeout=1):16        # https://stackoverflow.com/a/3537000817        # https://docs.python.org/3/library/socket.html#socket.socket.settimeout18        with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:19            sock.settimeout(timeout)20            return sock.connect_ex((host, port))21    @staticmethod22    def ping_check(hostname):23        # -c in windows is routing compartment (admin command)24        parm = "-n" if "Windows" in os.environ["OS"] else "-c"25        count = 126        ping_command = "ping {} {} {}".format(parm, count, hostname)27        # response = os.system(ping_command)28        # https://stackoverflow.com/questions/5596911/python-os-system-without-the-output29        response = check_call(ping_command, stdout=DEVNULL, stderr=STDOUT)30        # return "Connection to host is good" if response == 0 else "Check Connection"...

main.py

Source:main.py

1from multiping import MultiPing2import time3import traceback4import sys5def ping_check():6    mp = MultiPing(["8.8.8.8", "youtube.com", "127.0.0.1", "191.1.1.1"])7    try:8        mp.send()9        time.sleep(10)10        responses, no_responses = mp.receive(1)11        for good_respond in responses:12            print("this address pinges successfully ", good_respond)13        if no_responses:14            for bad_respond in no_responses:15                print("Address ", bad_respond, "Wasnt pinged seccuessfully!")16                if no_responses:17                     sys.exit(2)18            ping_check()19    except:20        print("There is an Error")21        sys.exit(3)22    ping_check()23while True:24    try:25        ping_check()26    except:27        print('There is a problem!!!')28        traceback.print_exc()29        sys.exit(3)...

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.