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

How to use test_start_event method in locust

Best Python code snippet using locust

gateway.py

Source:gateway.py

1"""2A python version of the gateway. Used primarily for bootstraping the rest of3the model.4"""5import threading6from log import now, Log, StandardWriter7from net import Stub, TCPServer8def nth_prime(n):9    def is_prime(v):10        return all(v % k > 0 for k in range(2, v))11    i = 012    j = 113    while i < n+1:14        if is_prime(j):15            i += 116        j += 117    return j - 118def do_cpu(intensity):19    """20    Simulates a CPU intensive task by calculating the nth prime, where n is21    2^10 * intensity, and intensity varies between 0 and 1.22    """23    if intensity < 0 or intensity > 1:24        raise AttributeError(25            'Cannot simulate CPU intensity: value must be between 0 and 1')26    n = int((2 ** 12) * intensity)27    nth_prime(n)28class GatewayAPI():29    """ API callable by the nameservice.30    """31    def __init__(self, start_test_event):32        self.start_test_event = start_test_event33    def get_timestamp(self):34        return now()35    def start_test(self):36        self.start_test_event.set()37class GatewayServer(threading.Thread):38    logger = Log.get_logger('GatewayServer')39    def __init__(self):40        threading.Thread.__init__(self)41        self.daemon = True42        self.test_start_event = threading.Event()43        self.server = TCPServer(('', 0), GatewayAPI(self.test_start_event))44    def run(self):45        while True:46            self.server.accept()47    def hostname(self):48        return self.server.hostname()49    def wait_for_start(self):50        self.test_start_event.wait()51    def close(self):52        self.server.close()53class PassiveGateway():54    """ A passive gateway is a TCP client that requests events from passive55    devices.56    """57    logger = Log.get_logger('PassiveGateway')58    local_logger = Log.get_logger('PassiveGatewayLocal', StandardWriter)59    def __init__(self, nsaddress, cpu_intensity, io_intensity):60        self.server = GatewayServer()61        self.ns = Stub(nsaddress)62        self.cpu_intensity = cpu_intensity63        self.io_intensity = io_intensity64    def start_server(self):65        self.server.start()66    def close_server(self):67        self.server.close()68    def verify_configuration(self, configuration):69        addr = self.server.hostname()70        return self.ns.verify_gateway(configuration, addr)71    def register_devices(self):72        self.devices = self.ns.hostnames()73        PassiveGateway.local_logger.info(74            'Retrieved devices from nameserver: {}',75            self.devices)76    def wait_for_start(self):77        self.server.wait_for_start()78    def get_events(self):79        for port in self.devices:80            device = Stub(('0.0.0.0', port))81            if (device.status() == 1):82                event = device.next_event()83                PassiveGateway.logger.info('EVENT_LIFECYCLE_RETRIEVED:{}',84                        event)85                PassiveGateway.logger.info('EVENT_LIFECYCLE_DISPATCHED:{}',86                        event)87                do_cpu(self.cpu_intensity)88                PassiveGateway.logger.info('EVENT_LIFECYCLE_DONE:{}',...

test_event.py

Source:test_event.py

...14    def test_catch_event(self):15        event_id = '1'16        event = CatchEvent(event_id)17        self.assertTrue(isinstance(event, Event))18    def test_start_event(self):19        event_id = '1'20        event = StartEvent(event_id)21        self.assertTrue(isinstance(event, CatchEvent))22    def test_end_event(self):23        event_id = '1'24        event = EndEvent(event_id)25        self.assertTrue(isinstance(event, ThrowEvent))26    def test_empty_start_event(self):27        event_id = '1'28        event = EmptyStartEvent(event_id)29        self.assertTrue(isinstance(event, StartEvent))30    def test_empty_end_event(self):31        event_id = '1'32        event = EmptyEndEvent(event_id)...

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.