Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use ping_async method in lisa

Best Python code snippet using lisa_python

scanner.py

Source:scanner.py

1import netaddr2from datetime import datetime3import time4import re5import os6import asyncio7from modules.logger import logging8from modules.config import config9subnets = config['subnet_dict']10def scan(devices):11    device_count = len(devices)12    logging.debug(f"Scanning {device_count}")13    start_time = time.monotonic()14    results = asyncio.run(main(devices))15    end_time = time.monotonic()16    logging.info(f'Completed {device_count} in {end_time - start_time} seconds')17    return results18    19async def main(devices):20    return await asyncio.gather(*[survey_device(device) for device in devices])21async def survey_device(device):22    ping_string, ping_code = await ping_device(device.id)23    device = interpret_ping_result(ping_string, ping_code, device)24    cur_time = datetime.utcnow().strftime("%Y-%m-%d %H:%M")25    device.time_stamp = cur_time26    if (device.ping_code == 0):27        device.lastup = cur_time28    return device29async def ping_device(device):30    if os.name == 'posix':31        ping_async = await asyncio.create_subprocess_exec(32            'ping',33            f'{device}',34            '-c 1',35            #'-4'36            stdout=asyncio.subprocess.PIPE,37            stderr=asyncio.subprocess.PIPE)38    elif os.name == 'nt':39        ping_async = await asyncio.create_subprocess_shell(40            f'ping {device} /4 /n 1', 41            stdout=asyncio.subprocess.PIPE,42            stderr=asyncio.subprocess.PIPE) 43    stdout, stderr = await ping_async.communicate()44    if stdout:45        ping_result = stdout.decode()46    elif stderr:47        ping_result = stderr.decode()48    ping_code = int(ping_async.returncode)49    if os.name == 'nt' and ping_code == 1:50        ping_code = 251    return ping_result, ping_code52def interpret_ping_result(ping_result, ping_code, device):53    device.ping_code = ping_code54    if os.name == 'nt':55        if device.ping_code == 1:56            device.ping_code = 257        if device.ping_code == 0 and "Destination host unreachable" in ping_result:58            device.ping_code = 159    if device.ping_code == 0:60        device.ip = get_IP(ping_result)61        device.location = get_location(ping_result)62        device.ping_time = get_time(ping_result)63    else:64        device.location = 'Unknown'65        device.ip = '0.0.0.0'66        device.ping_time = '0.0'67    return device68def get_time(string):69    try:70        if os.name == 'posix':71            return str((re.search(r'time=\d+\.\d+', str(string))).group(0)).replace("time=","")72        elif os.name == 'nt':73            return str((re.search(r'time\D\d+', str(string))).group(0)).replace("time","").replace('=','').replace('<','')74    except:75        return '0.0'76def get_IP(string):77    try:78        return (re.search(r'\d+\.\d+\.\d+\.\d+', str(string))).group(0)79    except:80        return '0.0.0.0'81def get_location(string):82    ip = netaddr.IPAddress(get_IP(string))83    for key in subnets:84        if (ip in netaddr.IPNetwork(key)):85            return subnets[key]...

ping.py

Source:ping.py

...13    )14    @property15    def command(self) -> str:16        return "ping"17    def ping_async(18        self,19        target: str = "",20        nic_name: str = "",21        count: int = 5,22        interval: float = 0.2,23        package_size: Optional[int] = None,24        sudo: bool = False,25    ) -> Process:26        if not target:27            target = INTERNET_PING_ADDRESS28        args: str = f"{target} -c {count} -i {interval} -O"29        if nic_name:30            args += f" -I {nic_name}"31        if package_size:32            args += f" -s {package_size}"33        return self.run_async(args, force_run=True, sudo=sudo)34    def ping(35        self,36        target: str = "",37        nic_name: str = "",38        count: int = 5,39        interval: float = 0.2,40        package_size: Optional[int] = None,41        ignore_error: bool = False,42        sudo: bool = False,43    ) -> bool:44        if not target:45            target = INTERNET_PING_ADDRESS46        result = self.ping_async(47            target=target,48            nic_name=nic_name,49            count=count,50            interval=interval,51            package_size=package_size,52            sudo=sudo,53        ).wait_result()54        # for some distro like RHEL, ping with -I nic_name needs sudo55        # otherwise, ping fails with below output:56        # 'ping: SO_BINDTODEVICE: Operation not permitted'57        if not sudo and self._no_permission_pattern.findall(result.stdout):58            result = self.ping_async(59                target=target,60                nic_name=nic_name,61                count=count,62                interval=interval,63                package_size=package_size,64                sudo=True,65            ).wait_result()66        if not ignore_error:67            result.assert_exit_code(68                message=(69                    "failed on ping. The server may not be reached."70                    f" ping result is {result.stdout}"71                ),72            )...

协程版本.py

Source:协程版本.py

...3sem = asyncio.Semaphore(10)4def raw_to_useful_async():5    loop = asyncio.get_event_loop()6    raw_proxies = ['60.13.42.83:9999', '37.187.127.216:8080', '111.194.108.198:9797']7    tasks = [ping_async(proxy) for proxy in raw_proxies]8    loop.run_until_complete(asyncio.wait(tasks))9    for i in tasks:10        print(i)11def valid_useful_proxy_async(proxy):12    ret = yield ping_async(proxy)13    if ret:...14        # self.redis.hset(self.useful_proxy_queue, raw_proxy, 0)15        # self.redis.hdel(self.raw_proxy_queue, raw_proxy)16        # self.log.info('ProxyRefreshSchedule: %s validation pass' % raw_proxy)17    ret = yield ping_async(None)18    if ret:...19        # count += 120        # if count >= FAIL_COUNT:21        #     self.redis.hdel(self.raw_proxy_queue, raw_proxy)22        # else:23        #     self.redis.hset(self.raw_proxy_queue, raw_proxy, count)24        # self.log.info('ProxyRefreshSchedule: %s validation fail' % raw_proxy)25    raise Exception("æ æ³è®¿é®å¤ç½ï¼éªè¯ä»£çç¨åºç»æ¢")26async def ping_async(proxy):27    async with sem:28        conn = aiohttp.TCPConnector(ssl=False)29        async with aiohttp.ClientSession(connector=conn) as session:30            proxy = proxy and "{sch}://{proxy}".format(sch="http", proxy=proxy)31            try:32                async with session.get('http://httpbin.org/ip', proxy=proxy, timeout=1, allow_redirects=False) as response:33                    if response.status == 200 and 'origin' in await response.json():34                        print('éªè¯æå', proxy)35                        return True36                    print('éªè¯...', proxy, response.status)37                    return False38            except:39                print('è¶æ¶äºå§')40                return False...

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.