AppUi自动化框架appController.py代码

from lib.tool import Tool
from appium import webdriver
from lib.logger import logger
from lib.path import LOGPATH, APPPICTUREPATH
import os
import time
import queue
import subprocess
# 存放driver的对列
driver_queue = queue.Queue()


class Controller(object):
    def __init__(self):
        self.tool = Tool()
        # 配置信息
        self.yml = self.tool.app_data
        # 所有手机配置信息
        self.devices = self.yml.get(‘devices‘)
        # 测试app的信息
        self.app = self.yml.get(‘tester‘)
        # Android OR iOS
        self.device_type = self.yml.get(‘device_type‘)
        # 用于校验服务是否成功，存储端口号
        self.ports = []
        # 手机名
        self.deviceName = self.devices.get(self.device_type)[0].get(‘name‘)

    def kill_server(self):
        """
        adb如果重启  夜游神将不会被查到
        :return:
        """
        logger.debug(‘执行[KILL SERVER]操作:%s‘ % subprocess.getoutput("taskkill /F /IM node.exe /t"))
        # logger.debug(‘重启ADB服务！%s‘ % subprocess.getoutput("adb kill-server"))

    def start_server(self):
        self.kill_server()
        device = self.devices.get(self.device_type)[0]
        self.ports.append(device.get(‘port‘))
        device.update({‘log_path‘: os.path.join(LOGPATH, ‘%s.log‘ % device.get(‘name‘))})
        command = ‘appium -a {ip} -p {port} -U {deviceName} -g {log}‘.format(ip = device.get(‘ip‘),
                                                                             port = device.get(‘port‘),
                                                                             deviceName = device.get(
                                                                                 ‘deviceName‘),
                                                                             log = device.get(‘log_path‘))
        logger.debug(‘启动服务执行的命令：%s‘ % command)
        # 以shell命令方式执行命令，并输入日志到log文件，相当于在cmd窗口输入上面的命令command
        subprocess.Popen(command, stdout = open(device.get(‘log_path‘), ‘a+‘), stderr = subprocess.PIPE, shell = True)

    def start_driver(self):
        device = self.devices.get(self.device_type)[0]
        device.update(self.app)
        desired_caps = {‘platformName‘: ‘‘, ‘platformVersion‘: ‘‘, ‘deviceName‘: ‘‘,
                              "unicodeKeyboard": "True",
                              "resetKeyboard": "True", ‘udid‘: ‘‘, ‘noReset‘: ‘True‘}
        desired_caps.update(device)
        url = ‘http://{ip}:{port}/wd/hub‘.format(port = desired_caps.get(‘port‘),
                                                 ip = desired_caps.get(‘ip‘))
        logger.debug(‘启动的Url：%s‘ % url)
        driver = webdriver.Remote(url, desired_caps)
        # 通过消息对列传递driver驱动
        driver_queue.put(driver)
        # 清理上次的图片
        self.tool.app_clear()

    def test_server(self):
        while True:
            for port in self.ports:
                # 通过查看是否有返回值来确定是否启动
                test_out_put = subprocess.getoutput("netstat -ano | findstr %s" % port)
                # 如果有 则从list中删除这个端口 直到这个list为空时 代表启动成功 跳出循环
                if test_out_put:
                    logger.debug(‘检验服务启动：%s‘ % test_out_put)
                    self.ports.remove(port)
                else:
                    logger.debug(‘端口 [%s] 服务启动失败5秒钟后尝试‘ % port)
            if not self.ports:
                break
            time.sleep(5)
        logger.debug(‘全部服务启动成功！‘)
        return True
# if __name__ == ‘__main__‘:
#     c = Controller()