KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use get_throughput method in lisa
Best Python code snippet using lisa_python
test_revival_spike_resistant_ema_throughput_measurement.py
Source:test_revival_spike_resistant_ema_throughput_measurement.py 
...55 tm.add_request(5)56 tm.add_request(8)57 assert tm.window_start_ts == 058 assert tm.reqs_in_window == 359 tm.get_throughput(14)60 assert tm.window_start_ts == 061 assert tm.reqs_in_window == 362 # [15, 30)63 tm.get_throughput(15)64 assert tm.window_start_ts == 065 assert tm.reqs_in_window == 366 tm.add_request(16)67 tm.get_throughput(16)68 assert tm.window_start_ts == 1569 assert tm.reqs_in_window == 170 # [30, 45)71 tm.get_throughput(42)72 assert tm.window_start_ts == 1573 assert tm.reqs_in_window == 174def test_rsr_ema_tm_past_windows_processed_on_get_throughput_after_safe_start(tm):75 # [300, 315)76 tm.add_request(301)77 tm.add_request(305)78 tm.add_request(308)79 assert tm.window_start_ts == 30080 assert tm.reqs_in_window == 381 tm.get_throughput(314)82 assert tm.window_start_ts == 30083 assert tm.reqs_in_window == 384 # [315, 330)85 tm.get_throughput(315)86 assert tm.window_start_ts == 31587 assert tm.reqs_in_window == 088 tm.add_request(316)89 tm.get_throughput(316)90 assert tm.window_start_ts == 31591 assert tm.reqs_in_window == 192 # [330, 345)93 tm.get_throughput(342)94 assert tm.window_start_ts == 33095 assert tm.reqs_in_window == 096# TESTS OF STATE MACHINE97@pytest.fixture(scope="function")98def tm_after_start(tm):99 assert tm.state == State.FADED100 return tm101def test_rsr_ema_tm_after_start_stays_in_faded_while_windows_are_empty(tm_after_start):102 tm = tm_after_start103 # [0, 15) - [45, 60)104 # [60, 75)105 assert tm.get_throughput(62) is None106 assert tm.state == State.FADED107 assert tm.throughput == 0108 assert tm.throughput_before_idle == 0109 assert tm.idle_start_ts == 0110 assert tm.empty_windows_count == 0 # it should be 0 because any requests weren't added111def test_rsr_ema_tm_after_start_switches_to_revival_on_not_empty_window(tm_after_start):112 tm = tm_after_start113 # [0, 15) - [30, 45)114 # [45, 60)115 tm.add_request(50)116 # [60, 75)117 tm.add_request(62)118 throughput = tm.get_throughput(62)119 assert tm.state == State.REVIVAL120 assert throughput is None121 assert tm.throughput is None122 assert tm.throughput_before_idle == 0123 assert tm.idle_start_ts == 0124 assert tm.empty_windows_count == 3125 assert tm.revival_start_ts == 45126 assert tm.revival_windows_count == 1127 assert tm.reqs_during_revival == 1128def test_rsr_ema_tm_after_start_switches_to_revival_if_first_window_is_not_empty(tm_after_start):129 tm = tm_after_start130 # [0, 15)131 tm.add_request(14)132 # [15, 30)133 tm.add_request(15)134 assert tm.get_throughput(15) is None135 assert tm.state == State.REVIVAL136 assert tm.throughput is None137 assert tm.throughput_before_idle == 0138 assert tm.idle_start_ts == 0139 assert tm.empty_windows_count == 0140 assert tm.revival_start_ts == 0141 assert tm.revival_windows_count == 1142 assert tm.reqs_during_revival == 1143@pytest.fixture(scope="function")144def tm_in_normal(tm_after_start):145 tm = tm_after_start146 # [0, 15)147 for ts in range(0, 15, 5):148 tm.add_request(ts)149 # [15, 30) - [225, 240) -- up to 16 not empty windows150 tm.add_request(15)151 assert tm.state == State.REVIVAL152 for ts in range(20, 240, 5):153 tm.add_request(ts)154 # [240, 255)155 tm.get_throughput(240)156 assert tm.state == State.NORMAL157 return tm158def test_rsr_ema_tm_in_normal_stays_in_normal_while_windows_are_not_empty(tm_in_normal):159 tm = tm_in_normal160 # [240, 255)161 for ts in range(240, 255, 5):162 tm.add_request(ts)163 # [255, 270)164 tm.add_request(255)165 # [270, 285)166 throughput = tm.get_throughput(272)167 assert tm.state == State.NORMAL168 assert throughput is not None169def test_rsr_ema_tm_in_normal_switches_to_idle_on_empty_window(tm_in_normal):170 tm = tm_in_normal171 # [240, 255)172 tm.add_request(240)173 tm.add_request(245)174 # [255, 270)175 throughput_gotten_before_idle = tm.get_throughput(269)176 # [270, 285)177 throughput = tm.get_throughput(272)178 assert tm.state == State.IDLE179 assert throughput is not None180 assert tm.throughput_before_idle is not None181 assert tm.throughput_before_idle == throughput_gotten_before_idle182 assert tm.idle_start_ts == 255183 assert tm.empty_windows_count == 1184@pytest.fixture(scope="function")185def tm_in_idle_and_throughput_gotten_before_idle(tm_in_normal):186 tm = tm_in_normal187 # [240, 255) - [285, 300)188 for ts in range(240, 300, 5):189 tm.add_request(ts)190 # [300, 315)191 throughput_gotten_before_idle = tm.get_throughput(300)192 assert tm.state == State.NORMAL193 # [315, 330)194 tm.get_throughput(315)195 assert tm.state == State.IDLE196 assert tm.idle_start_ts == 300197 return tm, throughput_gotten_before_idle198@pytest.fixture(scope="function")199def tm_in_idle(tm_in_idle_and_throughput_gotten_before_idle):200 tm, _ = tm_in_idle_and_throughput_gotten_before_idle201 return tm202@pytest.fixture(scope="function")203def throughput_gotten_before_idle(tm_in_idle_and_throughput_gotten_before_idle):204 _, throughput = tm_in_idle_and_throughput_gotten_before_idle205 return throughput206def test_rsr_ema_tm_in_idle_stays_in_idle_while_windows_empty_and_less_min_cnt(207 tm_in_idle, throughput_gotten_before_idle):208 tm = tm_in_idle209 # [315, 330) - [510, 525) -- up to 15 empty windows210 # [525, 540)211 throughput = tm.get_throughput(531)212 assert tm.state == State.IDLE213 assert throughput is not None214 assert tm.throughput_before_idle == throughput_gotten_before_idle215 assert tm.idle_start_ts == 300216 assert tm.empty_windows_count == 15217def test_rsr_ema_tm_in_idle_switches_to_normal_on_not_empty_window(tm_in_idle):218 tm = tm_in_idle219 # [315, 330) - [345, 360)220 # [360, 375)221 tm.add_request(370)222 # [375, 390)223 throughput = tm.get_throughput(381)224 assert tm.state == State.NORMAL225 assert throughput is not None226def test_rsr_ema_tm_in_idle_switches_to_faded_on_min_cnt_empty_windows(227 tm_in_idle, throughput_gotten_before_idle):228 tm = tm_in_idle229 # [315, 330) - [525, 540) -- up to 16 empty windows230 # [540, 555)231 throughput = tm.get_throughput(540)232 assert tm.state == State.FADED233 assert throughput is not None234 assert tm.throughput_before_idle == throughput_gotten_before_idle235 assert tm.idle_start_ts == 300236 assert tm.empty_windows_count == 16237@pytest.fixture(scope="function")238def tm_in_faded(tm_in_idle):239 tm = tm_in_idle240 # [315, 330) - [525, 540) -- up to 16 empty windows241 # [540, 555)242 tm.get_throughput(540)243 assert tm.state == State.FADED244 assert tm.idle_start_ts == 300245 return tm246def test_rsr_ema_tm_in_faded_stays_in_faded_while_windows_are_empty(247 tm_in_faded, throughput_gotten_before_idle):248 tm = tm_in_faded249 # [540, 555) - [585, 600)250 # [600, 615)251 throughput = tm.get_throughput(600)252 assert tm.state == State.FADED253 assert throughput is not None254 assert tm.throughput_before_idle == throughput_gotten_before_idle255 assert tm.idle_start_ts == 300256 assert tm.empty_windows_count == 20257def test_rsr_ema_tm_in_faded_switches_to_revival_on_not_empty_window(258 tm_in_faded, throughput_gotten_before_idle):259 tm = tm_in_faded260 # [540, 555) - [570, 585)261 # [585, 600)262 tm.add_request(590)263 tm.add_request(595)264 # [600, 615)265 throughput = tm.get_throughput(600)266 assert tm.state == State.REVIVAL267 assert throughput is None268 assert tm.throughput_before_idle == throughput_gotten_before_idle269 assert tm.idle_start_ts == 300270 assert tm.empty_windows_count == 19271 assert tm.revival_start_ts == 585272 assert tm.revival_windows_count == 1273 assert tm.reqs_during_revival == 2274@pytest.fixture(scope="function")275def tm_in_revival(tm_in_faded):276 tm = tm_in_faded277 # [540, 555) - [585, 600)278 # [600, 615)279 for ts in range(600, 615, 5):280 tm.add_request(ts)281 # [615, 630)282 tm.get_throughput(615)283 assert tm.state == State.REVIVAL284 assert tm.idle_start_ts == 300285 assert tm.revival_start_ts == 600286 return tm287def test_rsr_ema_tm_in_revival_stays_in_revival_while_windows_not_empty_and_less_min_cnt(288 tm_in_revival, throughput_gotten_before_idle):289 tm = tm_in_revival290 # [615, 630) - [810, 825) -- up to 15 not empty windows291 for ts in range(615, 825, 5):292 tm.add_request(ts)293 # [825, 840)294 throughput = tm.get_throughput(829)295 assert tm.state == State.REVIVAL296 assert throughput is None297 assert tm.throughput_before_idle == throughput_gotten_before_idle298 assert tm.idle_start_ts == 300299 assert tm.empty_windows_count == 20300 assert tm.revival_start_ts == 600301 assert tm.revival_windows_count == 15302 assert tm.reqs_during_revival == 45303def test_rsr_ema_tm_in_revival_switches_to_normal_on_min_cnt_not_empty_windows(tm_in_revival):304 tm = tm_in_revival305 # [615, 630) - [810, 840) -- up to 16 not empty windows306 for ts in range(615, 840, 5):307 tm.add_request(ts)308 # [840, 855)309 throughput = tm.get_throughput(840)310 assert tm.state == State.NORMAL311 assert throughput is not None312def test_rsr_ema_tm_in_revival_switches_to_idle_on_empty_window(tm_in_revival):313 tm = tm_in_revival314 # [615, 630)315 tm.add_request(615)316 # [630, 645)317 # [645, 660)318 throughput = tm.get_throughput(649)319 assert tm.state == State.IDLE320 assert throughput is not None321 assert tm.throughput_before_idle is not None322 assert tm.throughput_before_idle > throughput323 assert tm.idle_start_ts == 630324 assert tm.empty_windows_count == 1325# TESTS OF THROUGHPUT CALCULATION326def test_rsr_ema_tm_throughput_in_safe_start(tm):327 # [0, 15)328 assert tm.get_throughput(2) is None329 # [15, 30)330 tm.add_request(15)331 assert tm.get_throughput(15) is None332 assert tm.throughput is not None333 # [240, 255) - after safe start334 assert tm.get_throughput(240) is not None335def test_rsr_ema_tm_throughput_in_normal_state(tm):336 # [0, 15) - [225, 240) -- 16 not empty windows337 for ts in range(0, 240, 5):338 tm.add_request(ts)339 # [240, 255)340 throughput_before = tm.get_throughput(240)341 assert tm.state == State.NORMAL342 for ts in range(240, 255, 1): # load increases343 tm.add_request(ts)344 # [255, 270)345 throughput = tm.get_throughput(255)346 assert tm.state == State.NORMAL347 assert isclose(throughput,348 (2 / 17) * 1 + (1 - 2 / 17) * throughput_before)349def test_rsr_ema_tm_throughput_on_switch_from_normal_to_idle_state(tm):350 # [0, 15) - [225, 240) -- 16 not empty windows351 for ts in range(0, 240, 5):352 tm.add_request(ts)353 # [240, 255)354 throughput_before = tm.get_throughput(254)355 assert tm.state == State.NORMAL356 # [255, 270)357 throughput = tm.get_throughput(255)358 assert tm.state == State.IDLE359 assert isclose(throughput,360 (2 / 17) * 0 + (1 - 2 / 17) * throughput_before)361def test_rsr_ema_tm_throughput_in_idle_state(tm):362 # [0, 15) - [225, 240) -- 16 not empty windows363 for ts in range(0, 240, 5):364 tm.add_request(ts)365 # [240, 255)366 # [255, 270)367 throughput_before = tm.get_throughput(269)368 assert tm.state == State.IDLE369 # [270, 285)370 throughput = tm.get_throughput(270)371 assert tm.state == State.IDLE372 assert isclose(throughput,373 (2 / 17) * 0 + (1 - 2 / 17) * throughput_before)374def test_rsr_ema_tm_throughput_on_switch_from_idle_to_normal_state(tm):375 # [0, 15) - [225, 240) -- 16 not empty windows376 for ts in range(0, 240, 5):377 tm.add_request(ts)378 # [240, 255)379 # [255, 270)380 throughput_before = tm.get_throughput(255)381 assert tm.state == State.IDLE382 for ts in range(255, 270, 1): # load increases after pause383 tm.add_request(ts)384 # [270, 285)385 throughput = tm.get_throughput(270)386 assert tm.state == State.NORMAL387 assert isclose(throughput,388 (2 / 17) * 1 + (1 - 2 / 17) * throughput_before)389def test_rsr_ema_tm_throughput_on_switch_from_idle_to_faded_state(tm):390 # [0, 15) - [225, 240) -- 16 not empty windows391 for ts in range(0, 240, 5):392 tm.add_request(ts)393 # [240, 255) - [465, 480) -- 16 empty windows394 throughput_before = tm.get_throughput(479)395 assert tm.state == State.IDLE396 # [480, 495)397 throughput = tm.get_throughput(480)398 assert tm.state == State.FADED399 assert isclose(throughput,400 (2 / 17) * 0 + (1 - 2 / 17) * throughput_before)401def test_rsr_ema_tm_throughput_in_faded_state(tm):402 # [0, 15) - [225, 240) -- 16 not empty windows403 for ts in range(0, 240, 5):404 tm.add_request(ts)405 # [240, 255) - [465, 480) -- 16 empty windows406 # [480, 495)407 throughput_before = tm.get_throughput(494)408 assert tm.state == State.FADED409 # [495, 510)410 throughput = tm.get_throughput(495)411 assert tm.state == State.FADED412 assert isclose(throughput,413 (2 / 17) * 0 + (1 - 2 / 17) * throughput_before)414def test_rsr_ema_tm_throughput_on_switch_from_revival_to_normal_state(tm):415 # [0, 15) - [225, 240) -- 16 not empty windows416 for ts in range(0, 240, 3):417 tm.add_request(ts)418 # [240, 255) - [705, 720) -- 32 empty windows419 throughput_before_idle = tm.get_throughput(240)420 assert tm.state == State.NORMAL421 # [720, 960) -- 16 not empty windows422 for ts in range(720, 960, 1): # spike occurs on revival423 tm.add_request(ts)424 tm.get_throughput(959)425 assert tm.state == State.REVIVAL426 # [960, 975)427 throughput = tm.get_throughput(960)428 assert tm.state == State.NORMAL429 thr = throughput_before_idle430 # load is leveled over IDLE, FADED and REVIVAL periods431 for ts in range(255, 975, 15):432 thr = (2 / 17) * (1 / 3) + (1 - 2 / 17) * thr433 expected_throughput = thr434 assert isclose(throughput, expected_throughput)435def test_rsr_ema_tm_throughput_on_switch_from_revival_to_idle_state(tm):436 # [0, 15) - [225, 240) -- 16 not empty windows437 for ts in range(0, 240, 3):438 tm.add_request(ts)439 # [240, 255) - [585, 600) -- 24 empty windows440 throughput_before_idle = tm.get_throughput(240)441 assert tm.state == State.NORMAL442 # [600, 780) -- 12 not empty windows443 for ts in range(600, 780, 1): # spike occurs on revival but then load stops444 tm.add_request(ts)445 # [780, 795)446 tm.get_throughput(794)447 assert tm.state == State.REVIVAL448 # [795, 810)449 throughput = tm.get_throughput(795)450 assert tm.state == State.IDLE451 thr = throughput_before_idle452 # load is leveled over IDLE, FADED and REVIVAL periods453 for ts in range(255, 795, 15):454 thr = (2 / 17) * (1 / 3) + (1 - 2 / 17) * thr455 thr = (2 / 17) * 0 + (1 - 2 / 17) * thr456 expected_throughput = thr...
broker_zmq_results.py
Source:broker_zmq_results.py
...11 for per in percentiles:12 results[f"{per}%"] = content[str(number)]["latency distribution"][count]13 count += 114 return results15def get_throughput(number, io_length, kind):16 with open(17 f"measurements/long_zmq_{io_length}/formatted_results_{kind}.txt", "r"18 ) as file:19 content = loads(file.read())20 return content[str(number)]["throughput"]21latency_threads_1 = {22 1: get_latency((1, 1), 1, "worker_threads"),23 2: get_latency(2, 1, "thread"),24 4: get_latency(4, 1, "thread"),25 8: get_latency(8, 1, "thread"),26 16: get_latency(16, 1, "thread"),27 32: get_latency(32, 1, "thread"),28 64: get_latency(64, 1, "thread"),29 128: get_latency(128, 1, "thread"),30}31throughput_threads_1 = {32 1: get_throughput((1, 1), 1, "worker_threads"),33 2: get_throughput(2, 1, "thread"),34 4: get_throughput(4, 1, "thread"),35 8: get_throughput(8, 1, "thread"),36 16: get_throughput(16, 1, "thread"),37 32: get_throughput(32, 1, "thread"),38 64: get_throughput(64, 1, "thread"),39 128: get_throughput(128, 1, "thread"),40}41latency_workers_1 = {42 1: get_latency((1, 1), 1, "worker_threads"),43 2: get_latency(2, 1, "worker"),44 4: get_latency(4, 1, "worker"),45 8: get_latency(8, 1, "worker"),46 16: get_latency(16, 1, "worker"),47 32: get_latency(32, 1, "worker"),48 64: get_latency(64, 1, "worker"),49 128: get_latency(128, 1, "worker"),50}51throughput_workers_1 = {52 1: get_throughput((1, 1), 1, "worker_threads"),53 2: get_throughput(2, 1, "worker"),54 4: get_throughput(4, 1, "worker"),55 8: get_throughput(8, 1, "worker"),56 16: get_throughput(16, 1, "worker"),57 32: get_throughput(32, 1, "worker"),58 64: get_throughput(64, 1, "worker"),59 128: get_throughput(128, 1, "worker"),60}61latency_threads_worker_1 = {62 (2, 32): get_latency((2, 32), 1, "worker_threads"),63 (3, 32): get_latency((3, 32), 1, "worker_threads"),64 (4, 16): get_latency((4, 16), 1, "worker_threads"),65 (3, 16): get_latency((3, 16), 1, "worker_threads"),66 (2, 64): get_latency((2, 64), 1, "worker_threads"),67}68throughput_threads_worker_1 = {69 (2, 32): get_throughput((2, 32), 1, "worker_threads"),70 (3, 32): get_throughput((3, 32), 1, "worker_threads"),71 (4, 16): get_throughput((4, 16), 1, "worker_threads"),72 (3, 16): get_throughput((3, 16), 1, "worker_threads"),73 (2, 64): get_throughput((2, 64), 1, "worker_threads"),74}75latency_threads_50 = {76 1: get_latency((1, 1), 50, "worker_threads"),77 2: get_latency(2, 50, "thread"),78 4: get_latency(4, 50, "thread"),79 8: get_latency(8, 50, "thread"),80 16: get_latency(16, 50, "thread"),81 32: get_latency(32, 50, "thread"),82 64: get_latency(64, 50, "thread"),83 128: get_latency(128, 50, "thread"),84}85throughput_threads_50 = {86 1: get_throughput((1, 1), 50, "worker_threads"),87 2: get_throughput(2, 50, "thread"),88 4: get_throughput(4, 50, "thread"),89 8: get_throughput(8, 50, "thread"),90 16: get_throughput(16, 50, "thread"),91 32: get_throughput(32, 50, "thread"),92 64: get_throughput(64, 50, "thread"),93 128: get_throughput(128, 50, "thread"),94}95latency_workers_50 = {96 1: get_latency((1, 1), 50, "worker_threads"),97 2: get_latency(2, 50, "worker"),98 4: get_latency(4, 50, "worker"),99 8: get_latency(8, 50, "worker"),100 16: get_latency(16, 50, "worker"),101 32: get_latency(32, 50, "worker"),102 64: get_latency(64, 50, "worker"),103 128: get_latency(128, 50, "worker"),104}105throughput_workers_50 = {106 1: get_throughput((1, 1), 50, "worker_threads"),107 2: get_throughput(2, 50, "worker"),108 4: get_throughput(4, 50, "worker"),109 8: get_throughput(8, 50, "worker"),110 16: get_throughput(16, 50, "worker"),111 32: get_throughput(32, 50, "worker"),112 64: get_throughput(64, 50, "worker"),113 128: get_throughput(128, 50, "worker"),114}115latency_threads_worker_50 = {116 (2, 32): get_latency((2, 32), 50, "worker_threads"),117 (3, 32): get_latency((3, 32), 50, "worker_threads"),118 (4, 16): get_latency((4, 16), 50, "worker_threads"),119 (3, 16): get_latency((3, 16), 50, "worker_threads"),120 (2, 64): get_latency((2, 64), 50, "worker_threads"),121}122throughput_threads_worker_50 = {123 (2, 32): get_throughput((2, 32), 50, "worker_threads"),124 (3, 32): get_throughput((3, 32), 50, "worker_threads"),125 (4, 16): get_throughput((4, 16), 50, "worker_threads"),126 (3, 16): get_throughput((3, 16), 50, "worker_threads"),127 (2, 64): get_throughput((2, 64), 50, "worker_threads"),128}129def get_contentt(io_length, kind):130 with open(f"measurements/formatted_{io_length}_{kind}_results.txt", "r") as file:131 content = loads(file.read())132 return content133detailed_thread_latency_1 = get_contentt(1, "threads")134detailed_thread_latency_50 = get_contentt(50, "threads")135detailed_worker_latency_1 = get_contentt(1, "worker")136detailed_worker_latency_50 = get_contentt(50, "worker")137detailed_worker_thread_latency_1 = get_contentt(50, "worker_thread")...
solve.py
Source:solve.py
1def get_throughput(time, start_end_times):2 throughput = 03 start_time = time4 end_time = start_time + 10005 for start_end_time in start_end_times:6 if start_end_time[1] >= start_time and start_end_time[0] < end_time:7 throughput += 18 return throughput9def solution(lines):10 answer = 011 start_end_times = []12 for line in lines:13 _, time, duration = line.split()14 time_arr = time.split(':')15 # msë¡ íë³í16 duration = float(duration[:-1]) * 100017 end_time = (int(time_arr[0]) * 3600 + int(time_arr[1]) * 60 + float(time_arr[2])) * 100018 start_time = end_time - duration + 119 start_end_times.append([start_time, end_time])20 for start_end_time in start_end_times:21 temp1 = get_throughput(start_end_time[0], start_end_times)22 temp2 = get_throughput(start_end_time[1], start_end_times)23 answer = max(answer, temp1, temp2)...
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:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run lisa 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!!
Was this article helpful?
