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

How to use _run_client method in autotest

Best Python code snippet using autotest_python

test_cleanup.py

Source:test_cleanup.py

...44    client doesn't disconnect explicitely.45    """46    shmids_before = _get_shmids()47    with FrameBufferServer(buffer_size) as server:48        def _run_client():49            client = FrameBufferClient(server.get_key())50            client.start()51        p = multiprocessing.Process(target=_run_client)52        p.start()53        p.join()54        shmids_during = _get_shmids()55    shmids_after = _get_shmids()56    assert shmids_before == shmids_after57    assert shmids_before != shmids_during58    assert len(shmids_during - shmids_before) == 159def test_shm_cleanup_client_kill():60    """61    Test whether server properly cleans up its shared memory segment after62    client is abruptly terminated.63    """64    client_event = multiprocessing.Event()65    shutdown_event = multiprocessing.Event()66    shmids_before = _get_shmids()67    with FrameBufferServer(buffer_size) as server:68        def _run_client():69            client = FrameBufferClient(server.get_key())70            client.start()71            client_event.set()72            shutdown_event.wait()  # this one will never come :)73        p = multiprocessing.Process(target=_run_client)74        p.start()75        client_event.wait()76        p.kill()77        p.join()78        shmids_during = _get_shmids()79    shmids_after = _get_shmids()80    assert shmids_before == shmids_after81    assert shmids_before != shmids_during82    assert len(shmids_during - shmids_before) == 183def test_shm_cleanup_server_hang():84    """85    Test whether client properly cleans up its shared memory segment after86    server.87    """88    client_event = multiprocessing.Event()89    shutdown_event = multiprocessing.Event()90    shmids_before = _get_shmids()91    with FrameBufferServer(buffer_size) as server:92        def _run_client():93            client = FrameBufferClient(server.get_key())94            client.start()95            client_event.set()96            shutdown_event.wait()97            client.stop()98        p = multiprocessing.Process(target=_run_client)99        p.start()100        client_event.wait()101    shmids_during = _get_shmids()102    shutdown_event.set()103    p.join()104    shmids_after = _get_shmids()105    assert shmids_before == shmids_after106    assert shmids_before != shmids_during107    assert len(shmids_during - shmids_before) == 1108def test_shm_cleanup_server_hang_client_kill():109    """110    Test whether client properly cleans up its shared memory segment after111    server if it's killed.112    """113    client_event = multiprocessing.Event()114    shutdown_event = multiprocessing.Event()115    shmids_before = _get_shmids()116    with FrameBufferServer(buffer_size) as server:117        def _run_client():118            client = FrameBufferClient(server.get_key())119            client.start()120            client_event.set()121            shutdown_event.wait()122            client.stop()123        p = multiprocessing.Process(target=_run_client)124        p.start()125        client_event.wait()126    shmids_during = _get_shmids()127    p.kill()128    p.join()129    shmids_after = _get_shmids()130    assert shmids_before == shmids_after131    assert shmids_before != shmids_during...

streaming_client.py

Source:streaming_client.py

...92# def timeline(on_msg, currency='btccny'):93#     data = {94#         'msgType': 'reqTimeLine',95#     }96#     _run_client(on_msg, data, currency)97#98#99# def kline(on_msg, period='1min', currency='btccny'):100#     data = {101#         'msgType': 'reqKLine',102#         'period': period,103#     }104#     _run_client(on_msg, data, currency)105#106#107# def market_depth_top(on_msg, currency='btccny'):108#     data = {109#         'msgType': 'reqMarketDepthTop',110#     }111#     _run_client(on_msg, data, currency)112#113#114# def market_depth(on_msg, percent=10, currency='btccny'):115#     data = {116#         'msgType': 'reqMarketDepth',117#         'percent': percent,118#     }119#     _run_client(on_msg, data, currency)120#121#122# def trade_detail_top(on_msg, count=None, currency='btccny'):123#     data = {124#         'msgType': 'reqTradeDetailTop',125#     }126#     if count:127#         data['count'] = count128#     _run_client(on_msg, data, currency)129#130#131# def market_detail(on_msg, currency='btccny'):132#     data = {133#         'msgType': 'reqMarketDetail',134#     }135#     _run_client(on_msg, data, currency)136#137#138# def _run_client(on_msg, data, currency):139#     sc = StreamingClient()140#     data['symbolId'] = currency141#     sc.req_data.update(data)142#     del sc.req_data['symbolList']143#     print(sc.req_data)...

close_channel_test.py

Source:close_channel_test.py

...71        gevent.sleep()72        group.killone(greenlet, exception=Exception)73        self.assertFalse(self._unhandled_exception, "Unhandled exception")74        self.assertRaises(Exception, greenlet.get)75    def _run_client(self, call):76        try:77            call.with_call(messages_pb2.SimpleRequest())78        except grpc.RpcError as e:79            if e.code() != grpc.StatusCode.CANCELLED:80                raise81    def _global_exception_handler(self, exctype, value, tb):82        if exctype == gevent.GreenletExit:83            self._unhandled_exception = True84            return85        sys.__excepthook__(exctype, value, tb)86if __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.