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

How to use on_start method in locust

Best Python code snippet using locust

test_streams.py

Source:test_streams.py

...45    #endregion46    def test_stream_on_document_reference(self):47        ref = self._create(None)["ref"]48        stream = None49        def on_start(event):50            self.assertEqual(event.type, 'start')51            self.assertTrue(isinstance(event.event, int))52            stream.close()53        stream = self.stream_sync(ref, None, on_start=on_start)54        stream.start()55    def test_stream_on_set(self):56        stream = None57        def on_start(evt):58            self._create(None)59        def on_set(evt):60            self.assertEqual(evt.type, 'set')61            self.assertEqual(evt.event['action'], 'add')62            stream.close()63        stream = self.stream_sync(64            query.documents(self.collection_ref),65            on_start = on_start,66            on_set = on_set67        )68        stream.start()69    def test_stream_max_open_streams(self):70        m = 10171        expected = [i for i in range(m)]72        actual = []73        def threadFn(n):74            ref = self._create(n)["ref"]75            stream = None76            def on_start(event):77                self.assertEqual(event.type, 'start')78                self.assertTrue(isinstance(event.event, int))79                self._q(query.update(ref, {"data": {"k": n}}))80            def on_version(event):81                self.assertEqual(event.type, 'version')82                actual.append(n)83                self.assertTrue(isinstance(event.event, dict))84                while(len(actual) != m):85                    sleep(0.1)86                stream.close()87            stream = self.stream_sync(ref, None, on_start=on_start, on_version=on_version)88            stream.start()89        threads = []90        for i in range(m):91            th = Thread(target=threadFn, args=[i])92            th.start()93            threads.append(th)94        for th  in threads:95            th.join()96        actual.sort()97        self.assertEqual(actual, expected)98    def test_stream_reject_non_readonly_query(self):99        q = query.create_collection({"name": "c"})100        stream = None101        def on_error(error):102            self.assertEqual(error.type, 'error')103            self.assertTrue(isinstance(error.error, BadRequest))104            self.assertEqual(error.error._get_description(),105                              'Write effect in read-only query expression.')106            stream.close()107        stream= self.stream_sync(q, on_error=on_error)108        stream.start()109    def test_stream_select_fields(self):110        ref = self._create()["ref"]111        stream = None112        fields = {"document", "diff"}113        def on_start(event):114            self.assertEqual(event.type, 'start')115            self.assertTrue(isinstance(event.event, int))116            self._q(query.update(ref, {"data":{"k": "v"}}))117        118        def on_version(event):119            self.assertEqual(event.type, 'version')120            self.assertTrue(isinstance(event.event, dict))121            self.assertTrue(isinstance(event.txn, int))122            keys = set(event.event.keys())123            self.assertEqual(keys, {"document", "diff"})124            stream.close()125        options = {"fields": list(fields)}126        stream = self.stream_sync(ref, options, on_start=on_start, on_version=on_version)127        stream.start()128    def test_stream_update_last_txn_time(self):129        ref = self._create()["ref"]130        last_txn_time = self.client.get_last_txn_time()131        stream = None132        def on_start(event):133            self.assertEqual(event.type, 'start')134            self.assertTrue(self.client.get_last_txn_time() > last_txn_time)135            #for start event, last_txn_time maybe be updated to response X-Txn-Time header136            # or event.txn. What is guaranteed is the most recent is used- hence >=.137            self.assertTrue(self.client.get_last_txn_time() >= event.txn)138            self._q(query.update(ref, {"data": {"k": "v"}}))139        def on_version(event):140            self.assertEqual(event.type, 'version')141            self.assertEqual(event.txn, self.client.get_last_txn_time())142            stream.close()143        stream = self.stream_sync(ref, on_start=on_start, on_version=on_version)144        stream.start()145    def test_stream_handle_request_failures(self):146        stream=None147        def on_error(event):148            self.assertEqual(event.type, 'error')149            self.assertTrue(isinstance(event.error, BadRequest))150            self.assertEqual(event.error._get_description(),151                             'Expected a Document Ref or Version, got String.')152        stream=self.stream_sync('invalid stream', on_error=on_error )153        stream.start()154    def test_start_active_stream(self):155        ref = self._create(None)["ref"]156        stream = None157        def on_start(event):158            self.assertEqual(event.type, 'start')159            self.assertTrue(isinstance(event.event, int))160            self.assertRaises(FaunaError, lambda: stream.start())161            stream.close()162        stream = self.stream_sync(ref, None, on_start=on_start)163        stream.start()164    def test_stream_auth_revalidation(self):165        ref = self._create()["ref"]166        stream = None167        server_key = self.root_client.query(168            query.create_key({"database": self.db_ref, "role": "server"}))169        client = self.root_client.new_session_client(170            secret=server_key["secret"])171        def on_start(event):172            self.assertEqual(event.type, 'start')173            self.assertTrue(isinstance(event.event, int))174            self.root_client.query(query.delete(server_key["ref"]))175            self.client.query(query.update(ref, {"data": {"k": "v"}}))176        def on_error(event):177            self.assertEqual(event.type, 'error')178            self.assertEqual(event.code, 'permission denied')179            self.assertEqual(event.description,180                             'Authorization lost during stream evaluation.')181            stream.close()182        stream = client.stream(ref, on_start=on_start, on_error=on_error)...

test_component.py

Source:test_component.py

...41class ComponentTestClass(unittest.TestCase):42    def tearDown(self):43        component._ComponentRegistry.components = {}44    def test_start_component(self):45        def on_start(result, c):46            self.assertEquals(c._component_state, "Started")47            self.assertEquals(c.start_count, 1)48        c = testcomponent("test_start_c1")49        d = component.start(["test_start_c1"])50        d.addCallback(on_start, c)51        return d52    def test_start_depends(self):53        def on_start(result, c1, c2):54            self.assertEquals(c1._component_state, "Started")55            self.assertEquals(c2._component_state, "Started")56            self.assertEquals(c1.start_count, 1)57            self.assertEquals(c2.start_count, 1)58        c1 = testcomponent("test_start_depends_c1")59        c2 = testcomponent("test_start_depends_c2")60        c2._component_depend = ["test_start_depends_c1"]61        d = component.start(["test_start_depends_c2"])62        d.addCallback(on_start, c1, c2)63        return d64    def start_with_depends(self):65        c1 = testcomponent_delaystart("test_start_all_c1")66        c2 = testcomponent("test_start_all_c2")67        c3 = testcomponent_delaystart("test_start_all_c3")68        c4 = testcomponent("test_start_all_c4")69        c5 = testcomponent("test_start_all_c5")70        c3._component_depend = ["test_start_all_c5", "test_start_all_c1"]71        c4._component_depend = ["test_start_all_c3"]72        c2._component_depend = ["test_start_all_c4"]73        d = component.start()74        return (d, c1, c2, c3, c4, c5)75    def finish_start_with_depends(self, *args):76        for c in args[1:]:77            component.deregister(c)78    def test_start_all(self):79        def on_start(*args):80            for c in args[1:]:81                self.assertEquals(c._component_state, "Started")82                self.assertEquals(c.start_count, 1)83        ret = self.start_with_depends()84        ret[0].addCallback(on_start, *ret[1:])85        ret[0].addCallback(self.finish_start_with_depends, *ret[1:])86        return ret[0]87    def test_register_exception(self):88        c1 = testcomponent("test_register_exception_c1")89        self.assertRaises(90            component.ComponentAlreadyRegistered,91            testcomponent,92            "test_register_exception_c1")93    def test_stop_component(self):94        def on_stop(result, c):95            self.assertEquals(c._component_state, "Stopped")96            self.assertFalse(c._component_timer.running)97            self.assertEquals(c.stop_count, 1)98        def on_start(result, c):99            self.assertEquals(c._component_state, "Started")100            return component.stop(["test_stop_component_c1"]).addCallback(on_stop, c)101        c = testcomponent_update("test_stop_component_c1")102        d = component.start(["test_stop_component_c1"])103        d.addCallback(on_start, c)104        return d105    def test_stop_all(self):106        def on_stop(*args):107            for c in args[1:]:108                self.assertEquals(c._component_state, "Stopped")109                self.assertEquals(c.stop_count, 1)110        def on_start(*args):111            for c in args[1:]:112                self.assertEquals(c._component_state, "Started")113            return component.stop().addCallback(on_stop, *args[1:])114        ret = self.start_with_depends()115        ret[0].addCallback(on_start, *ret[1:])116        ret[0].addCallback(self.finish_start_with_depends, *ret[1:])117        return ret[0]118    def test_update(self):119        def on_start(result, c1, counter):120            self.assertTrue(c1._component_timer)121            self.assertTrue(c1._component_timer.running)122            self.assertNotEqual(c1.counter, counter)123            return component.stop()124        c1 = testcomponent_update("test_update_c1")125        cnt = int(c1.counter)126        d = component.start(["test_update_c1"])127        d.addCallback(on_start, c1, cnt)128        return d129    def test_pause(self):130        def on_pause(result, c1, counter):131            self.assertEqual(c1._component_state, "Paused")132            self.assertNotEqual(c1.counter, counter)133            self.assertFalse(c1._component_timer.running)134        def on_start(result, c1, counter):135            self.assertTrue(c1._component_timer)136            self.assertNotEqual(c1.counter, counter)137            d = component.pause(["test_pause_c1"])138            d.addCallback(on_pause, c1, counter)139            return d140        c1 = testcomponent_update("test_pause_c1")141        cnt = int(c1.counter)142        d = component.start(["test_pause_c1"])143        d.addCallback(on_start, c1, cnt)144        return d145    def test_shutdown(self):146        def on_shutdown(result, c1):147            self.assertTrue(c1.shutdowned)148            self.assertEquals(c1._component_state, "Stopped")149            self.assertEquals(c1.stop_count, 1)150        def on_start(result, c1):151            d = component.shutdown()152            d.addCallback(on_shutdown, c1)153            return d154        c1 = testcomponent_shutdown("test_shutdown_c1")155        d = component.start(["test_shutdown_c1"])156        d.addCallback(on_start, c1)...

joystick.py

Source:joystick.py

...18                        onMove=self.handle_move,19                        onEnd=self.handle_end)20    def handle_start(self, msg):21        if self.on_start is not None:22            return self.on_start(msg)23        return False24    def handle_move(self, msg):25        if self.on_move is not None:26            return self.on_move(msg)27        return False28    def handle_end(self, msg):29        if self.on_end is not None:30            return self.on_end(msg)31        return False32class Joystick(Element):33    def __init__(self, *,34                 on_start: Optional[Callable] = None,35                 on_move: Optional[Callable] = None,36                 on_end: Optional[Callable] = None,...

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.