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

How to use start_thread method in localstack

Best Python code snippet using localstack_python

exp.5.domotica.py

Source:exp.5.domotica.py

...33BEAT_TIME = 0.2534def start_thread_do(function, delay=0.0, params=[]):35    sleep(delay)36    Thread(target=function, args=params).start()37def start_thread(function, delay=0.0, params=[]):38    Thread(target=start_thread_do, args=[function, delay, params]).start()39def nature_rain_beginning():40    sample(S_RAIN, amp=2, sustain=15, release=10)41    start_thread(nature_wind_beginning, 7.55)42def nature_wind_beginning():43    use_fx(['ixi_techno'])44    sample(S_WIND, amp=2, sustain=10, release=4, finish=(14/S_WIND_LENGTH))45    use_fx([])46    start_thread(house_external_wind_chime, (5.99 - 4.625), [4.625])47def house_external_wind_chime(bit_crusher):48    bit_crusher_p = bit_crusher/S_WIND_CHIME_LENGTH49    end = 15.7050    end_p = end/S_WIND_CHIME_LENGTH51    sample(S_WIND_CHIME, amp=4, finish=bit_crusher_p)52    sleep(bit_crusher)53    use_fx(['bitcrusher', 'wobble', 'reverb'])54    sample(S_WIND_CHIME, amp=11, release=5, start=bit_crusher_p, finish=end_p)55    use_fx([])56    start_thread(director, 2.43)57def director():58    global BEAT59    while BEAT < 14*16:60        if BEAT < 1*16:     # 061            start_thread(rhythmic_foundation, 0, [False, True])62        elif BEAT < 2*16:   # 1-763            start_thread(rhythmic_foundation, 0, [True, True])64        elif BEAT < 8*16:   # 8-965            start_thread(rhythmic_foundation, 0, [True, True])66            start_thread(rhythmic_support, 0, [1])67        elif BEAT < 10*16:  # 10-1168            start_thread(rhythmic_foundation, 0, [True, True])69            start_thread(rhythmic_support, 0, [2])70        elif BEAT < 12*16:  # 12-1471            start_thread(rhythmic_foundation, 0, [True, True, True])72            start_thread(rhythmic_support, 0, [3])73        # prepare next beat and sleep until then74        sleep(BEAT_TIME)75        BEAT += 176def rhythmic_telephone_ring():77    use_fx(['reverb'])78    sample(S_TELEPHONE_RING_3, amp=2)79    use_fx([])80def rhythmic_echoed():81    use_fx(['bitcrusher', 'echo'])82    sample(S_BUTTON_7, amp=0.8, rate=0.5)83    use_fx([])84def rhythmic_happy():85    use_fx(['reverb', 'rhpf'])86    sample(S_BUTTON_2, amp=1.5, rate=1, start=0.5/1.326)87    use_fx([])88def rhythmic_foundation(telephone=False, echo=False, happy=False):89    if BEAT % 16 == 0:90        if telephone:91            start_thread(rhythmic_telephone_ring)92        if echo:93            start_thread(rhythmic_echoed, BEAT_TIME*6)94            start_thread(rhythmic_echoed, BEAT_TIME*8)95        if happy:96            start_thread(rhythmic_happy, BEAT_TIME*12)97def rhythmic_support_15():98    use_fx(['reverb'])99    sample(S_BUTTON_15, amp=0.25, rate=0.8755)100    use_fx([])101def rhythmic_support_17():102    use_fx(['krush'])103    sample(S_BUTTON_17, amp=0.8, start=0.15, finish=0.5)104    use_fx([])105def rhythmic_support_17_grave():106    use_fx(['krush'])107    sample(S_BUTTON_17, amp=0.8, rate=0.75, start=0.15, finish=0.4)108    use_fx([])109def rhythmic_support_47():110    sample(S_BUTTON_47, amp=0.8, start=(0.10/0.17))111def rhythmic_support(p_type=0):112    if BEAT % 8 == 0:113        if p_type == 1:114            start_thread(rhythmic_support_17, BEAT_TIME*4)115            start_thread(rhythmic_support_47, BEAT_TIME*6)116        if p_type == 2:117            start_thread(rhythmic_support_17, BEAT_TIME*0)118            start_thread(rhythmic_support_17_grave, BEAT_TIME*2)119            start_thread(rhythmic_support_17, BEAT_TIME*4)120            start_thread(rhythmic_support_47, BEAT_TIME*6)121        if p_type == 3:122            start_thread(rhythmic_support_17, BEAT_TIME*0)123            start_thread(rhythmic_support_17_grave, BEAT_TIME*2)124            start_thread(rhythmic_support_17, BEAT_TIME*4)125            start_thread(rhythmic_support_47, BEAT_TIME*6)126            start_thread(rhythmic_support_15, BEAT_TIME*0)127            start_thread(rhythmic_support_15, BEAT_TIME*1)128            start_thread(rhythmic_support_15, BEAT_TIME*2)129            start_thread(rhythmic_support_15, BEAT_TIME*3)130            start_thread(rhythmic_support_15, BEAT_TIME*4)131            start_thread(rhythmic_support_15, BEAT_TIME*5)132            start_thread(rhythmic_support_15, BEAT_TIME*6)133            start_thread(rhythmic_support_15, BEAT_TIME*7)134# Main135#BEAT = 8*16136start_thread(nature_rain_beginning)...

demo_test_multi_thread.py

Source:demo_test_multi_thread.py

...17    # runs after test steps complete18    thread_3 = MThread(func, 3, 20, extra_arg="run for a long time")19    thread_4 = MThread(nova_helper.create_flavor, 'threading', 'auto', vcpus=2, ram=1024)20    LOG.tc_step("Starting threads")21    thread_1.start_thread()22    thread_2.start_thread()23    thread_3.start_thread()24    thread_4.start_thread()25    LOG.tc_step("Finished starting threads")26    LOG.tc_step("Waiting for threads to finish")27    thread_1.wait_for_thread_end()28    thread_2.wait_for_thread_end()29    thread_4.wait_for_thread_end()30    LOG.tc_step("Threads have finished")31    id_ = thread_4.get_output()[1]32    LOG.info("flav_id = {}".format(id_))33    ResourceCleanup.add(resource_type='flavor', resource_id=id_)34def test_copies_of_threads():35    LOG.tc_step("Make multiple threads with same params")36    threads = []37    for i in range(1, 5):38        threads.append(MThread(func, i, 10, extra_arg="number: {}".format(i)))39    for thread in threads:40        thread.start_thread()41    for thread in threads:42        thread.wait_for_thread_end()43def test_timing():44    threads = []45    flav_id = nova_helper.create_flavor('thread_testing')[1]46    ResourceCleanup.add(resource_type='flavor', resource_id=flav_id)47    start_1 = time()48    for i in range(0, 6):49        thread = MThread(vm_helper.boot_vm, 'threading_vm', flavor=flav_id)50        thread.start_thread(240)51        threads.append(thread)52    for thread in threads:53        thread.wait_for_thread_end()54    for thread in threads:55        ResourceCleanup.add(resource_type='vm', resource_id=thread.get_output()[1])56    end_1 = time()57    start_2 = time()58    for i in range(0, 2):59        vm_id = vm_helper.boot_vm('loop_vm', flav_id)[1]60        ResourceCleanup.add(resource_type='vm', resource_id=vm_id)61    end_2 = time()62    LOG.info("Time results:\n"63             "Multithreading: {}\n"64             "Single loop: {}".format(end_1 - start_1, end_2 - start_2))65def events_func(func_num, reps, event):66    for i in range(0, reps):67        if i > reps / 2:68            event.wait_for_event(30)69        LOG.info("Function #{}".format(func_num))70        sleep(1)71def test_events():72    e = Events("functions should wait here")73    LOG.tc_step("Create multiple threads")74    thread_1 = MThread(events_func, 1, 10, e)75    thread_2 = MThread(events_func, 2, 15, e)76    thread_1.start_thread(60)77    thread_2.start_thread(60)78    sleep(20)79    LOG.tc_step("Setting event")80    e.set()81    thread_1.wait_for_thread_end()82    thread_2.wait_for_thread_end()83    LOG.tc_step("Threads have finished")84    e.clear()85    e.wait_for_event(20, fail_ok=True)86def barr_func(func_num, rep, barrier):87    barrier.wait(10)88    for i in range(0, rep):89        LOG.info("function #{}".format(func_num))90        sleep(1)91def test_barriers():92    LOG.tc_step("Negative barrier example (not enough threads waiting)")93    barrier = TiSBarrier(2, timeout=20)94    thread_1 = MThread(barr_func, 1, 4, barrier)95    thread_1.start_thread(timeout=30)96    thread_1.wait_for_thread_end(fail_ok=True)97    LOG.tc_step("Positive barrier example")98    barrier = TiSBarrier(2, timeout=20)99    thread_1 = MThread(barr_func, 2, 4, barrier)100    thread_2 = MThread(barr_func, 3, 4, barrier)101    thread_1.start_thread(timeout=30)102    thread_2.start_thread(timeout=30)103    thread_1.wait_for_thread_end()104    thread_2.wait_for_thread_end()105def get_lock(lock, th_num):106    sleep(1)107    LOG.info("{} getting lock".format(th_num))108    if lock.acquire():109        LOG.info("{} got lock".format(th_num))110        sleep(5)111        LOG.info("{} release lock".format(th_num))112    else:113        LOG.info("Didn't get lock")114    lock.release()115    LOG.info("{} released lock".format(th_num))116def get_lock_with(lock, th_num):117    sleep(1)118    LOG.info("{} getting lock".format(th_num))119    with lock as got_lock:120        if got_lock:121            LOG.info("{} got lock".format(th_num))122            sleep(5)123            LOG.info("{} release lock".format(th_num))124        else:125            LOG.info("Didn't get lock")126    LOG.info("{} released lock".format(th_num))127def test_lock():128    LOG.tc_step("Positive lock example")129    lock = TiSLock(True)130    thread_1 = MThread(get_lock, lock, 1)131    thread_2 = MThread(get_lock, lock, 2)132    thread_1.start_thread(30)133    sleep(1)134    thread_2.start_thread(30)135    thread_1.wait_for_thread_end(0)136    thread_2.wait_for_thread_end(30)137    LOG.tc_step("Negative lock example")138    lock = TiSLock(True, 2)139    thread_1 = MThread(get_lock, lock, 1)140    thread_2 = MThread(get_lock, lock, 2)141    thread_1.start_thread(30)142    sleep(1)143    thread_2.start_thread(30)144    thread_1.wait_for_thread_end(0, fail_ok=True)...

test_threading_progress.py

Source:test_threading_progress.py

1import pytest2from napari._qt import qthreading3from napari._qt.widgets.qt_progress_bar import QtLabeledProgressBar4pytest.importorskip(5    'qtpy', reason='Cannot test threading progress without qtpy.'6)7def test_worker_with_progress(qtbot):8    test_val = [0]9    def func():10        yield 111        yield 112    def test_yield(v):13        test_val[0] += 114    thread_func = qthreading.thread_worker(15        func,16        connect={'yielded': test_yield},17        progress={'total': 2},18        start_thread=False,19    )20    worker = thread_func()21    with qtbot.waitSignal(worker.yielded):22        worker.start()23        assert worker.pbar.n == test_val[0]24def test_function_worker_nonzero_total_warns():25    def not_a_generator():26        return27    with pytest.warns(RuntimeWarning):28        thread_func = qthreading.thread_worker(29            not_a_generator,30            progress={'total': 2},31            start_thread=False,32        )33        thread_func()34def test_worker_may_exceed_total(qtbot):35    test_val = [0]36    def func():37        yield 138        yield 139    def test_yield(v):40        test_val[0] += 141    thread_func = qthreading.thread_worker(42        func,43        connect={'yielded': test_yield},44        progress={'total': 1},45        start_thread=False,46    )47    worker = thread_func()48    with qtbot.waitSignal(worker.yielded):49        worker.start()50        if test_val[0] < 2:51            assert worker.pbar.n == test_val[0]52        else:53            assert worker.pbar.total == 054def test_generator_worker_with_description():55    def func():56        yield 157    thread_func = qthreading.thread_worker(58        func,59        progress={'total': 1, 'desc': 'custom'},60        start_thread=False,61    )62    worker = thread_func()63    assert worker.pbar.desc == 'custom'64def test_function_worker_with_description():65    def func():66        for _ in range(10):67            pass68    thread_func = qthreading.thread_worker(69        func,70        progress={'total': 0, 'desc': 'custom'},71        start_thread=False,72    )73    worker = thread_func()74    assert worker.pbar.desc == 'custom'75def test_generator_worker_with_no_total():76    def func():77        yield 178    thread_func = qthreading.thread_worker(79        func,80        progress=True,81        start_thread=False,82    )83    worker = thread_func()84    assert worker.pbar.total == 085def test_function_worker_with_no_total():86    def func():87        for _ in range(10):88            pass89    thread_func = qthreading.thread_worker(90        func,91        progress=True,92        start_thread=False,93    )94    worker = thread_func()95    assert worker.pbar.total == 096def test_function_worker_0_total():97    def func():98        for _ in range(10):99            pass100    thread_func = qthreading.thread_worker(101        func,102        progress={'total': 0},103        start_thread=False,104    )105    worker = thread_func()106    assert worker.pbar.total == 0107def test_unstarted_worker_no_widget(make_napari_viewer):108    viewer = make_napari_viewer()109    def func():110        for _ in range(5):111            yield112    thread_func = qthreading.thread_worker(113        func,114        progress={'total': 5},115        start_thread=False,116    )117    thread_func()118    assert not bool(119        viewer.window._qt_viewer.window()._activity_dialog.findChildren(120            QtLabeledProgressBar121        )...

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.