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

How to use socket_timeout method in autotest

Best Python code snippet using autotest_python

robust2.py

Source:robust2.py

1from utime import ticks_ms,ticks_diff2from .  import simple23class MQTTClient(simple2.MQTTClient):4	DEBUG=False;KEEP_QOS0=True;NO_QUEUE_DUPS=True;MSG_QUEUE_MAX=55	def __init__(A,*B,**C):super().__init__(*B,**C);A.msg_to_send=[];A.sub_to_send=[];A.msg_to_confirm={};A.sub_to_confirm={};A.conn_issue=None6	def set_callback_status(A,f):A._cbstat=f7	def cbstat(A,pid,stat):8		E=stat;C=pid9		try:A._cbstat(C,E)10		except AttributeError:pass11		for (D,B) in A.msg_to_confirm.items():12			if C in B:13				if E==0:A.msg_to_send.append(D)14				B.remove(C)15				if not B:A.msg_to_confirm.pop(D)16				return17		for (D,B) in A.sub_to_confirm.items():18			if C in B:19				if E==0:A.sub_to_send.append(D)20				B.remove(C)21				if not B:A.sub_to_confirm.pop(D)22	def connect(A,clean_session=True,socket_timeout=-1):23		B=clean_session24		if B:A.msg_to_send[:]=[];A.msg_to_confirm.clear()25		try:C=super().connect(B,socket_timeout);A.conn_issue=None;return C26		except (OSError,simple2.MQTTException)as D:A.conn_issue=D,127	def log(A):28		if A.DEBUG:29			if type(A.conn_issue)is tuple:B,C=A.conn_issue30			else:B=A.conn_issue;C=031			D='?','connect','publish','subscribe','reconnect','sendqueue','disconnect','ping','wait_msg','keepalive';print('MQTT (%s): %r'%(D[C],B))32	def reconnect(A,socket_timeout=-1):33		try:B=super().connect(False,socket_timeout=socket_timeout);A.conn_issue=None;return B34		except (OSError,simple2.MQTTException)as C:A.conn_issue=C,435	def add_msg_to_send(A,data):36		A.msg_to_send.append(data)37		if len(A.msg_to_send)>A.MSG_QUEUE_MAX:A.msg_to_send.pop(0)38	def disconnect(A,socket_timeout=-1):39		try:return super().disconnect(socket_timeout=socket_timeout)40		except (OSError,simple2.MQTTException)as B:A.conn_issue=B,641	def ping(A,socket_timeout=-1):42		try:return super().ping(socket_timeout=socket_timeout)43		except (OSError,simple2.MQTTException)as B:A.conn_issue=B,744	def publish(A,topic,msg,retain=False,qos=0,socket_timeout=-1):45		E=topic;C=retain;B=qos;D=E,msg,C,B46		if C:A.msg_to_send[:]=[B for B in A.msg_to_send if not(E==B[0]and C==B[2])]47		try:48			F=super().publish(E,msg,C,B,False,socket_timeout)49			if B==1:A.msg_to_confirm.setdefault(D,[]).append(F)50			return F51		except (OSError,simple2.MQTTException)as G:52			A.conn_issue=G,253			if A.NO_QUEUE_DUPS:54				if D in A.msg_to_send:return55			if A.KEEP_QOS0 and B==0:A.msg_to_send.append(D)56			elif B==1:A.msg_to_send.append(D)57	def subscribe(A,topic,qos=0,socket_timeout=-1):58		B=topic;C=B,qos;A.sub_to_send[:]=[C for C in A.sub_to_send if B!=C[0]]59		try:D=super().subscribe(B,qos,socket_timeout);A.sub_to_confirm.setdefault(C,[]).append(D);return D60		except (OSError,simple2.MQTTException)as E:61			A.conn_issue=E,362			if A.NO_QUEUE_DUPS:63				if C in A.sub_to_send:return64			A.sub_to_send.append(C)65	def send_queue(A,socket_timeout=-1):66		H=socket_timeout;D=[]67		for B in A.msg_to_send:68			E,J,K,C=B69			try:70				F=super().publish(E,J,K,C,False,H)71				if C==1:A.msg_to_confirm.setdefault(B,[]).append(F)72				D.append(B)73			except (OSError,simple2.MQTTException)as G:A.conn_issue=G,5;return False74		A.msg_to_send[:]=[B for B in A.msg_to_send if B not in D];del D;I=[]75		for B in A.sub_to_send:76			E,C=B77			try:F=super().subscribe(E,C,H);A.sub_to_confirm.setdefault(B,[]).append(F);I.append(B)78			except (OSError,simple2.MQTTException)as G:A.conn_issue=G,5;return False79		A.sub_to_send[:]=[B for B in A.sub_to_send if B not in I];return True80	def is_conn_issue(A):81		B=ticks_diff(ticks_ms(),A.last_cpacket)//100082		if 0<A.keepalive<B:A.conn_issue=simple2.MQTTException(7),983		if A.conn_issue:A.log()84		return bool(A.conn_issue)85	def wait_msg(A,socket_timeout=None):86		try:return super().wait_msg(socket_timeout)...

rpc.py

Source:rpc.py

1# -*- coding: utf-8 -*-2from __future__ import absolute_import3import contextlib4import warnings5from _shaded_thriftpy.protocol import TBinaryProtocolFactory6from _shaded_thriftpy.server import TThreadedServer7from _shaded_thriftpy.thrift import TProcessor, TClient8from _shaded_thriftpy.transport import (9    TBufferedTransportFactory,10    TServerSocket,11    TSocket,12)13def make_client(service, host="localhost", port=9090, unix_socket=None,14                proto_factory=TBinaryProtocolFactory(),15                trans_factory=TBufferedTransportFactory(),16                timeout=None):17    if unix_socket:18        socket = TSocket(unix_socket=unix_socket)19    elif host and port:20        socket = TSocket(host, port, socket_timeout=timeout)21    else:22        raise ValueError("Either host/port or unix_socket must be provided.")23    transport = trans_factory.get_transport(socket)24    protocol = proto_factory.get_protocol(transport)25    transport.open()26    return TClient(service, protocol)27def make_server(service, handler,28                host="localhost", port=9090, unix_socket=None,29                proto_factory=TBinaryProtocolFactory(),30                trans_factory=TBufferedTransportFactory()):31    processor = TProcessor(service, handler)32    if unix_socket:33        server_socket = TServerSocket(unix_socket=unix_socket)34    elif host and port:35        server_socket = TServerSocket(host=host, port=port)36    else:37        raise ValueError("Either host/port or unix_socket must be provided.")38    server = TThreadedServer(processor, server_socket,39                             iprot_factory=proto_factory,40                             itrans_factory=trans_factory)41    return server42@contextlib.contextmanager43def client_context(service, host="localhost", port=9090, unix_socket=None,44                   proto_factory=TBinaryProtocolFactory(),45                   trans_factory=TBufferedTransportFactory(),46                   timeout=3000, socket_timeout=3000, connect_timeout=None):47    if timeout:48        warnings.warn("`timeout` deprecated, use `socket_timeout` and "49                      "`connect_timeout` instead.")50        socket_timeout = connect_timeout = timeout51    if unix_socket:52        socket = TSocket(unix_socket=unix_socket,53                         connect_timeout=connect_timeout,54                         socket_timeout=socket_timeout)55    elif host and port:56        socket = TSocket(host, port,57                         connect_timeout=connect_timeout,58                         socket_timeout=socket_timeout)59    else:60        raise ValueError("Either host/port or unix_socket must be provided.")61    try:62        transport = trans_factory.get_transport(socket)63        protocol = proto_factory.get_protocol(transport)64        transport.open()65        yield TClient(service, protocol)66    finally:...

test_marionette_arguments.py

Source:test_marionette_arguments.py

...3# file, You can obtain one at http://mozilla.org/MPL/2.0/.4import pytest5from marionette_harness.runtests import MarionetteArguments6@pytest.mark.parametrize("socket_timeout", ['A', '10', '1B-', '1C2', '44.35'])7def test_parse_arg_socket_timeout(socket_timeout):8    argv = ['marionette', '--socket-timeout', socket_timeout]9    parser = MarionetteArguments()10    def _is_float_convertible(value):11        try:12            float(value)13            return True14        except:15            return False16    if not _is_float_convertible(socket_timeout):17        with pytest.raises(SystemExit) as ex:18            parser.parse_args(args=argv)19        assert ex.value.code == 220    else:21        args = parser.parse_args(args=argv)...

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.