Unlock 30% off on Manual Testing Annual Plans this Holiday Season.
Copied to Clipboard!
How to use recv_with_timeout method in Airtest

Best Python code snippet using Airtest
xiaproxy.py
Source:xiaproxy.py
...31    thread.start()32    #thread = threading.Thread(target=showwarning, args=("Invalid Content Hash", "Firefox received bad content"))33    #thread.start()34    #result = showwarning("Invalid Content Hash", "Firefox received content that does not match the reqeusted CID.")35def recv_with_timeout(sock, timeout=5, transport_proto=XSP):36    # Receive data37    start_time = time.time()   # current time in seconds since the epoch38    received_data = False39    reply = '<html><head><title>XIA Error</title></head><body><p>&nbsp;</p><p>&nbsp;</p><p style="text-align: center; font-family: Tahoma, Geneva, sans-serif; font-size: xx-large; color: #666;">Sorry, something went wrong.</p><p>&nbsp;</p><p style="text-align: center; color: #999; font-family: Tahoma, Geneva, sans-serif;"><a href="mailto:xia-dev@cs.cmu.edu">Report a bug</a></p></body></html>'40    try:41        while (time.time() - start_time < timeout and not received_data):42            try:43# we haven't ported Xselect into SWIG yet, and normal select no longer works on xsocket44#               (rl, wl, xl) = select.select([sock], [], [], 0.02)45#               if not rl:46#                   continue47                if transport_proto == XSP:48                    reply = Xrecv(sock, XIA_MAXBUF, 0)49                elif transport_proto == XDP:50                    (reply, reply_dag) = Xrecvfrom(sock, XIA_MAXBUF, 0)51                received_data = True52            except IOError:53                received_data = False54            except Exception, msg:55                print 'ERROR: xiaproxy.py: recv_with_timeout: %s' % msg56    except (KeyboardInterrupt, SystemExit), e:57        Xclose(sock)58        sys.exit()59    if (not received_data):60        print "Recieved nothing"61        raise IOError62        63    if transport_proto == XSP:64        return reply65    elif transport_proto == XDP:66    	return reply, reply_dag67    68    69def readcid_with_timeout(sock, cid, timeout=0.1):70    # Receive data71    start_time = time.time()   # current time in seconds since the epoch72    received_data = False73    reply = '<html><head><title>XIA Error</title></head><body><p>&nbsp;</p><p>&nbsp;</p><p style="text-align: center; font-family: Tahoma, Geneva, sans-serif; font-size: xx-large; color: #666;">Sorry, something went wrong.</p><p>&nbsp;</p><p style="text-align: center; color: #999; font-family: Tahoma, Geneva, sans-serif;"><a href="mailto:xia-dev@cs.cmu.edu">Report a bug</a></p></body></html>'74    try:75        while (time.time() - start_time < timeout and not received_data):76            try:77                status = XgetChunkStatus(sock, cid)78                if status & READY_TO_READ == READY_TO_READ or status & INVALID_HASH == INVALID_HASH:79                    reply = XreadChunk(sock, 65521, 0, cid)80                    received_data = True81                    if status & INVALID_HASH == INVALID_HASH:82                        warn_bad_content()83                        return False84            except IOError:85                received_data = False86            except Exception, msg:87                print 'ERROR: xiaproxy.py: readcid_with_timeout: %s' % msg88    except (KeyboardInterrupt, SystemExit), e:89        Xclose(sock)90        sys.exit()91    if (not received_data):  # TIMEOUT92        print "%s: Recieved nothing; requesting retransmit" % time.time()93        XrequestChunk(sock, cid)94        return readcid_with_timeout(sock, cid)95        raise IOError96    return reply97    98    99def check_for_and_process_CIDs(dstAD, dst4ID, dstHID, message, browser_socket):100    rt = message.find('cid')101    if (rt!= -1):102        http_header = message[0:rt]103        try:104            content = get_content_from_cid_list(dstAD, dst4ID, dstHID, message[rt:].split('.')[2])105        except:106            print "ERROR: xiaproxy.py: check_for_and_process_CIDs: Couldn't retrieve content. Closing browser_socket"107            browser_socket.close()108            return True109        send_to_browser(http_header, browser_socket)110        send_to_browser(content, browser_socket)111        return True112    else:113        return False114def process_videoCIDlist(dstAD, dstHID, message, browser_socket, socks):115    rt = message.find('CID') 116    cidlist = list()117    while(rt != -1):118	CID = message[rt+4:rt+44]119	content_dag = 'CID:%s' % CID120        #content_dag = 'RE %s %s %s' % (AD1, HID1, content_dag)121        content_dag = 'DAG 2 0 - \n %s 2 1 - \n %s 2 - \n %s' % (dstAD, dstHID, content_dag)122	cidlist.append(content_dag)123        #XrequestChunk(moresock, content_dag)124        #content = Xrecv(moresock, 65521, 0)125	#browser_socket.send(content)126	rt = message.find('CID', rt+44)127    ## issue multiple request128    ## and receive multiple content129    ## first issue all the requests130    for i in range(len(cidlist)):131        try:132            XrequestChunk(socks[i], cidlist[i])133        except:134            print 'ERROR: xiaproxy.py: process_videoCIDlist: error requesting CID %s' % cidlist[i]135    ## then retrieve them136    for i in range(len(cidlist)):137        try:138            content = readcid_with_timeout(socks[i], cidlist[i], 2)139	except:140	    browser_socket.close()141	    print "closing browser socket6"142	    return False143        if not send_to_browser(content, browser_socket):144            return False        145    return True146 147def sendVideoSIDRequest(ddag, payload, browser_socket):148    sock = Xsocket(XSOCK_STREAM)149    if (sock<0):150        print "error opening socket"151        return152  153    status = Xconnect(sock, ddag)154    if (status != 0):155       	print "Unexpected error:", sys.exc_info()[0]156       	Xclose(sock)157    	print "send_sid_request() Closing browser socket "158    	browser_socket.close()159	return160    161    print "Connected. OK\n"162    # Send request for number of chunks163    asknumchunks = "numchunks";164    Xsend(sock, asknumchunks, 0)165    #Xsend(sock, payload, 0)166    # Receive reply167    print 'send_sid_request: about to receive reply'168    try:169        reply = recv_with_timeout(sock) # = Xrecv(sock, 65521, 0)170    except:171        Xclose(sock)172    	print "closing browser socket7"173	browser_socket.close()174	return175    Xclose(sock)176    numchunks = int(reply)177    print "send_sid_request: received reply for number of chunks ",numchunks178    ## may be send http header along with first content179    ## return ogg header180    http_header = "HTTP/1.0 200 OK\r\nDate: Tue, 01 Mar 2011 06:14:58 GMT\r\nConnection: close\r\nContent-type: video/ogg\r\nServer: lighttpd/1.4.26\r\n\r\n"181    ## next get chunks, at most 20 in a go182    threshold = 20183    socks = list()184    for i in range(threshold):185	sockcid = Xsocket(XSOCK_CHUNK)186	socks.append(sockcid)187    num_iterations = (numchunks/threshold) + 1188    for i in range(num_iterations):189        st_cid = i * threshold190        end_cid = (i+1) * threshold191        if(end_cid > numchunks):192            end_cid = numchunks193        cidreqrange = str(st_cid) + ":" + str(end_cid)194        print "Requesting for ",cidreqrange195        try:196            sock = Xsocket(XSOCK_STREAM)197            198            status = Xconnect(sock, ddag)199            if (status != 0):200            	print "Unexpected error:", sys.exc_info()[0]201        	Xclose(sock)202    		print "send_sid_request() Closing browser socket "203    		browser_socket.close()204		return205            206            Xsend(sock, cidreqrange, 0)207        except:208            print 'ERROR: xiaproxy.py: sendVideoSIDRequest: error requesting cidreqrange %s' % cidreqrange209	210	try:211	    reply= recv_with_timeout(sock) # = Xrecv(sock, 1024, 0)212    	except:213     	    print "closing browser socket8"214	    browser_socket.close()215	    break;216	Xclose(sock)217	#print reply218    	# Extract dst AD and HID from ddag	219   	start_index = ddag.find('AD:')220    	dstAD = ddag[start_index:start_index+3+40] 221    	start_index = ddag.find('HID:')222    	dstHID = ddag[start_index:start_index+4+40]  	223	224	if(i == 0):225    		send_to_browser(http_header, browser_socket)226	ret = process_videoCIDlist(dstAD, dstHID, reply, browser_socket, socks)227	if (ret==False):228	    break;229	## process CIDs 230    for i in range(threshold):231        Xclose(socks[i])232    return233def requestVideoCID(dstAD, dstHID, CID, fallback):234    sock = Xsocket(XSOCK_CHUNK)235    if (sock<0):236        print "error opening socket"237        return238    # Request content239    content_dag = 'CID:%s' % CID240    if fallback:241        content_dag = 'RE %s %s %s' % (dstAD, dstHID, content_dag)242    #print 'Retrieving content with ID: \n%s' % content_dag243    try:244        XrequestChunk(sock, content_dag)245    except:246        print 'ERROR: xiaproxy.py: requestVideoCID: error requesting CID \n%s' % content_dag247    # Get content248    try:249        data = readcid_with_timeout(sock, content_dag)250        Xclose(sock)251    except:252        return  None253    return data254def getrandSID():255    sid = "SID:"+ ("%040d"% int(random.random()*1e40))256    assert len(sid)==44257    return  sid258def send_sid_request(ddag, payload, browser_socket, transport_proto=XSP):259    # Create socket260    if transport_proto == XSP:261        sock = Xsocket(XSOCK_STREAM)262    elif transport_proto == XDP:263        sock = Xsocket(XSOCK_DGRAM)264    else:265        print "ERROR: xiaproxy.py: send_sid_request: Bad transport protocol specified"266        return267    if (sock<0):268        print "ERROR: xiaproxy.py: send_sid_request: could not open socket"269        return270    271    try:272        if transport_proto == XSP:273            # Connect to service274            status = Xconnect(sock, ddag)275            if (status != 0):276                print "send_sid_request() Closing browser socket "277                print "Unexpected error:", sys.exc_info()[0]278                Xclose(sock)279                browser_socket.close()280                return281        rtt = time.time() 282        # Send request283        if transport_proto == XSP:284            Xsend(sock, payload, 0)285        elif transport_proto == XDP:286            Xsendto(sock, payload, 0, ddag)287    except IOError:288        print 'ERROR: xiaproxy.py: send_sid_request: error binding to sdag, connecting to ddag, or sending SID request:\n%s' % payload289        290    # Receive reply and close socket291    try:292        if transport_proto == XSP:293            reply = ''294            reply += recv_with_timeout(sock)295            content_length_start_index = reply.find('Length: ')296            content_length_end_index = reply.find('\n\n')297            content_length = reply[content_length_start_index+8:content_length_end_index]298            content_length = int(content_length)299            while content_length > (len(reply) - content_length_end_index - 2):300                reply += recv_with_timeout(sock)301        elif transport_proto == XDP:302            (reply, reply_dag) = recv_with_timeout(sock, 5, XDP)303            Xclose(sock);304    except IOError:305        print "ERROR: xiaproxy.py: send_sid_request(): Closing browser socket "306        print "Unexpected error:", sys.exc_info()[0]307        Xclose(sock)308        browser_socket.close()309        return 310    if transport_proto == XSP:311        Xclose(sock)312    # Extract dst AD and HID from ddag	313    start_index = ddag.find('AD:')314    dstAD = ddag[start_index:start_index+3+40] 315    start_index = ddag.find('IP:')316    dst4ID = ddag[start_index:start_index+3+40] ...
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.