1. 程式人生 > >appium+Python 學習筆記(1)

appium+Python 學習筆記(1)

Appium  簡介及工作原理         開源、跨平臺、原生/混合app、 Ios/Android         工作原理講解;

Appium  環境搭建:       指令碼----Appium  Server ----UiAutomator---手機執行;              指令碼:Python         Appium  Server: Appium ----Node.js       UiAutomator: AndroidSDK----- jdk       手機執行:手機/模擬器       Appium -----Appium-Python-Client----  Python               1、jdk ;  2、AndroidSDK;  3、Node.js;  4、Appium;     5、Python ;  6、Appium-Python-Client ;       第一步:1、安裝jdk1.8.0_05,安裝路徑記住; 2、環境變數的配置; JAVA_HOME 、 Path  、 CLASSPATH  ;3、環境檢驗: Java -version;              第二步:AndroidSDK (完整目錄:add-ons、build-tools、extras、platforms、platform-tools、system-images、tools)            1、 環境變數配置:ANDROID_HOME  :         Path:  %ANDROID_HOME%\tools;                                                          %ANDROID_HOME%\platform-tools;            2、環境檢驗:輸入adb      aapt         第三步:安裝Node.js;              一路傻瓜式安裝,安裝完成後,執行cmd,輸入node –v檢視版本號,然後輸入npm       第四步:Appium       第五步:Python(2.7)       第六步:Appium-Python-Client         Appium  執行起來       保證Appium 出現方格,顯示debug 模式;指令碼才可以正常執行。    Android  工具使用介紹       adb connect 127.0.0.1:62001       adb  devices                     aapt  dump  badging  xxx.apk       sdk/build-tools/android-4.3/aapt.exe       檢視package  和 activity 的屬性:aapt d bading 路徑               Appium  啟動程式碼配置講解       #coding = utf-8       from appium import webdriver       capabilities = {          "platformName" : "Android",          "deviceName" : "127.0.0.1:4723",          "app": "E:\\PythonAppium\\AutoTestAppium\\apps\\xxx.apk"       }       driver = webdriver.Remote("http://127.0.0.1:4723/wd/hub",capabilities)         啟動日誌分析:

頁面滑動及初級使用          driver.swipe(x,y,x1,y1,time)              # 獲取螢幕的size     size = driver.get_window_size()     print(size)     # 螢幕寬度width     print(size['width'])     # 螢幕高度width     print(size['height'])

頁面簡單滑動函式封裝

    #獲取螢幕的寬高     def get_size():         size = driver.get_window_size()         width = size['width']         heigth = size['height']         return  width,heigth     #向左邊滑動       def swipe_left():         x1 = get_size()[0]/10*9         y1 = get_size()[1]/2         x  = get_size()[0]/10         driver.swipe(x1,y1,x,y1)     swipe_left()            參考網址:https://www.cnblogs.com/yoyoketang/p/7766878.html      driver 和滑動函式封裝結合

    from appium import webdriver     def get_driver():       capabilities = {          "platformName" : "Android",          "deviceName" : "127.0.0.1:21503",          "app": "E:\\PythonAppium\\AutoTestAppium\\apps\\"       }       driver = webdriver.Remote("http://127.0.0.1:4723/wd/hub",capabilities)       return driver

    #獲取螢幕的寬高     def get_size():         size = driver.get_window_size()         width = size['width']         heigth = size['height']         return  width,heigth     #向左邊滑動       def swipe_left():         x1 = get_size()[0]/10*9         y1 = get_size()[1]/2         x  = get_size()[0]/10         driver.swipe(x1,y1,x,y1)            #向右邊滑動       def swipe_right():         x1 = get_size()[0]/10         y1 = get_size()[1]/2         x  = get_size()[0]/10*9         driver.swipe(x1,y1,x,y1)                def login():         driver.find_element_by_id('').         driver.find_element_by_id('').         driver.find_element_by_id('').click()            #向上滑動       def swipe_up():         x1 = get_size()[0]/2         y1 = get_size()[1]/10*9         y  = get_size()[1]/10         driver.swipe(x1,y1,x1,y)                #向下滑動       def swipe_down():         x1 = get_size()[0]/2         y1 = get_size()[1]/10         y  = get_size()[1]/10*9         driver.swipe(x1,y1,x1,y)              def swipe_on(direction):         if direction == 'up':            swipe_up()         if direction == 'down':            swipe_down()         if direction == 'left':            swipe_left()         else:            swipe_right()                 driver = get_driver()     swipe_on('left')     login()         如何切換activity 解決真機無法啟動問題    "appWaitActivity":"cn.com.open.mooc.index.spain"

     id 定位進行登入操作

className 定位元素    重置應用: “noReset”:"true"        element = driver.find_element_by_class_name('')    print element    elements = driver.find_elements_by_class_name('')    print len(elements)

層級定位思想分析

      element = driver.find_element_by_id("cn.com.open.mooc:id/rv_child")       elements = element.find_elements_by_class_name("android.widget.RelativeLayout")       elements[1].click()       for ele in elements:          ele.click()

層級定位和list定位結合實戰

   element = driver.find_element_by_id('')    elements = element.find_element_by_class_name('')                 Element = driver.find_element_by_class_name('android.widget.EditText')     print(Element)     elements = driver.find_elements_by_class_name('android.widget.EditText')     elements[0].send_keys('jzp')     print(len(elements))     for i in elements:         i.send_keys('jzp')         time.sleep(2)         i.send_keys('123')         time.sleep(2)

        driver.find_elements_by_class_name('android.widget.EditText')[0].send_keys('jzp')     driver.find_elements_by_class_name('android.widget.EditText')[1].send_keys('123')     driver.find_element_by_class_name('android.widget.Button').click()

           可以先用find_elements定位一組物件,再通過下標索引[0]取出第一個就可以了    driver.find_elements_by_class_name("android.widget.Image")[0].click()     uiautomator定位元素  

   driver.find_element_by_android_uiautomator('new UiSelector().index("2")').send_keys('jzp')    driver.find_element_by_android_uiautomator('new UiSelector().index("4")').send_keys('123')     通過xpath進位

     driver.find_element_by_xpath('//android.widget.TextView[@text="忘記密碼"]').click()                  driver.find_element_by_xpath('//*[contains(@text,"忘記密碼")]').click()                  driver.find_element_by_xpath('//android.widget.TextView[@resource-id="cn.com.open.mooc:id/login_lable"]/../')     #返回父層節點      driver.find_element_by_xpath('//android.widget.TextView[@resource-id="cn.com.open.mooc:id/login_lable"]/../preceding-sibling::android.widget.RelativeLayout').send_keys('')   #返回父層節點/兄弟節點(兄弟節點的屬性)      

原生APP 和H5 進行相互切換程式碼實戰       見整個頁面就是一個webview,              def get_web_view():           time.sleep(10)            webview = driver.contexts           print(webview)               如何獲取tost元素                   from appium import webdriver     def get_driver():       capabilities = {          "platformName" : "Android",          “automationName”:"UiAutomator2"             #必須新增這個          "deviceName" : "127.0.0.1:21503",          "app": "E:\\PythonAppium\\AutoTestAppium\\apps\\xxx.apk"       }       driver = webdriver.Remote("http://127.0.0.1:4723/wd/hub",capabilities)       return driver                     from selenium.webdriver.support.ui import WebDriverWait       from selenium.webdriver.support  import expected_conditions as EC                            def get_tost():           tost_element = ("xpath","//*[contain(@text,'請輸入密碼')]")           WebDriverWait(driver,10,0.1).until(EC.presence_of_element_located())                                             讀取配置檔案基礎程式碼講解        在Python2下,需要大寫:import  ConfigParser        在PYthon3下,需要小寫:import   configparser

      在new ----Directory---       工程目錄下建立一個Config       建立一個 localElement.ini              [login_element]       username=email       password=pass       login_button=loginbutton                     工程目錄下建立一個util       建立一個 read_init.py              #conding=utf-8       import ConfigParser         #在命令視窗pip install ConfigParser             class ReadIni:            def _init_(self,file_path=None):                if file_path == None:                    self.file_path = '路徑 + /config/LocalElement.ini'                else:                    self.file_path = file_path                self.data = self.read_ini()                        def read_ini(self):                 read_ini = ConfigParser.ConfigParser()                 data = read_ini.read('路徑 + /config/LocalElement.ini')                 return read_ini                             #通過KEY獲取對應的value            def get_value(self,key,section=None):                if section == None:                   section = 'login_element'                try:                   value =  self.data.get(section,key)                except:                   value = None                return value                                        if __name__ == '__main__':             read_ini = ReadIni()             print read_ini.get_value("password")                                  python3.6 使用查詢的一些方法:             http://www.pythondoc.com/pythontutorial3/modules.html#tut-packages                         import sound.effects.echo              from sound.effects import echo              from sound.effects.echo import echofilter                                   #coding=utf-8       import configparser                     print read_ini.geet('login_element','useername')              class ReadIni:            def __init__(self,file_path=None):                if file_path == None:                   self.file_path = '路徑 + /config/LocalElement.ini'                else:                   self.file_path = file_path                self.data = self.read_ini()            def read_ini(self):                read_ini = configparser.ConfigParser()                read_ini.read(self.file_path)                  return read_ini                            def get_value(self,section=None,key):                return self.data.get('login_element',key)                        if __name__ == '__main__':            read_ini = ReadIni()            print read_ini.get_value("password")             封裝定位資訊:       在util 下面建立一個get_by_local.py              #coding=utf-8       from read_init import ReadIni              class GetByLocal:            def __init__(self,driver):                self.driver = driver            def get_element(self,key):                read_ini = ReadIni()                local = read_ini.get_value(key)                by = local.split('>')[0]                local_by = local.split('>')[1]                if by == 'id':                   return self.driver.find_element_by_id(local_by)                elif by == 'className'                   return self.driver.find_element_by_class_name(local_by)                else:                   return self.driver.find_element_by_xpath(local_by)                                       分層思想     建立page 層     在page層中建立一個login_page.py          #coding=utf-8     from util.get_by_local import GetByLocal      from util.read_init import ReadIni     class loginPage:            #獲取登入頁面所有的頁面元素資訊                        def __init__(self,driver):                self.driver = driver                self.get_by_local = GetByLocal(driver)                            def get_username_element(self):                ...                獲取使用者名稱元素資訊                ...                return self.get_by_local.get_element('username')                            def get_password_element(self)                ...                獲取密碼元素資訊                ...                return self.get_by_local.get_element('password')                            def get_login_button_element(self)                      ...                獲取登入按鈕元素資訊                ...                return self.get_by_local.get_element('login_button')                                      建立handle 層      在page層中建立一個login_handle.py      #coding=utf-8      from page.login_page import loginPage            class ClassName:            def __init__(self,driver):                self.login_page = LoginPage(driver)   #這裡書寫不一樣                            def send_username(self,user):                                self.login_page.get_username_element().send_keys(user)                  def send_password(self,password):                self.login_page.get_password_element().send_keys(password)                            def click_login(self):                self.login_page.get_login_button_element().click                      建立business 層      在business 層中建立一個login_business.py             #coding=utf-8      from handle.login_handle import ClassName(LoginHandle)   #這裡書寫不一樣            class LoginBusiness:           def __init__(self):               self.login_handle = LoginHandle(driver)                          def login_pass(self):               self.login_handle.send_username('xxxx')               self.login_handle.send_password('111111')               self.login_handle.click_login                           def login_user_error(self):               self.login_handle.send_username('xxxx')               self.login_handle.send_password('111111')               self.login_handle.click_login                                              unittest 的簡單使用              在case 層中建立一個test_case.py                          #coding=utf-8        import unittest        class CaseTest(unittest.TestCase) :              @classmethod              def setUpClass(cls):                  print("this is class")                                def setUp(self):                   print("this is setup")                                  def test_01(self):                  print("this is case")              def test_02(self):                  print("this is case02")                                                     def tearDown(self):                  print("this is teardown")                                @classmethod              def tearDownClass(cls):                  print("this is class")                          if __name__ == '__main__':             unittest.main()                         unittest中斷言的使用

unittest 的api

       flag = True        self.assertEqual(1,2,'資料錯誤')  #前2個值對比判斷,第3個值輸出;        self.assertNotEqual(1,2)        self.assertTrue(flag)        self.assertFalse(flag)         unittest 中case 的管理

      #coding=utf-8        import unittest        class CaseTest(unittest.TestCase) :              @classmethod              def setUpClass(cls):                  print("this is class")                                def setUp(self):                   print("this is setup")                                  def test_01(self):                  print("this is case")              def test_02(self):                  print("this is case02")                                                     def tearDown(self):                  print("this is teardown")                                @classmethod              def tearDownClass(cls):                  print("this is class")                          if __name__ == '__main__':             #unittest.main()             suite = unittest.TestSuite()             suite.addTest(CaseTest("test_02"))             suite.addTest(CaseTest("test_01"))             unittest.TextTestRunner().run(suite)                           unittest 中 HTMLTestRunner 的使用

   在報告中看結果:    1)把HTMLTestRunner.py 檔案下載下來,放到Python 的Lib 檔案中;    2)在cmd 中執行 python ,在輸入import HTMLTestRunner 檢查是否報錯。        在程式碼中如何使用:      #coding=utf-8        import unittest        import HTMLTestRunner        class CaseTest(unittest.TestCase) :              @classmethod              def setUpClass(cls):                  print("this is class")                                def setUp(self):                   print("this is setup")                                  def test_01(self):                  print("this is case")              def test_02(self):                  print("this is case02")                                                     def tearDown(self):                  print("this is teardown")                                @classmethod              def tearDownClass(cls):                  print("this is class")                          if __name__ == '__main__':             #unittest.main()             suite = unittest.TestSuite()             suite.addTest(CaseTest("test_02"))             suite.addTest(CaseTest("test_01"))             #unittest.TextTestRunner().run(suite)             html_file = ""               #路徑需要新增\\ 斜槓             fp = file(html_file,"wb")   #file 有時候顯示不出來需要用open             HTMLTestRunner.HTMLTestRunner(fp).run(suite)                       建立report 層          在report層中建立一個report.html 檔案          在html_file = "" 新增  路徑+report.html           

多執行緒的初級使用        #coding=utf-8        import threading        def sum(a)            print(a+1)            threads = []        for i in rang(3):            print i            t = threading.Thread(target=sum,args=(i,))            t.start()        for j in threads:            j.start()             多執行緒和unittest、HTMLTestRunner 結合使用                #coding=utf-8        import unittest        import HTMLTestRunner        class CaseTest(unittest.TestCase) :              @classmethod              def setUpClass(cls):                  print("this is class")                                def setUp(self):                   print("this is setup")                                  def test_01(self):                  print("this is case")              def test_02(self):                  print("this is case02")                                                     def tearDown(self):                  print("this is teardown")                                @classmethod              def tearDownClass(cls):                  print("this is class")                           def get_suite(i):             suite = unittest.TestSuite()             suite.addTest(CaseTest("test_02"))             suite.addTest(CaseTest("test_01"))             #unittest.TextTestRunner().run(suite)             html_file = "D:\\py_testt\\report\\report"+str(i)+".html"               #路徑需要新增\\ 斜槓             fp = file(html_file,"wb")   #file 有時候顯示不出來需要用open             HTMLTestRunner.HTMLTestRunner(fp).run(suite)                          if __name__ == '__main__':             threads = []             for i in range(3):                 print i                 t = threading.Thread(target=get_suite,args=(i,))                 t.start()             for j in threads:                 j.start()                                   appium 服務介紹      解除安裝Appium的客戶端,使用命令列操作;      appium 翻牆軟體      https://blog.csdn.net/Candy_mi/article/details/80764319          如果使用輸入cnpm  顯示不是內部命令時,需要進入對應的檔案目錄執行。      npm install -g cnpm --registry=http://registry.taobao.org      cnpm      cnpm install -g appium             在命令列輸入 appium  自動啟動   按終止程式Ctrl c   shutting down      啟動成功會出現版本號:            指定埠      appium -p 4725   啟動後出現埠為4725      多個裝置      appium -p 4700 -bp 4701 -U 127.0.0.1:21503       命令列啟動和指令碼結合

     檢查通訊是否成功;            把配置連線放到 setUpClass(cls) 中;            保證環境可以正常執行起來;      

Page 層封裝driver       1、需要把login_business 引入到 test_case 中          from business.login_business import LoginBusiness                     cls.login_business = LoginBusiness(driver)                    self.login_business.login_pass                 2、在base 層中 新建一個 base_driver.py       #coding=utf-8       import time       from appium import Webdriver       class BaseDriver:                    def android_driver(self):                 capabilities = {                    "platformName" : "Android",                    "automationName":"UiAutomator2"             #必須新增這個                    "deviceName" : "127.0.0.1:21503",                    "app": "E:\\PythonAppium\\AutoTestAppium\\apps\\mukewang.apk"                    "appWaitActivity":"cn.com.open.mooc.index.splash.MCSplashActivity"                    "noReset":"true"                 }                 driver = webdriver.Remote("http://127.0.0.1:4723/wd/hub",capabilities)                 time.sleep(10)                 return driver      3、修改login_page 中的driver          from base.base_driver import BaseDriver                                                            def __init__(self,driver):                base_driver = BaseDriver()                self.driver = base_driver.android_driver()                            business 和 handle 、login_page 去掉driver;                                     服務端設計思路:

封裝執行命令方法:

    在util 目錄下建立一個dos_cmd.py 檔案          #coding=utf-8     import os     class DosCmd:           def excute_cmd_result(self,command):               result_list = []               result = os.popen(command).readlines()               for i in result:                   if i=='\n':                       continue                   result_list.append(i.strip('\n'))               return result_list           def excute_cmd(self,command):               os.system(command)                        if __name__ == '__main__':         dos = DosCmd()         print dos.excute_cmd_result('adb devices')           重構封裝獲取裝置資訊類       在util 目錄下建立一個server.py 檔案          #coding=utf-8     from util.dos_cmd import DosCmd     class Server:                def get_devices(self):               self.dos = DosCmd()               devices_list = []               result_list = self.dos.excute_cmd_result('adb devices')               if len(result_list)>=2:                  for i in result_list:                      if 'List' in i:                         continue                      devices_info = i.split('\t')                      if devices_info[1] == 'device':                          devices_list.append(devices_info[0])                  return devices_list               else:                  return None                        if __name__ == '__main__':          server = Server()          print server.get_devices()              檢查埠是否被佔用       在util 目錄下建立一個port.py 檔案              #coding=utf-8       from util.dos_cmd import DosCmd       class Port:             def port_is_used(self,port_num):                              flag = None                 self.dos = DosCmd()                 command = 'netstat -ano | findstr ' + port_num                 result = self.dos.excute_cmd_result(command)                 if len(result) > 0:                    flag = True                 else:                    flag = False                 return flag                                         if  __name__ == '__main__':             port = Port()             print (port.port_is_used('1234'))                                     封裝生成可用埠方法:

    #coding=utf-8       from util.dos_cmd import DosCmd       class Port:             def port_is_used(self,port_num):                              flag = None                 self.dos = DosCmd()                 command = 'netstat -ano | findstr ' + str(port_num)                  result = self.dos.excute_cmd_result(command)                 if len(result) > 0:                    flag = True                 else:                    flag = False                 return flag                               def create_port_list(self,start_port,device_list):                                  '''start_port                 生成可用埠                                  @parameter  start_port                 @parameter  device_list                 '''                                  port_list = []                 if device_list != None:                    while len(port_list) != len(device_list)  :                           if self.port_is_user(start_port) != True:                              port_list.append(start_port)                                start_port = start_port +1                    return port_list                 else:                    print ("生成可用埠失敗")                      return None                             if  __name__ == '__main__':             port = Port()            # print (port.port_is_used('1234'))                        li = [1,2,3]            print (port.create_port_list(4725,li))          封裝生成啟動命令列函式      使用命令生成2個埠      appium -p 4723 -bp 4724 -U 127.0.0.1:62001

     在server.py 檔案中設定:             

    #coding=utf-8     from util.dos_cmd import DosCmd     from util.port import Port     class Server:                def get_devices(self):               self.dos = DosCmd()               devices_list = []               result_list = self.dos.excute_cmd_result('adb devices')               if len(result_list)>=2:                  for i in result_list:                      if 'List' in i:                         continue                      devices_info = i.split('\t')                      if devices_info[1] == 'device':                          devices_list.append(devices_info[0])                  return devices_list               else:                  return None                              def create_port_list(self,start_port):                '''                建立可用埠                '''                port = Port()                port_list = []                port_list = port.create_port_list(start_port,self.get_devices())                return port_list                            def create_command_list(self):                #appium -p 4700 -bp 4701 -U 127.0.0.1:21503                command_list = []                appium_port_list = self.creare_port_list(4700)                bootstrap_port_list = self.create_port_list(4900)                device_list = self.get_devices()                for i in range(len(device_list)):                    command = "appium -p " + str(appium_port_list[i]) + " -bp " + str( bootstrap_port_list[i])+ " -U " +device_list[i]+ " --no-reset --session-override"                    command_list.append(command)                return command_list                                        if __name__ == '__main__':          server = Server()          print server.create_command_list()           封裝多執行緒啟動appium 服務              #coding=utf-8     from util.dos_cmd import DosCmd     from util.port import Port     import threading      class Server:                def get_devices(self):               self.dos = DosCmd()               devices_list = []               result_list = self.dos.excute_cmd_result('adb devices')               if len(result_list)>=2:                  for i in result_list:                      if 'List' in i:                         continue                      devices_info = i.split('\t')                      if devices_info[1] == 'device':                          devices_list.append(devices_info[0])                  return devices_list               else:                  return None                              def create_port_list(self,start_port):                '''                建立可用埠                '''                port = Port()                port_list = []                port_list = port.create_port_list(start_port,self.get_devices())                return port_list                            def create_command_list(self):                #appium -p 4700 -bp 4701 -U 127.0.0.1:21503                command_list = []                appium_port_list = self.creare_port_list(4700)                bootstrap_port_list = self.create_port_list(4900)                device_list = self.get_devices()                for i in range(len(device_list)):                    command = "appium -p " + str(appium_port_list[i]) + " -bp " + str( bootstrap_port_list[i])+ " -U " +device_list[i]+ " --no-reset --session-override"                    command_list.append(command)                return command_list                            def start_server(self,i):                self.start_list = self.create_command_list()                self.dos.excute_cmd(self.start_list[i])                            def main(self):                for i in range(len(self.create_command_list())):                    appium_start = threading.Thread(target=self.start_server,args=(i,))                    appium_start.start()                        if __name__ == '__main__':          server = Server()          print server.main()                      在工作管理員中檢視Node.exe 是否啟動多個(遇實際一致不)                       清理 appium 環境  

    #coding=utf-8     from util.dos_cmd import DosCmd     from util.port import Port     import threading      class Server:                def __init__(self):               self.dos = DosCmd()                          def get_devices(self):                              devices_list = []               result_list = self.dos.excute_cmd_result('adb devices')               if len(result_list)>=2:                  for i in result_list:                      if 'List' in i:                         continue                      devices_info = i.split('\t')                      if devices_info[1] == 'device':                          devices_list.append(devices_info[0])                  return devices_list               else:                  return None                              def create_port_list(self,start_port):                '''                建立可用埠                '''                port = Port()                port_list = []                port_list = port.create_port_list(start_port,self.get_devices())                return port_list                            def create_command_list(self):                #appium -p 4700 -bp 4701 -U 127.0.0.1:21503                command_list = []                appium_port_list = self.creare_port_list(4700)                bootstrap_port_list = self.create_port_list(4900)                device_list = self.get_devices()                for i in range(len(device_list)):                    command = "appium -p " + str(appium_port_list[i]) + " -bp " + str( bootstrap_port_list[i])+ " -U " +device_list[i]+ " --no-reset --session-override"                    command_list.append(command)                return command_list                            def start_server(self,i):                self.start_list = self.create_command_list()                self.dos.excute_cmd(self.start_list[i])                            def kill_server(self):                server_list = self.dos.excute_cmd_result('tasklist | find "node.exe"')                if len(server_list)>0:                   self.dos.excute_cmd('taskkill -F -PID node.exe')                            def main(self):                self.kill_server()                for i in range(len(self.create_command_list())):                    appium_start = threading.Thread(target=self.start_server,args=(i,))                    appium_start.start()                        if __name__ == '__main__':          server = Server()          print server.main()                       通過yaml檔案獲取命令列資料     config 層中建立一個userconfig.yaml 檔案;          user_info_0: {bp: '4900',deviceName: '127.0.0.1:62001',port: '4723'}     user_info_1: {bp: '4901',deviceName: '127.0.0.1:62002',port: '4724'}     user_info_2: {bp: '4902',deviceName: '127.0.0.1:62003',port: '4725'}     util 層中建立一個write_user_command.py 檔案          #coding=utf-8     import yaml                                #如果報錯,需要下載安裝pip install pyyaml     class WriteUserCommand:           def read_data(self):               with open("../config/userconfig.yaml") as fr:            #路徑要加雙斜槓python3.6中                           data = yaml.load(fr)               return data                          def get_value(self,key,port):               data = self.read_data()               value = data[key][port]               return value                    if __name__ == '__main__':          write_file = WriteUserCommand()          print(write_file.get_value('user_info_2','bp'))

多執行緒啟動appium 和寫入命令到檔案     在write_user_command.py     #coding=utf-8     import yaml                                #如果報錯,需要下載安裝pip install pyyaml     class WriteUserCommand:           def read_data(self):               with open("../config/userconfig.yaml") as fr:            #路徑要加雙斜槓python3.6中                           data = yaml.load(fr)               return data                          def get_value(self,key,port):               data = self.read_data()               value = data[key][port]               return value                          def write_data(self,i,device,bp,port):               data = self.join_data(i,device,bp,port)               with open("../config/userconfig.yaml","a") as fr:                    yaml.dump(data,fr)                               def join_data(self,i,device,bp,port):               data = {                "user_info_"+str(i):{                 "deviceName":device,                 "bp":bp,                 "port":port                }               }               return data                                                        def clear_data(self):               with open("../config/userconfig.yaml","w") as fr:                    fr.truncate()               fr.close()                    if __name__ == '__main__':          write_file = WriteUserCommand()          print(write_file.get_value('user_info_2','bp'))             

     在server.py                 #coding=utf-8     from util.dos_cmd import DosCmd     from util.port import Port     import threading      from util.write_user_command import WriteUserCommand               。。。。。。。。。。。。     class Server:                def __init__(self):               self.dos = DosCmd()                          def get_devices(self):                              devices_list = []               result_list = self.dos.excute_cmd_result('adb devices')               if len(result_list)>=2:                  for i in result_list:                      if 'List' in i:                         continue                      devices_info = i.split('\t')                      if devices_info[1] == 'device':                          devices_list.append(devices_info[0])                  return devices_list               else:                  return None                              def create_port_list(self,start_port):                '''                建立可用埠                '''                port = Port()                port_list = []                port_list = port.create_port_list(start_port,self.get_devices())                return port_list                            def create_command_list(self):                #appium -p 4700 -bp 4701 -U 127.0.0.1:21503                write_file = WriteUserCommand()              。。。。。。。。。。。。。。。。。。                command_list = []                appium_port_list = self.creare_port_list(4700)                bootstrap_port_list = self.create_port_list(4900)                device_list = self.get_devices()                for i in range(len(device_list)):                    command = "appium -p " + str(appium_port_list[i]) + " -bp " + str( bootstrap_port_list[i])+ " -U " +device_list[i]+ " --no-reset --session-override"                    command_list.append(command)                    write_file.write_data(i,device_list[i],str( bootstrap_port_list[i]),str(appium_port_list[i]),)                 。。。。。。。。。。。。。。。                                    return command_list                            def start_server(self,i):                self.start_list = self.create_command_list()                self.dos.excute_cmd(self.start_list[i])                            def kill_server(self):                server_list = self.dos.excute_cmd_result('tasklist | find "node.exe"')                if len(server_list)>0:                   self.dos.excute_cmd('taskkill -F -PID node.exe')                            def main(self):                self.kill_server()                self.write_file.clear_data()                for i in range(len(self.create_command_list())):                    appium_start = threading.Thread(target=self.start_server,args=(i,))                    appium_start.start()                        if __name__ == '__main__':          server = Server()          print server.main()                                                             在server.py                 #coding=utf-8     from util.dos_cmd import DosCmd     from util.port import Port     import threading      from util.write_user_command import WriteUserCommand               。。。。。。。。。。。。     class Server:                def __init__(self):               self.dos = DosCmd()               self.deivce_list = self.geet_devices()               self.write_file = WriteUserCommand()              。。。。。。。。。。。。。。。。。。                          def get_devices(self):                              devices_list = []               result_list = self.dos.excute_cmd_result('adb devices')               if len(result_list)>=2:                  for i in result_list:                      if 'List' in i:                         continue                      devices_info = i.split('\t')                      if devices_info[1] == 'device':                          devices_list.append(devices_info[0])                  return devices_list               else:                  return None                              def create_port_list(self,start_port):                '''                建立可用埠                '''                port = Port()                port_list = []                port_list = port.create_port_list(start_port,self.deivce_list)                return port_list                            def create_command_list(self,i):                #appium -p 4700 -bp 4701 -U 127.0.0.1:21503                                command_list = []                appium_port_list = self.creare_port_list(4700)                bootstrap_port_list = self.create_port_list(4900)                device_list = self.deivce_list                               command = "appium -p " + str(appium_port_list[i]) + " -bp " + str( bootstrap_port_list[i])+ " -U " +device_list[i]+ " --no-reset --session-override"                command_list.append(command)                write_file.write_data(i,device_list[i],str( bootstrap_port_list[i]),str(appium_port_list[i]),)                 。。。。。。。。。。。。。。。                                    return command_list                            def start_server(self,i):                self.start_list = self.create_command_list(i)                self.dos.excute_cmd(self.start_list[0])                            def kill_server(self):                server_list = self.dos.excute_cmd_result('tasklist | find "node.exe"')                if len(server_list)>0:                   self.dos.excute_cmd('taskkill -F -PID node.exe')                            def main(self):                self.kill_server()                self.write_file.clear_data()                for i in range(len(self.deivce_list):                    appium_start = threading.Thread(target=self.start_server,args=(i,))                    appium_start.start()                        if __name__ == '__main__':          server = Server()          print server.main()                                 多執行緒和Unittes 和po 結合問題分析

      在base_driver 中匯入util.write_user_command import WriteUserCommand                       #coding=utf-8       import time       from appium import Webdriver       from util.write_user_command import WriteUserCommand       class BaseDriver:                    def android_driver(self,i):                              write_file = WriteUserCommand()                 devices = write_file.get_value('user_info_'+str(i),'deviceName')                 port = write_file.get_value('user_info_'+str(i),'port')                 capabilities = {                    "platformName" : "Android",                    "automationName":"UiAutomator2"             #必須新增這個                    "deviceName" : devices,                    "app": "E:\\PythonAppium\\AutoTestAppium\\apps\\mukewang.apk"                    "appWaitActivity":"cn.com.open.mooc.index.splash.MCSplashActivity"                    "noReset":"true"                 }                 driver = webdriver.Remote("http://127.0.0.1:"+ port +"/wd/hub",capabilities)                 time.sleep(10)                 return driver                             在test_case 中如何使用                     在login_page 、login_handle  、login_business中需要傳入i  ;                   多執行緒、unittest、啟動服務邏輯串聯:

       #coding=utf-8        import unittest        import HTMLTestRunner        import threading        from util.server import Server        import time        from appium import webdriver        from business.login_business import loginBusiness                class ParameTestCase(unittest.TestCase)              def __init__(self,methodName='runTest',parame=None):                  super(ParameTestCase,self).__init__(methodName)                  global parames                  parames = parame                                  class CaseTest(ParameTestCase) :              @classmethod              def setUpClass(cls):                  print ("setUpclass------->",parames)                  #cls.login_business = LoginBusiness(i)                                def setUp(self):                   print("this is setup")                                  def test_01(self):                  print("test case 裡面的引數", parames)                  #print("this is case")              def test_02(self):                  print("this is case02")                                                     def tearDown(self):                  print("this is teardown")                                @classmethod              def tearDownClass(cls):                  time.sleep(1)                  print("this is class teardown\n")                           def appium_init():             server = Server()             server.main()                           def get_suite(i):             print ("get_suite裡面的",i)             suite = unittest.TestSuite()             suite.addTest(CaseTest("test_02",parame=i))             #suite.addTest(CaseTest("test_01"))             unittest.TextTestRunner().run(suite)             #html_file = "D:\\py_testt\\report\\report"+str(i)+".html"               #路徑需要新增\\ 斜槓             #fp = file(html_file,"wb")   #file 有時候顯示不出來需要用open             #HTMLTestRunner.HTMLTestRunner(fp).run(suite)                          if __name__ == '__main__':             appium_init()             threads = []             for i in range(3):                 print i                 t = threading.Thread(target=get_suite,args=(i,))                 t.start()             for j in threads:                 j.start()                            多執行緒、Unittest、啟動服務、流程梳理程式碼重構

           #coding=utf-8        import unittest        import HTMLTestRunner        import threading        from util.server import Server        import time        from appium import webdriver        from business.login_business import loginBusiness                class ParameTestCase(unittest.TestCase)              def __init__(self,methodName='runTest',parame=None):                  super(ParameTestCase,self).__init__(methodName)                  global parames                  parames = parame                                  class CaseTest(ParameTestCase) :              @classmethod              def setUpClass(cls):                  print ("setUpclass------->",parames)                  #cls.login_business = LoginBusiness(parames)                                def setUp(self):                   print("this is setup")                                  def test_01(self):                  print("test case 裡面的引數", parames)                  #print("this is case")              def test_02(self):                  print("this is case02")                                                     def tearDown(self):                  print("this is teardown")                                @classmethod              def tearDownClass(cls):                  time.sleep(1)                  print("this is class teardown\n")                           def appium_init():             server = Server()             server.main()                           def get_suite(i):             print ("get_suite裡面的",i)             suite = unittest.TestSuite()             suite.addTest(CaseTest("test_02",parame=i))             #suite.addTest(CaseTest("test_01",parame=i))             unittest.TextTestRunner().run(suite)             #html_file = "D:\\py_testt\\report\\report"+str(i)+".html"               #路徑需要新增\\ 斜槓             #fp = file(html_file,"wb")   #file 有時候顯示不出來需要用open             #HTMLTestRunner.HTMLTestRunner(fp).run(suite)                          if __name__ == '__main__':             appium_init()             threads = []             for i in range(3):                 print i                 t = threading.Thread(target=get_suite,args=(i,))                 t.start()             for j in threads:                 j.start()                     

第五章  複習

   appium 命令列啟動命令: appium -p 4723 -bp 4724 -u 127.0.0.1:21503    獲取裝置資訊、埠號;

    1、裝置資訊獲取方法,完成後最終結果:['127.0.0.1:62001','127.0.0.1:62025']    在util 下面建立一個dos_cmd.py 檔案(如何封裝去獲取裝置資訊)  #coding=utf-8  import os  class DosCmd:        def excute_cmd_result(self,command):                                 #第一種收集命令顯示裝置資訊excute_cmd_result;            result_list = []            result = os.popen(command).readlines()            for i in in result:                if i == '\n':                   continue                result_list.append(i.strip('\n'))            return result_list        

       def excute_cmd(self,command):                                        #第二種不收集命令顯示裝置資訊excute_cmd;            os.system(command)                                               使用command 原因是執行命令可能是其他命令(command 代替adb devices)    if __name__ == '__main__':       dos = DosCmd()       print(dos.excute_cmd_result('adb devices'))                                                             #目前拿到的裝置資訊如下:                                                     ['List of devices attached', '127.0.0.1:62001\tdevice', '127.0.0.1:62025\tdevice']

    在util 下面建立一個sever.py 檔案 (裝置資訊進行處理需要拿到127.0.0.1:62001)

 #coding=utf-8  form dos_cmd import DosCmd  class Server:        def get_devices(self):            '''            獲取裝置資訊            '''            self.dos = DosCmd()            devices_list = []            result_list = self.dos.excute_cmd_result('adb devices')            if len(result_list) >= 2 :     #如果裝置大於 等於2進行下面判斷                  for i in result_list:                    if 'List' in i:        #如果是List跳出下一個迴圈                        continue                                       devices_info = i.split('\t')          #下一個迴圈用'\t'拆分,如果等於device有效,如果等於offile無效;這裡拆分完之後又是一個list[0]list[1]                    if devices_info[1] == 'device'        #如果list[1]==device,就把list[0]放到devices_list = []中                       devices_list.append(devices_info[0])

               return devices_list            else:                return None

 if __name__ == '__main__':       server = Server()       print(server.get_devices())                                                      #目前拿到的裝置資訊如下: ['127.0.0.1:62001','127.0.0.1:62025']                                                       拿到裝置資訊還缺少埠-p  -bp 

2、根據裝置資訊獲取埠方法,完成後最終結果:[4725, 4726, 4727]    cmd命令視窗檢查埠是否被佔用的命令  netstat -ano | findstr 4723 如果無內容出現說明埠沒有被佔用;   #coding=utf-8   form dos_cmd import DosCmd   class Port:         def port_is_used(self,port_num):

            '''             判斷埠是否被佔用

            '''             flag = None             self.dos = DosCmd()             command = 'netstat -ano | findstr ' + str(port_num)             result = self.dos.excute_cmd_result(command)      

if len(result)>0:                flag = True                     #如果字元大於0,返回為True,端口占用;             else:                flag = Fasle             return flag                                   def create_port_list(self,start_port,device_list):            '''            生成可用埠            @parameter  start_port            @parameter  device_list            '''            port_list = []                                       #存放生成完可用埠            if devices_list != None:                             #判斷如果裝置資訊不等於None,執行下列方法;如果裝置資訊等於None,輸出“生成可用埠失敗”               while len(port_list) != len(device_list):         #如果埠長度不等於裝置資訊長度,執行下列方法;                           if self.port_is_used(start_port) != True:   #如果你這個端口占用的時候,增加一個埠                        port_list.append(start_port)                     start_port = start_port + 1               return port_list            else:               print ("生成可用埠失敗")               return None    if __name__ == '__main__':      port = Port()      li = [1,2,3]      print(port.create_port_list(4723,li))                                                    #目前拿到的埠資訊如下: [4725, 4726, 4727]                                                                                                                                                                                                                 3、使用多執行緒啟動appium命令,啟動之前清理環境:         appium 命令列啟動命令: appium -p 4723 -bp 4724 -u 127.0.0.1:21503      分析:目前已生成埠和裝置資訊,然後需要生成啟動命令。      在server檔案中獲取裝置資訊、建立可用埠、拼接appium命令、多執行緒啟動appium服務、                       #coding=utf-8  form dos_cmd import DosCmd  from port import Port  import threading  class Server:        def __init__(self):            self.dos = DosCmd()                    def get_devices(self):            '''            獲取裝置資訊            '''                        devices_list = []            result_list = self.dos.excute_cmd_result('adb devices')            if len(result_list) >= 2 :     #如果裝置大於 等於2進行下面判斷                  for i in result_list:                    if 'List' in i:        #如果是List跳出下一個迴圈                        continue                                       devices_info = i.split('\t')          #下一個迴圈用'\t'拆分,如果等於device有效,如果等於offile無效;這裡拆分完之後又是一個list[0]list[1]                    if devices_info[1] == 'device'        #如果list[1]==device,就把list[0]放到devices_list = []中                       devices_list.append(devices_info[0])

               return devices_list            else:                return None                         def create_port_list(self,start_port):             '''             建立可用埠             '''             port = Port()             port_list = []             port_list = port.create_port_list(start_port,self.get_devices())             return port_list                                   def create_command_list(self):             '''             拼接命令             '''              command_list = []             appium_port_list = self.create_port_list(4700)             bootstrap_port_list = self.create_port_list(4900)                                 appium -p 4723 -bp 4724 -u 127.0.0.1:21503             device_list = self.get_devices()             for i in range(len(device_list)):                 command = "appium -p "  + str(appium_port_list[i])  + " -bp " + str(bootstrap_port_list[i]) + " -U " + device_list[i] + " --no-reset --session-override"                   command_list.append(command)               return command_list                                      def start_server(self,i):             self.start_list = self.create_command_list()             self.dos.excute_cmd(self.start_list[i])                                                def kill_server(self):             server_list = self.dos.excute_cmd_result('tasklist | find "node.exe"')     #查詢是否有程序,如果有程序,長度大於0;             if len(server_list)>0:                                                     #如果長度                self.dos.excute_cmd('taskkill -F -PID node.exe')                                   def main(self):             self.kill_server()             for i in range(len(self.create_command_list())):                 appium_start = threading.Thread(target= self.start_server,args=(i,))                 appium_start.start()               

 if __name__ == '__main__':       server = Server()       print(server.main())                                                       在config資料夾下建立一個userconfig.yaml存放資料的檔案     er_info_0: {bp: '4902',deviceName: '127.0.0.1:21503',port: '4723'}                        在util中建立一個write_user_command.py檔案(需要從yaml 檔案中拿去資料)   #coding=utf-8  import yaml  class WriteUserCommand:        def read_data(self):            '''            載入yaml資料            '''            with open("../config/userconfig.yaml") as fr: