/usr/local/bin/appium -a ip -p port

import subprocess
import abc
import socket
class Driver：
	__metaclass__ = abc.ABCMeta
	self._host = '127.0.0.1'
	@abc.abstractmethod
	def connect_appium(self, port, n)
		"""
		待實(shí)現(xiàn)的連接設(shè)備方法
		"""
		return
	def start_appium(self, port):
		server = self.get_local_server_path()
        host = readConfig.ReadConfig().get_commend("host")
        log_path = root_path + '/result/log'
        cmd = "%s -a %s -p %s" % (server, host, str(port))
        if self.check_port(int(port)):
            subprocess.Popen(cmd, shell=True, stdout=open('%s/AppiumServer%s.log' % (log_path, port), 'w'))
            log.logger.info('%s/AppiumServer%s.log' % (log_path, port))
        else:
            log.logger.info("關(guān)閉被占用的端口號(hào)：%s" % str(port))
            self.kill_appium()
            log.logger.info("端口釋放完畢！啟動(dòng)Appium-server，端口號(hào)：%s" % str(port))
            subprocess.Popen(cmd, shell=True, stdout=open('%s/AppiumServer%s.log' % (log_path, port), 'w'))
            log.logger.info("Appium日志信息存儲(chǔ)地址： %s/AppiumServer%s.log" % (log_path, port))
    def check_port(self, port):
        """
        檢查端口占用情況
        :param port:
        :return:
        """
        try:
            host = local_read_config.get_commend("host")
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            log.logger.info(s.connect((host, port)))
            s.shutdown(2)
        except OSError:
            log.logger.info("端口：%s 可用" % str(port))
            return True
        else:
            log.logger.info("端口：%s 已被占用" % str(port))
            return False

# Android environment
from appium import webdriver
desired_caps = dict(
    platformName='Android',
    platformVersion='10',
    automationName='uiautomator2',
    deviceName='Android Emulator',
    app=PATH('../../../apps/selendroid-test-app.apk')
)
self.driver = webdriver.Remote('http://localhost:4723/wd/hub', desired_caps)
el = self.driver.find_element_by_accessibility_id('item')
el.click()

# iOS environment
from appium import webdriver
desired_caps = dict(
    platformName='iOS',
    platformVersion='13.4',
    automationName='xcuitest',
    deviceName='iPhone Simulator',
    app=PATH('../../apps/UICatalog.app.zip')
)
self.driver = webdriver.Remote('http://localhost:4723/wd/hub', desired_caps)
el = self.driver.find_element_by_accessibility_id('item')
el.click()

driver = webdriver.Remote('http://localhost:4723/wd/hub', desired_caps)

from Driver import Driver
class AndroidDriver(Driver):
    def __init__(self):
        self.driver = None

    def get_desired_caps(self):
        """
        實(shí)現(xiàn)繼承的抽象類方法；獲取鏈接設(shè)備的配置信息
        返回設(shè)備配置信息
        :return:desired_caps
        """
        desired_list = []
        package = local_read_config.get_value("ANDROID", "package")
        activity = local_read_config.get_value("ANDROID", "activity")
        devices_info = self.update_devices_info()
        for i in range(len(devices_info)):
            udid = devices_info[i].get("udid")
            device_name = devices_info[i].get("devices_name")
            platform_version = devices_info[i].get("version")
            system_port1 = 8200 + 2 * i
            desired_caps = {
                "platformName": "Android",
                "platformVersion": platform_version,
                "appPackage": package,
                "appActivity": activity,
                "deviceName": device_name,
                "automationName": "uiautomator2",
                "udid": udid,
                "systemPort": system_port1,
                "newCommandTimeout": 3000,
                # "adbExecTimeout": 50000
            }
            desired_list.append(desired_caps)
        return desired_list
    def connect_appium(self, port, n):
        """
        根據(jù)傳入的port，啟動(dòng)appium服務(wù)
        :param port:
        :param n:
        :return:
        """
        set_adb_path()
        desired_caps = self.get_desired_caps()
        try:
            self.driver = webdriver.Remote("%s:%s/wd/hub" % (super()._remote_url, str(port)), desired_caps[n])
            return self.driver
        except WebDriverException:
            raise WebDriverException
        except ConnectionError:
            raise ConnectionError

    def get_element(self, element_id):
        """
        獲取指定頁(yè)面的元素路徑數(shù)據(jù)
        :param element_id: 元素ID
        :return: 獲取的元素對(duì)象
        """
        element_type = self.page.get(element_id).get("pathType")
        element_value = self.page.get(element_id).get("pathValue")
        element = None
        if element_type == "ID":
            element = self.driver.find_element_by_id(element_value)
        elif element_type == "CLASSNAME":
            element = self.driver.find_element_by_class_name(element_value)
        elif element_type == "XPATH":
            element = self.driver.find_element_by_xpath(element_value)
        elif element_type == "NAME":
            element = self.driver.find_element_by_name(element_value)
        elif element_type == "ACB_ID":
            element = self.driver.find_element_by_accessibility_id(element_value)
        elif element_type == "PREDICATE":
            element = self.driver.find_element_by_ios_predicate(element_value)
        return element

    def swipe_to_up(self):
        """
        向上劃，頁(yè)面滾動(dòng)到最下方
        :return:
        """
        width = self.driver.get_window_size()["width"]
        height = self.driver.get_window_size()["height"]
        self.driver.swipe(width / 2, height * 3 / 5, width / 2, height / 5, duration=500)

    def swipe_to_up(self):
        """
        向上滑動(dòng)
        :return:
        """
        self.driver.execute_script('mobile: swipe', {'direction': 'up'})

from multiprocessing import Process
import pytest
import time
import os, re
import subprocess
from appiums.common import read_files
from appiums.driver.iOSDriver import IOSDriver
from driver.androidDriver import AndroidDriver
from driver import Driver
from elements import Element

class Run(Process):
    def __init__(self, name, args):
        super(Run, self).__init__()
        self.name = name
        self.args = args
        self.root_path = os.getcwd()
        self.device_name = re.sub('[\']', '', str(args[2].get("deviceName")).replace(" ", "_"))
    def run_test(self):
        """
        執(zhí)行測(cè)試用例
        :return:
        """
        pytest.main([
                     '--alluredir', '%s/result/data/%s' % (self.root_path, self.device_name)])
        time.sleep(2)
    def generate_report(self):
        """
        整合測(cè)試報(bào)告到項(xiàng)目根目錄下的result/report目錄下
        :return: none
        """
        cmd = "allure generate %s/result/data/%s -o %s/result/report/%s --clean" \
              % (self.root_path, self.device_name, self.root_path, self.device_name)
        stdout = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, text=True)
        log.logger.info("測(cè)試報(bào)告查看路徑：%s" % str(stdout.stdout.readlines()[0]).split(" ")[-1][:-1])
    def get_environment_info(self):
        """
        獲取測(cè)試環(huán)境的信息
        :return:
        """
        env = {
            "測(cè)試平臺(tái)": self.args[2].get("platformName"),
            "設(shè)備名稱": self.device_name,
            "設(shè)備系統(tǒng)版本": self.args[2].get("platformVersion"),
            "設(shè)備udid": self.args[2].get("udid"),
            "應(yīng)用名稱": self.args[2].get("bundleId") if str(self.args[2].get("platformName")).lower() == 'ios' else self.args[2].get("appPackage"),
        }
        return env
    def run(self):
        """
        執(zhí)行線程中的任務(wù)
        :return:
        """
        Driver.Driver().start_appium(self.args[0])
        time.sleep(5)
        self.set_driver()
        time.sleep(1)
        self.run_test()
        time.sleep(1)
        read_files.set_environment(self.device_name, self.get_environment_info())
        time.sleep(1)
        self.generate_report()

def main(desired_caps):
    """
    開啟測(cè)試進(jìn)程執(zhí)行測(cè)試
    """
    list_p = []
    process_num = len(desired_caps)
    if process_num > 0:
        for a in range(process_num):
            port1 = 4723 + 2 * a
            p = Run('測(cè)試進(jìn)程-%s' % str(port1), args=(port1, a, desired_caps[a]))
            p.start()
            log.logger.info("設(shè)備%s在進(jìn)程 %s 上進(jìn)行測(cè)試, 進(jìn)程ID：%s" % (desired_caps[a].get("deviceName"), p.name, p.pid))
            list_p.append(p)
        for b in list_p:
            b.join()
        Driver.Driver().kill_appium()
    else:
        log.logger.error("沒有設(shè)備可進(jìn)行測(cè)試，請(qǐng)重新連接設(shè)備后嘗試！")
        exit(-1)

def android_run():
    caps = AndroidDriver().get_desired_caps()
    main(caps)

def ios_run():
    caps = IOSDriver().get_desired_caps()
    main(caps)