Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use remove_worker method in locust

Best Python code snippet using locust

queue.py

Source:queue.py

...67        bs.remove_build_queue(username, queue_name)68        print("Removed queue %s" % queue_name)69        return70    if args.remove_worker:71        bs.remove_worker(username, queue_name, args.remove_worker)72        print("Removed worker %s from queue %s" % (args.remove_worker, queue_name))73        return74    if queue_name:75        print()76        show_queue(queue)77    else:78        show_queues(bs, username)79def add_parser(subparsers):80    parser = subparsers.add_parser('queue',81                                      help='Inspect build queue',82                                      description=__doc__,83                                      formatter_class=RawDescriptionHelpFormatter,84                                      )85    parser.add_argument('queue', nargs='?', metavar='USERNAME/QUEUENAME',...

test_scheduler_plugin.py

Source:test_scheduler_plugin.py

...22    assert counter.count == 323    s.remove_plugin(counter)24    assert counter not in s.plugins25@gen_cluster(nthreads=[], client=False)26async def test_add_remove_worker(s):27    events = []28    class MyPlugin(SchedulerPlugin):29        def add_worker(self, worker, scheduler):30            assert scheduler is s31            events.append(("add_worker", worker))32        def remove_worker(self, worker, scheduler):33            assert scheduler is s34            events.append(("remove_worker", worker))35    plugin = MyPlugin()36    s.add_plugin(plugin)37    assert events == []38    a = Worker(s.address)39    b = Worker(s.address)40    await a41    await b42    await a.close()43    await b.close()44    assert events == [45        ("add_worker", a.address),46        ("add_worker", b.address),47        ("remove_worker", a.address),48        ("remove_worker", b.address),49    ]50    events[:] = []51    s.remove_plugin(plugin)52    a = await Worker(s.address)53    await a.close()54    assert events == []55@gen_cluster(nthreads=[], client=False)56async def test_async_add_remove_worker(s):57    events = []58    class MyPlugin(SchedulerPlugin):59        async def add_worker(self, worker, scheduler):60            assert scheduler is s61            events.append(("add_worker", worker))62        async def remove_worker(self, worker, scheduler):63            assert scheduler is s64            events.append(("remove_worker", worker))65    plugin = MyPlugin()66    s.add_plugin(plugin)67    assert events == []68    async with Worker(s.address) as a:69        async with Worker(s.address) as b:70            pass71    assert set(events) == {72        ("add_worker", a.address),73        ("add_worker", b.address),74        ("remove_worker", a.address),75        ("remove_worker", b.address),76    }...

test_plugin.py

Source:test_plugin.py

...23    assert counter.count == 324    s.remove_plugin(counter)25    assert counter not in s.plugins26@gen_cluster(nthreads=[], client=False)27def test_add_remove_worker(s):28    events = []29    class MyPlugin(SchedulerPlugin):30        def add_worker(self, worker, scheduler):31            assert scheduler is s32            events.append(("add_worker", worker))33        def remove_worker(self, worker, scheduler):34            assert scheduler is s35            events.append(("remove_worker", worker))36    plugin = MyPlugin()37    s.add_plugin(plugin)38    assert events == []39    a = Worker(s.address)40    b = Worker(s.address)41    yield a42    yield b43    yield a.close()44    yield b.close()45    assert events == [46        ("add_worker", a.address),47        ("add_worker", b.address),...

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.