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

How to use check_stopped method in locust

Best Python code snippet using locust

wiacallback.py

Source:wiacallback.py

...45    stopped = False46    exc = None47    def BandedDataCallback(self, this, lMessage, lStatus, lPercentComplete, lOffset, lLength, lReserved, lResLength, pbBuffer):48        cls = DropboxWiaDataCallbackImpl49        def check_stopped():50            if cls.stopped:51                raise _TransferStopped()52        try:53            message = CALLBACK_MESSAGES.get(lMessage)54            TRACE('WIA message received: %r', message)55            check_stopped()56            if message == 'IT_MSG_DATA':57                datum = string_at(pbBuffer, lLength)58                cls.queue.put(datum, block=True, timeout=DATA_QUEUE_TIMEOUT)59            check_stopped()60            return 061        except _TransferStopped:62            TRACE('Cancelling WIA transfer because of an external request.')63        except Exception:64            cls.exc = sys.exc_info()65            unhandled_exc_handler(False)66            TRACE('Cancelling WIA transfer because an error occurred.')...

reorg_limit.py

Source:reorg_limit.py

...12    connect_nodes_bi,13    sync_blocks,14)15from time import sleep16def check_stopped(i, timeout=10):17    stopped = False18    for x in xrange(1, timeout):19        ret = check_node(i)20        if ret is None:21            sleep(1)22        else:23            stopped = True24            break25    return stopped26class ReorgLimitTest(BitcoinTestFramework):27    def run_test(self):28        assert(self.nodes[0].getblockcount() == 200)29        assert(self.nodes[2].getblockcount() == 200)30        self.split_network()31        print "Test the maximum-allowed reorg:"32        print "Mine 99 blocks on Node 0"33        self.nodes[0].generate(99)34        assert(self.nodes[0].getblockcount() == 299)35        assert(self.nodes[2].getblockcount() == 200)36        print "Mine competing 100 blocks on Node 2"37        self.nodes[2].generate(100)38        assert(self.nodes[0].getblockcount() == 299)39        assert(self.nodes[2].getblockcount() == 300)40        print "Connect nodes to force a reorg"41        connect_nodes_bi(self.nodes, 0, 2)42        self.is_network_split = False43        sync_blocks(self.nodes)44        print "Check Node 0 is still running and on the correct chain"45        assert(self.nodes[0].getblockcount() == 300)46        self.split_network()47        print "Test the minimum-rejected reorg:"48        print "Mine 100 blocks on Node 0"49        self.nodes[0].generate(100)50        assert(self.nodes[0].getblockcount() == 400)51        assert(self.nodes[2].getblockcount() == 300)52        print "Mine competing 101 blocks on Node 2"53        self.nodes[2].generate(101)54        assert(self.nodes[0].getblockcount() == 400)55        assert(self.nodes[2].getblockcount() == 401)56        print "Sync nodes to force a reorg"57        connect_nodes_bi(self.nodes, 0, 2)58        self.is_network_split = False59        # sync_blocks uses RPC calls to wait for nodes to be synced, so don't60        # call it here, because it will have a non-specific connection error61        # when Node 0 stops. Instead, we explicitly check for the process itself62        # to stop.63        print "Check Node 0 is no longer running"64        assert(check_stopped(0))65        # Dummy stop to enable the test to tear down66        self.nodes[0].stop = lambda: True67if __name__ == '__main__':...

test-send-file-and-cancel-immediately.py

Source:test-send-file-and-cancel-immediately.py

...15                         self.wait_for_ft_caps, None,16                         self.check_new_channel, None,17                         self.check_stopped, None,18                         self.close_channel, self.done]19    def check_stopped(self):20        state_event = self.q.expect ('dbus-signal',21                                        signal='FileTransferStateChanged',22                                        path=self.channel.object_path)23        state, reason = state_event.args24        assert state == cs.FT_STATE_CANCELLED25        assert reason == cs.FT_STATE_CHANGE_REASON_REMOTE_STOPPED26        # try to provide the file27        try:28            self.accept_file()29        except dbus.DBusException, e:30            assert e.get_dbus_name() == cs.NOT_AVAILABLE31        else:32            assert False33class SendFileAndClose (SendFileTest):...

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.