Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use test_multiprocess method in Molotov

Best Python code snippet using molotov_python

multiptest.py

Source:multiptest.py

...60    if ns:61        print("NS is set to %s" % ns)62        print(os.getpid())63        print(ns.memes.nd.dreams())64def test_multiprocess(ns=None):65    """66    Simplification of the steps to test mutliprocessing module connectivity67    :param SyncManager.Namespace ns: shared namespace68    """69    rp = Process(target=multicalled, args=(ns,))70    rp.start()71    rp.join()72def test_subprocess():73    """74    Simplification of the steps to test subprocessing module connectivity75    """76    p = subprocess.Popen(["python3.7", "multi2test.py"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)77    p.wait()78if __name__ == "__main__":79    print("Launching Server process")80    mgr = themanager()81    SP = Process(target=mgr.get_server().serve_forever)82    SP.start()83    print("\nTesting multiprocess connect")84    test_multiprocess()85    print("\nTesting subprocess connect")86    test_subprocess()87    # Testing with a SyncManager object instead88    # Nonsensical with subprocesses89    print("\nTesting with a syncmanager namespace")90    test_multiprocess(mainc)91    """92    print("\nTesting with a syncmanager namespace")93    NS = Manager().Namespace()94    NS.A = A95    test_multiprocess(NS)96    print("Setting a new value")97    B = NS.A98    B.value = "a brand new value"99    NS.A = B100    test_multiprocess(NS)101    """...

helper.py

Source:helper.py

1#!/usr/bin/env python2# -*- coding: utf-8 -*-3from __future__ import print_function4from random import randint, sample5from pytq import Task6n = 207input_data_queue2 = [i for i in range(1, n + 1)]8input_data_queue1 = list(sample(input_data_queue2, int(n / 4.0)))9class HashAndProcessImplement(object):10    def user_hash_input(self, input_data):11        """12        1 -> "1"13        """14        return str(input_data)15    def user_process(self, input_data):16        """17        1 -> 100018        """19        return input_data * 100020def validate_schduler_implement(scheduler,21                                test_multiprocess=True,22                                processes=None):23    """24    :param scheduler: the scheduler you wanna validate with.25    :param test_multiprocess: whether we test with multiprocess.26    :param processes: number of processes we want to use, don't exceed max27        db connection number!28    """29    # Reset a scheduler30    scheduler.clear_all()31    scheduler.log_on()32    scheduler.do(input_data_queue1, ignore_error=False)33    assert len(scheduler) == len(input_data_queue1)34    scheduler.do(input_data_queue2, ignore_error=False)35    assert len(scheduler) == len(input_data_queue2)36    for input_data in input_data_queue2:37        task = Task(id=scheduler._hash_input(38            input_data), input_data=input_data)39        assert scheduler._is_duplicate(task) is True40    for input_data in input_data_queue2:41        scheduler.get_output(input_data)42        assert scheduler.get_output(43            input_data) == scheduler.user_process(input_data)44    for id, output_data in scheduler.items():45        assert int(id) * 1000 == output_data46    if test_multiprocess:47        scheduler.clear_all()48        scheduler.do(49            input_data_queue1,50            multiprocess=True, processes=processes,51            ignore_error=False,52        )53        assert len(scheduler) == len(input_data_queue1)54        scheduler.do(55            input_data_queue2, multiprocess=True, processes=processes,56            ignore_error=False,57        )58        assert len(scheduler) == len(input_data_queue2)59        for id, output_data in scheduler.items():...

multi_process.py

Source:multi_process.py

...18    start = time.time()19    time.sleep(random.random() * 3)20    cost = time.time() - start21    print(f'Task {name} runs {cost:.2f} seconds.')22def test_multiprocess():23    """24    Test multiprocessing.25    """26    print('Parent process %s.' % os.getpid())27    p = Pool(4)28    for i in range(5):29        p.apply_async(long_time_task, args=(i,))30    print('Waiting for all subprocesses done...')31    p.close()32    p.join()33    print('All subprocesses done.')34def write(q: Queue) -> None:35    """36    Write data to a queue.37    """38    print(f'Process[{os.getpid()}] start to write...')39    for v in ['A', 'B', 'C']:40        q.put(v)41        print(f'Put {v} to queue...')42        time.sleep(random.random())43    print(f'Process[{os.getpid()}] to write done.')44def read(q: Queue) -> None:45    """46    Read data from a queue.47    """48    print(f'Process[{os.getpid()}] start to read...')49    while True:50        value = q.get(True)51        print(f'Get {value} from queue...')52def test_communication_among_processes():53    """54    Test communication among processes.55    """56    q = Queue()57    pw = Process(target=write, args=(q,))58    pr = Process(target=read, args=(q,))59    pw.start()60    pr.start()61    pw.join()62    pr.terminate()63if __name__ == '__main__':64    # test_multiprocess()...

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.