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

How to use client_listener method in locust

Best Python code snippet using locust

test_client.py

Source:test_client.py

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()...

client.py

Source:client.py

...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()...

__main__.py

Source:__main__.py

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__':
...

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.