How to use client_listener method in locust

Best Python code snippet using locust

test_client.py

Source: test_client.py Github

copy

Full Screen

1import threading2from foris_forwarder.client import Client3TIMEOUT = 30.04def test_connection(5 mosquitto_host,6 mosquitto_subordinate,7 prepare_ca,8 connection_settings,9 wait_for_disconnected,10):11 process, settings, _ = connection_settings12 client = Client(settings, keepalive=5)13 assert client.connected is False14 # add post_connect hook15 connect_event = threading.Event()16 def connect(client, userdata, flags, rc):17 connect_event.set()18 client.set_connect_hook(connect)19 # add post_disconnect hook20 disconnect_event = threading.Event()21 def disconnect(client, userdata, rc):22 disconnect_event.set()23 client.set_disconnect_hook(disconnect)24 # test connect25 client.connect()26 assert connect_event.wait(TIMEOUT)27 assert client.connected is True28 # test disconnect29 client.disconnect()30 assert disconnect_event.wait(TIMEOUT)31 assert client.connected is False32 # terminte message bus33 connect_event.clear()34 client.connect()35 assert connect_event.wait(TIMEOUT)36 assert client.connected is True37 disconnect_event.clear()38 process.kill()39 process.wait()40 assert disconnect_event.wait(TIMEOUT)41 assert client.connected is False42 wait_for_disconnected(client)43def test_messaging(mosquitto_host, mosquitto_subordinate, prepare_ca, connection_settings, wait_for_disconnected):44 process, settings1, settings2 = connection_settings45 # prepare listener46 client_listener = Client(settings1)47 subscribe_event = threading.Event()48 def subscribe(client, userdata, mid, granted_qos):49 subscribe_event.set()50 client_listener.set_subscribe_hook(subscribe)51 message_event = threading.Event()52 def message(client, userdata, message):53 message_event.set()54 client_listener.set_message_hook(message)55 # prepare publisher56 client_publisher = Client(settings2)57 publish_event = threading.Event()58 def publish(client, userdata, mid):59 publish_event.set()60 client_publisher.set_publish_hook(publish)61 # wait till listener is subscribed62 client_listener.connect()63 client_listener.wait_until_connected(TIMEOUT)64 client_publisher.connect()65 client_publisher.wait_until_connected(TIMEOUT)66 # wait till listener is subscribed for topic67 assert client_listener.subscribe([("/​messaging-test/​+", 0)])68 assert subscribe_event.wait(TIMEOUT)69 # wait till publisher publishes a message70 assert client_publisher.publish("/​messaging-test/​first", '{"some": "data"}') is not None71 assert publish_event.wait(TIMEOUT)72 # check that the message was obtained by the listener73 assert message_event.wait(TIMEOUT)74 # unsubscribe and wait to be sure that message is not recieved75 unsubscribe_event = threading.Event()76 def unsubscribe(client, userdata, mid):77 unsubscribe_event.set()78 client_listener.set_unsubscribe_hook(unsubscribe)79 assert client_listener.unsubscribe(["/​messaging-test/​+"])80 assert unsubscribe_event.wait(TIMEOUT)81 message_event.clear()82 publish_event.clear()83 assert client_publisher.publish("/​messaging-test/​first", '{"some": "data"}') is not None84 assert publish_event.wait(TIMEOUT)85 assert not message_event.wait(2.0) # no message should be recieved in 2 seconds86 # cleanup87 wait_for_disconnected(client_publisher)88 wait_for_disconnected(client_listener)89 process.kill()...

Full Screen

Full Screen

client.py

Source: client.py Github

copy

Full Screen

...3from time import sleep4from multiprocessing.connection import Listener5from multiprocessing import Process6import sys7def client_listener(info):8 print(f"Openning listener at {info}")9 cl = Listener(address = (info['address'], info['port']), authkey = info['authkey'])10 print('.............client listener starting')11 print('.............accepting conexions')12 while True:13 conn = cl.accept()14 print('.............connection accepted from ', cl.last_accepted)15 m = conn.recv()16 print('.............message received from server', m)17def main(server_address, info):18 print('trying to connect')19 with Client(address = (server_address, 6000), authkey = b'secret password server') as conn:20 cl = Process(target = client_listener, args = (info,))21 cl.start()...

Full Screen

Full Screen

__main__.py

Source: __main__.py Github

copy

Full Screen

1import logging.config2import os3from multiprocessing import Pipe, Process45from src.backup_scheduler.backup_scheduler import BackupScheduler6from src.client_listener.client_listener import ClientListener7from src.database.disk_database import DiskDatabase8910def main() -> int:11 logging.config.fileConfig('log.conf', disable_existing_loggers=True)12 port = int(os.getenv('PORT'))13 listen_backlog = 1014 backup_data_path = os.getenv('BACKUP_DATA_PATH')15 max_backup_processes = int(os.getenv('MAX_BACKUP_PROCESSES'))1617 backup_scheduler_recv, client_listener_send = Pipe(False)18 client_listener_recv, backup_scheduler_send = Pipe(False)1920 database = DiskDatabase(backup_data_path + "/​database")21 backup_scheduler = BackupScheduler(backup_data_path + "/​data",22 database, backup_scheduler_recv,23 backup_scheduler_send, max_backup_processes)24 client_listener = ClientListener(port, listen_backlog,25 client_listener_send, client_listener_recv)26 p = Process(target=client_listener)27 p.start()28 backup_scheduler()29 p.terminate()30 return 1 # This function should never end313233if __name__ == '__main__': ...

Full Screen

Full Screen

Blogs

Check out the latest blogs from LambdaTest on this topic:

How To Handle Dynamic Dropdowns In Selenium WebDriver With Java

Joseph, who has been working as a Quality Engineer, was assigned to perform web automation for the company’s website.

Top 17 Resources To Learn Test Automation

Lack of training is something that creates a major roadblock for a tester. Often, testers working in an organization are all of a sudden forced to learn a new framework or an automation tool whenever a new project demands it. You may be overwhelmed on how to learn test automation, where to start from and how to master test automation for web applications, and mobile applications on a new technology so soon.

Migrating Test Automation Suite To Cypress 10

There are times when developers get stuck with a problem that has to do with version changes. Trying to run the code or test without upgrading the package can result in unexpected errors.

Website Testing: A Detailed Guide

Websites and web apps are growing in number day by day, and so are the expectations of people for a pleasant web experience. Even though the World Wide Web (WWW) was invented only in 1989 (32 years back), this technology has revolutionized the world we know back then. The best part is that it has made life easier for us. You no longer have to stand in long queues to pay your bills. You can get that done within a few minutes by visiting their website, web app, or mobile app.

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run locust automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful