KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use start_async method in lisa
Best Python code snippet using lisa_python
test_websocket.py
Source:test_websocket.py 
...30 """"Open then immediately close"""31 def on_open():32 print('opened')33 client.close()34 await client.start_async(on_open=on_open, on_error=on_error)35 print('closed')36@pytest.mark.asyncio37async def test_trades(client, symbols):38 print(symbols)39 results = []40 def callback(msg):41 results.append(msg)42 client.close()43 def on_open():44 client.subscribe_trades(symbols=symbols, callback=callback)45 await client.start_async(on_open=on_open, on_error=on_error)46 result = results[0]47 assert result['stream'] == 'trades'48@pytest.mark.asyncio49async def test_market_diff(client, symbols):50 results = []51 def callback(msg):52 results.append(msg)53 client.close()54 def on_open():55 client.subscribe_market_diff(symbols=symbols, callback=callback)56 await client.start_async(on_open=on_open, on_error=on_error)57 result = results[0]58 assert result['stream'] == 'marketDiff'59@pytest.mark.asyncio60async def test_market_depth(client, symbols):61 results = []62 def callback(msg):63 results.append(msg)64 client.close()65 def on_open():66 client.subscribe_market_depth(symbols=symbols, callback=callback)67 await client.start_async(on_open=on_open, on_error=on_error)68 result = results[0]69 assert result['stream'] == 'marketDepth'70@pytest.mark.asyncio71async def test_kline(client, symbols):72 results = []73 def callback(msg):74 results.append(msg)75 client.close()76 def on_open():77 client.subscribe_kline(interval='1m', symbols=symbols, callback=callback)78 await client.start_async(on_open=on_open, on_error=on_error)79 result = results[0]80 assert result['stream'] == 'kline_1m'81@pytest.mark.asyncio82async def test_tickers(client, symbols):83 results = []84 def callback(msg):85 results.append(msg)86 client.close()87 def on_open():88 client.subscribe_ticker(symbols=symbols, callback=callback)89 await client.start_async(on_open=on_open, on_error=on_error)90 result = results[0]91 assert result['stream'] == 'ticker'92@pytest.mark.asyncio93async def test_all_tickers(client):94 results = []95 def callback(msg):96 results.append(msg)97 client.close()98 def on_open():99 client.subscribe_all_tickers(callback=callback)100 await client.start_async(on_open=on_open, on_error=on_error)101 result = results[0]102 assert result['stream'] == 'allTickers'103@pytest.mark.asyncio104async def test_mini_ticker(client, symbols):105 results = []106 def callback(msg):107 results.append(msg)108 client.close()109 def on_open():110 client.subscribe_mini_ticker(symbols=symbols, callback=callback)111 await client.start_async(on_open=on_open, on_error=on_error)112 result = results[0]113 assert result['stream'] == 'miniTicker'114@pytest.mark.asyncio115async def test_all_mini_ticker(client, symbols):116 results = []117 def callback(msg):118 results.append(msg)119 client.close()120 def on_open():121 client.subscribe_all_mini_tickers(callback=callback)122 await client.start_async(on_open=on_open, on_error=on_error)123 result = results[0]124 assert result['stream'] == 'allMiniTickers'125@pytest.mark.asyncio126async def test_blockheight(client):127 results = []128 def callback(msg):129 results.append(msg)130 client.close()131 def on_open():132 client.subscribe_blockheight(callback=callback)133 await client.start_async(on_open=on_open, on_error=on_error)134 result = results[0]135 assert 'stream' in result136@pytest.mark.asyncio137async def test_keepalive(client):138 def on_open():139 client.keepalive()140 client.close()141 await client.start_async(on_open=on_open, on_error=on_error)142@pytest.mark.asyncio143async def test_unsubscribe(client):144 results = []145 def callback(msg):146 results.append(msg)147 client.unsubscribe("blockheight")148 client.close()149 def on_open():150 client.subscribe_blockheight(callback=callback)151 await client.start_async(on_open=on_open, on_error=on_error)152 assert results153@pytest.mark.asyncio154async def test_decorator(client):155 @client.on('open')156 def callback():157 client.close()158 await client.start_async()159@pytest.mark.asyncio160async def test_decorator_async(client):161 @client.on('open')162 async def callback():163 client.close()164 await client.start_async()165@pytest.mark.asyncio166async def test_decorator_sub_queue(client):167 results = []168 @client.on("allTickers", symbols=["$all"])169 async def callback(msg):170 results.append(msg)171 client.close()172 await client.start_async()...
reid.py
Source:reid.py
...12 plt.imshow(img)13 plt.show()14 print("ffff")15 img = preprocess(img,size)16 infer_res = exec_net.start_async(request_id=0,inputs={input_layer:img})17 status=infer_res.wait()18 results = exec_net.requests[0].outputs[output_layer][0]19 for j,bbox1 in ids1.items():20 print("gggg")21 img1 = frame1[bbox1[1]:bbox1[3], bbox1[0]:bbox1[2]]22 #plt.imshow(img1)23 #plt.show()24 img1 = preprocess(img1,size)25 infer_res = exec_net.start_async(request_id=1,inputs={input_layer:img1})26 status=infer_res.wait()27 results1 = exec_net.requests[1].outputs[output_layer][0]28 print(results1.shape)29 x = torch.tensor(results).unsqueeze(0)30 y = torch.tensor(results1).unsqueeze(0)31 dis=torch.cosine_similarity(x, y)[0].numpy()32 if dis >= 0.5:33 track_id[i]= bbox34 35def reidentification(ids,track_id,frame):36 img=frame37 track_id_copy = track_id.copy()38 for i,bbox in ids.items():39 if len(track_id) == 0:40 track_id[i] = bbox41 else:42 img = frame[bbox[1]:bbox[3], bbox[0]:bbox[2]]43 #plt.imshow(img)44 #plt.show()45 img = preprocess(img,size)46 infer_res = exec_net.start_async(request_id=0,inputs={input_layer:img})47 status=infer_res.wait()48 results = exec_net.requests[0].outputs[output_layer][0]49 dis = []50 for j,bbox_t in track_id_copy.items():51 img1 = frame[bbox_t[1]:bbox_t[3], bbox_t[0]:bbox_t[2]]52 img1 = preprocess(img1,size)53 infer_res = exec_net.start_async(request_id=1,inputs={input_layer:img1})54 status=infer_res.wait()55 results1 = exec_net.requests[1].outputs[output_layer][0]56 #print(results1.shape)57 x = torch.tensor(results).unsqueeze(0)58 y = torch.tensor(results1).unsqueeze(0)59 d=torch.cosine_similarity(x, y)[0].numpy()60 dis.append(d)61 try:62 f_idx = dis.index(max(dis))63 f = dis[f_idx]64 if f >= 0.7:65 print("hhh")66 track_id[f_idx] = bbox67 i = f_idx68 else:69 track_id[i] = bbox70 i=i71 except Exception as e:72 print(e)73 cv2.putText(frame, str(i), bbox[:2], cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)74 cv2.rectangle(img=frame, pt1=bbox[:2], pt2=bbox[2:], color=(0,255,0), thickness=3)75 return track_id,frame76 77 78def emb(ids,frame):79 r= []80 b=[]81 for i,bbox in ids.items():82 img = frame[bbox[1]:bbox[3], bbox[0]:bbox[2]]83 img = preprocess(img,size)84 infer_res = exec_net.start_async(request_id=0,inputs={input_layer:img})85 status=infer_res.wait()86 results = exec_net.requests[0].outputs[output_layer][0]87 #print(type(results))88 r.append(results)89 b.append(bbox)90 91 return r,b92def ids_feature_(ids,frame):93 ids_feat={}94 for i,bbox in ids.items():95 img = frame[bbox[1]:bbox[3], bbox[0]:bbox[2]]96 img = preprocess(img,size)97 infer_res = exec_net.start_async(request_id=0,inputs={input_layer:img})98 status=infer_res.wait()99 results = exec_net.requests[0].outputs[output_layer][0]100 ids_feat[i]=results101 return ids_feat102def distance_(feature,feat1):103 x = torch.tensor(feature).unsqueeze(0)104 y = torch.tensor(feat1).unsqueeze(0)105 d=torch.cosine_similarity(x, y)[0].numpy()106 return d107def distance_list(feature,feat1):108 x = torch.tensor(feature[0]).unsqueeze(0)109 dis=[]110 for i in feat1:111 y = torch.tensor(i).unsqueeze(0)112 d=torch.cosine_similarity(x, y)[0].numpy()113 dis.append(d)114 115 return sum(dis)/len(dis)116def ids_feature_list(ids,frame):117 ids_feat={}118 for i,bbox in ids.items():119 img = frame[bbox[1]:bbox[3], bbox[0]:bbox[2]]120 img = preprocess(img,size)121 infer_res = exec_net.start_async(request_id=0,inputs={input_layer:img})122 status=infer_res.wait()123 results = exec_net.requests[0].outputs[output_layer][0]124 ids_feat[i]=[results]...
launcher_test.py
Source:launcher_test.py
...10 @mock.patch.object(launcher.update.status, 'get')11 def test_launches_update_when_no_update_is_running(self, mock_status_get,12 mock_clear, mock_popen):13 mock_status_get.return_value = (update_status.Status.NOT_RUNNING, '')14 launcher.start_async()15 mock_clear.assert_called()16 mock_popen.assert_called_once_with(17 ('sudo', '/usr/sbin/service', 'tinypilot-updater', 'start'))18 @mock.patch.object(launcher.subprocess, 'Popen')19 @mock.patch.object(launcher.update.result_store, 'clear')20 @mock.patch.object(launcher.update.status, 'get')21 def test_launches_update_when_previous_update_succeeded(22 self, mock_status_get, mock_clear, mock_popen):23 mock_status_get.return_value = (update_status.Status.DONE, '')24 launcher.start_async()25 mock_clear.assert_called()26 mock_popen.assert_called_once_with(27 ('sudo', '/usr/sbin/service', 'tinypilot-updater', 'start'))28 @mock.patch.object(launcher.subprocess, 'Popen')29 @mock.patch.object(launcher.update.result_store, 'clear')30 @mock.patch.object(launcher.update.status, 'get')31 def test_launches_update_when_previous_update_failed(32 self, mock_status_get, mock_clear, mock_popen):33 mock_status_get.return_value = (update_status.Status.DONE,34 'dummy updater failure message')35 launcher.start_async()36 mock_clear.assert_called_once()37 mock_popen.assert_called_once_with(38 ('sudo', '/usr/sbin/service', 'tinypilot-updater', 'start'))39 @mock.patch.object(launcher.subprocess, 'Popen')40 @mock.patch.object(launcher.update.result_store, 'clear')41 @mock.patch.object(launcher.update.status, 'get')42 def test_does_not_launch_if_update_is_already_running(43 self, mock_status_get, mock_clear, mock_popen):44 mock_status_get.return_value = (update_status.Status.IN_PROGRESS, '')45 with self.assertRaises(launcher.AlreadyInProgressError):46 launcher.start_async()47 mock_clear.assert_not_called()...
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:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run lisa 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!!
Was this article helpful?
