How to use enable_loopback method in autotest

Best Python code snippet using autotest_python Github


Full Screen

...197 @enable_tx.setter198 def enable_tx(self, value: bool) -> None:199 self._enable_tx = value200 @property201 def enable_loopback(self) -> bool:202 """Enable or disable loopback203 Note, loopback can't be enabled together with profile 2.204 """205 return self._enable_loopback206 @enable_loopback.setter207 def enable_loopback(self, value: bool) -> None:208 self._enable_loopback = value209 @property210 def phase_enhancement(self) -> bool:211 """Enable or disable phase enhancement212 If enabled, the data phase will be enhanced such that coherent distance filtering can be213 applied. Given a single reflection from an object, the phase will appear as "flat" around214 the amplitude peak.215 Enabling the phase enhancement increases the processing execution time.216 """217 return self._phase_enhancement218 @phase_enhancement.setter219 def phase_enhancement(self, value: bool) -> None:220 self._phase_enhancement = value221 @property...

Full Screen

Full Screen Github


Full Screen

1#!/usr/bin/python2"""3@file asyncMulticast.py4@author Woong Gyu La a.k.a Chris. <>5 <>6@date March 10, 20167@brief AsyncMulticast Interface8@version 0.19@section LICENSE10The MIT License (MIT)11Copyright (c) 2016 Woong Gyu La <>12Permission is hereby granted, free of charge, to any person obtaining a copy13of this software and associated documentation files (the "Software"), to deal14in the Software without restriction, including without limitation the rights15to use, copy, modify, merge, publish, distribute, sublicense, and/or sell16copies of the Software, and to permit persons to whom the Software is17furnished to do so, subject to the following conditions:18The above copyright notice and this permission notice shall be included in19all copies or substantial portions of the Software.20THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR21IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,22FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE23AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER24LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,25OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN26THE SOFTWARE.27@section DESCRIPTION28AsyncMulticast Class.29"""30import queue31import asyncio32import socket33import traceback34import threading35from .serverConf import *36from .callbackInterface import *37from .asyncController import AsyncController38# noinspection PyDeprecation39import copy40IP_MTU_DISCOVER = 1041IP_PMTUDISC_DONT = 0 # Never send DF frames.42IP_PMTUDISC_WANT = 1 # Use per route hints.43IP_PMTUDISC_DO = 2 # Always DF.44IP_PMTUDISC_PROBE = 3 # Ignore dst pmtu.45'''46Interfaces47variables48- callback_obj49functions50- def send(multicast_addr,port,data)51- def close() # close the socket52- def join(multicast_addr) # start receiving datagram from given multicast group53- def leave(multicast_addr) # stop receiving datagram from given multicast group54- def getgrouplist() # get group list55infos56- multicast address range: - linux : route add -net netmask dev eth0 58 to enable multicast59'''60class AsyncMulticast(asyncio.Protocol):61 # enable_loopback : 1 enable loopback / 0 disable loopback62 # ttl: 0 - restricted to the same host63 # 1 - restricted to the same subnet64 # 32 - restricted to the same site65 # 64 - restricted to the same region66 # 128 - restricted to the same continent67 # 255 - unrestricted in scope68 def __init__(self, port, callback_obj, ttl=1, enable_loopback=False, bind_addr=''):69 # self.lock = threading.RLock()70 self.MAX_MTU = 150071 self.callback_obj = None72 self.port = port73 self.multicastSet = set([])74 self.lock = threading.RLock()75 self.ttl = ttl76 self.enable_loopback = enable_loopback77 if callback_obj is not None and isinstance(callback_obj, IUdpCallback):78 self.callback_obj = callback_obj79 else:80 raise Exception('callback_obj is None or not an instance of IUdpCallback class')81 try:82 self.sock= socket.socket(socket.AF_INET, socket.SOCK_DGRAM)83 self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)84 try:85 self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)86 except AttributeError:87 pass # Some systems don't support SO_REUSEPORT88 # for both SENDER and RECEIVER to restrict the region89 self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, self.ttl)90 # for SENDER to choose whether to use loop back91 if self.enable_loopback:92 self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 1)93 else:94 self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 0)95 self.bind_addr = bind_addr96 if self.bind_addr is None or self.bind_addr == '':97 self.bind_addr = socket.gethostbyname(socket.gethostname())98 # for both SENDER and RECEIVER to bind to specific network adapter99 self.sock.setsockopt(socket.SOL_IP, socket.IP_MULTICAST_IF, socket.inet_aton(self.bind_addr))100 # for RECEIVE to receive from multiple multicast groups101 self.sock.bind(('', port))102 except Exception as e:103 print(e)104 traceback.print_exc()105 106 self.transport=None107 AsyncController.instance().add(self)108 if self.callback_obj is not None:109 self.callback_obj.on_started(self)110 self.loop = asyncio.get_event_loop()111 coro = self.loop.create_datagram_endpoint(lambda: self, sock=self.sock)112 AsyncController.instance().pause()113 (self.transport,_)=self.loop.run_until_complete(coro)114 AsyncController.instance().resume()115 # Even though UDP is connectionless this is called when it binds to a port116 def connection_made(self, transport):117 self.transport=transport118 # This is called everytime there is something to read119 def data_received(self, data, addr):120 try:121 if data and self.callback_obj is not None:122 self.callback_obj.on_received(self, addr, data)123 except Exception as e:124 print(e)125 traceback.print_exc()126 127 def connection_lost(self, exc):128 self.close()129 def close(self):130 self.handle_close()131 def error_received(self, exc):132 self.handle_close()133 def handle_close(self):134 try:135 delete_set = self.getgrouplist()136 for multicast_addr in delete_set:137 self.sock.setsockopt(socket.SOL_IP, socket.IP_DROP_MEMBERSHIP,138 socket.inet_aton(multicast_addr) + socket.inet_aton(''))139 if self.callback_obj is not None:140 self.callback_obj.on_leave(self, multicast_addr)141 with self.lock:142 self.multicastSet = set([])143 except Exception as e:144 print(e)145 print('asyncUdp close called')146 self.transport.close()147 AsyncController.instance().discard(self)148 try:149 if self.callback_obj is not None:150 self.callback_obj.on_stopped(self)151 except Exception as e:152 print(e)153 traceback.print_exc()154 # noinspection PyMethodOverriding155 def send(self, hostname, port, data):156 if len(data) <= self.MAX_MTU:157 self.transport.sendto(data,(hostname,port))158 else:159 raise ValueError("The data size is too large")160 # for RECEIVER to receive datagram from the multicast group161 def join(self, multicast_addr):162 with self.lock:163 if multicast_addr not in self.multicastSet:164 self.sock.setsockopt(socket.SOL_IP, socket.IP_ADD_MEMBERSHIP,165 socket.inet_aton(multicast_addr) + socket.inet_aton(self.bind_addr))166 self.multicastSet.add(multicast_addr)167 if self.callback_obj is not None:168 self.callback_obj.on_join(self, multicast_addr)169 # for RECEIVER to stop receiving datagram from the multicast group170 def leave(self, multicast_addr):171 with self.lock:172 try:173 if multicast_addr in self.multicastSet:174 self.sock.setsockopt(socket.SOL_IP, socket.IP_DROP_MEMBERSHIP,175 socket.inet_aton(multicast_addr) + socket.inet_aton(''))176 self.multicastSet.discard(multicast_addr)177 if self.callback_obj is not None:178 self.callback_obj.on_leave(self, multicast_addr)179 except Exception as e:180 print(e)181 def getgrouplist(self):182 with self.lock:183 return copy.copy(self.multicastSet)184 def gethostbyname(self, arg):185 return self.sock.gethostbyname(arg)186 def gethostname(self):187 return self.sock.gethostname()188# Echo udp server test189# def readHandle(sock,addr, data):190# sock.send(addr[0],addr[1],data)...

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 autotest 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?