一、問題分析
首先通過python編寫腳本來啟動Appium服務,這里需要subprocess模塊,該模塊可以創建新的進程,並且連接到進程的輸入、輸出、錯誤等管道信息並且可以獲取進程的返回值。
二、場景構造
在做多終端並發運行測試時候,Appium都要給不同終端都要分配唯一的服務:
- Appium服務器端口:4723,bp端口為4724
- Appium服務器端口:4725,bp端口為4726
備注:
1、bp端口( --bootstrap-port)是appium和設備之間通信的端口,如果不指定到時無法操作多台設備運行腳本
2、對於端口是否可用?如何處理?(后面會介紹,可以在運行的log日志或者通過shell命令查看當前端口是否可用)
三、單服務實現方式
1 import subprocess 2 from time import ctime 3 4 def appiumserver_start(host,port): 5 '''啟動appium server''' 6 bootstrap_port = str(port + 1) 7 cmd = 'start /b appium -a ' + host + ' -p ' + str(port) + ' -bp ' + str(bootstrap_port) 8 print('%s at %s' %(cmd,ctime())) 9 subprocess.Popen(cmd, shell=True,stdout=open('./appium_log/'+str(port)+'.log','a'),stderr=subprocess.STDOUT) 10 11 12 if __name__ == '__main__': 13 host = '127.0.0.1' 14 port=4723 15 appium_start(host,port)
四、檢測端口是否被占用
- 在appium服務啟動是否成功,此處需要校驗,方法如下:
- 首先查看有沒有生成對應的log文件,查看log里面的內容
- 命令查看(netstate -ano | findstr 端口號)
C:\Users\Administrator>netstat -ano |findstr "4723" TCP 127.0.0.1:4723 0.0.0.0:0 LISTENING 8224
- 服務啟動后如何關閉?方法如下:
- 通過netstat命令找到對應的Appium進程pid然后可以在系統任務管理器去關閉進程;
- 使用如下命令來關閉(taskkill -f -pid
appium進程id)
- 實現代碼
1 import socket 2 import os 3 4 def check_port(host, port): 5 """檢測指定的端口是否被占用""" 6 7 #創建socket對象 8 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 9 try: 10 s.connect((host, port)) 11 s.shutdown(2) 12 except OSError as msg: 13 print('port %s is available! ' %port) 14 print(msg) 15 return True 16 else: 17 print('port %s already be in use !' % port) 18 return False 19 20 21 def release_port(port): 22 """釋放指定的端口""" 23 24 #查找對應端口的pid 25 cmd_find='netstat -aon | findstr %s' %port 26 print(cmd_find) 27 28 #返回命令執行后的結果 29 result = os.popen(cmd_find).read() 30 print(result) 31 32 if str(port) and 'LISTENING' in result: 33 #獲取端口對應的pid進程 34 i=result.index('LISTENING') 35 start=i+len('LISTENING')+7 36 end=result.index('\n') 37 pid=result[start:end] 38 39 # 關閉被占用端口的pid 40 cmd_kill='taskkill -f -pid %s' %pid 41 print(cmd_kill) 42 os.popen(cmd_kill) 43 44 else: 45 print('port %s is available !' %port) 46 47 48 if __name__ == '__main__': 49 host='127.0.0.1' 50 port=4723 51 check_port(host,port) 52 release_port(port)
五、通過多進程並發啟動Appium多服務
- 此處需要用到multiprocessing多進程模塊
1 import multiprocessing 2 import subprocess 3 from time import ctime 4 5 def appiumserver_start(host,port): 6 '''啟動appium server''' 7 8 bootstrap_port = str(port + 1) 9 cmd = 'start /b appium -a ' + host + ' -p ' + str(port) + ' --bootstrap-port ' + str(bootstrap_port) 10 11 print('%s at %s' %(cmd,ctime())) 12 subprocess.Popen(cmd, shell=True,stdout=open('./appium_log/'+str(port)+'.log','a'),stderr=subprocess.STDOUT) 13 14 15 #構建appium進程組 16 appium_process=[] 17 18 #加載appium進程 19 for i in range(2): 20 host='127.0.0.1' 21 port = 4723 + 2 * i 22 appium=multiprocessing.Process(target=appiumserver_start,args=(host,port)) 23 appium_process.append(appium) 24 25 26 if __name__ == '__main__': 27 #並發啟動appium服務 28 for appium in appium_process: 29 appium.start() 30 for appium in appium_process: 31 appium.join()
