How to use recv_with_timeout method in Airtest

Best Python code snippet using Airtest Github


Full Screen

...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="">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) =[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: 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="">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: 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: 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: 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: 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: 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: send_sid_request: Bad transport protocol specified"266 return267 if (sock<0):268 print "ERROR: 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: 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: 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] ...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:


You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run Airtest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?