Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use enable_loopback method in autotest

Best Python code snippet using autotest_python

subsweep_config.py

Source:subsweep_config.py

...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...

asyncMulticast.py

Source:asyncMulticast.py

1#!/usr/bin/python2"""3@file asyncMulticast.py4@author Woong Gyu La a.k.a Chris. <juhgiyo@gmail.com>5        <http://github.com/juhgiyo/pyserver>6@date March 10, 20167@brief AsyncMulticast Interface8@version 0.19@section LICENSE10The MIT License (MIT)11Copyright (c) 2016 Woong Gyu La <juhgiyo@gmail.com>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: 224.0.0.0 - 239.255.255.25557- linux : route add -net 224.0.0.0 netmask 240.0.0.0 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('0.0.0.0'))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('0.0.0.0'))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)...

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.