测试人社区

多线程启动app返回对应driver,如何获取?

多线程启动app,启动成功后,想要根据对应uid拿到对应driver表达式,再根据不同对driver调用不同对方法,目前,driver拿到的是str类型,因为在Manager(),dict()方法内无法存储对象类型,则用repr强转为str类型存储,但是调用dill.loads(driver_emulator.encode())加载时报错:
obj = StockUnpickler.load(self)
_pickle.UnpicklingError: invalid load key, ‘<’.

class MultiAppium:

def __init__(self):
    self.host = '127.0.0.1'


def __check_port_is_used(self,port):
    p = platform.system()
    if p == 'Windows':
        sys_command = "netstat -ano|findstr %s" % port
        pipe = subprocess.Popen(sys_command, stdout=subprocess.PIPE, shell=True)
        out, error = pipe.communicate()
        if str(out, encoding='utf-8') != "" and "LISTENING" in str(out, encoding='utf-8'):
            pid = re.search(r"\s+LISTENING\s+(\d+)\r\n", str(out, encoding='utf-8')).groups()[0]
            return True, pid
        else:
            return False, None
    elif p == 'Darwin' or p == 'Linux':
        sys_command = "lsof -i:%s" % port
        pipe = subprocess.Popen(sys_command, stdout=subprocess.PIPE, shell=True)
        for line in pipe.stdout.readlines():
            if "LISTEN" in str(line, encoding='utf-8'):
                pid = str(line, encoding='utf-8').split()[1]
                return True, pid
        return False, None
    else:
        print('The platform is {} ,this platform is not support.'.format(p))


def __get_device_port(self):

    for i in range(10):
        port = random.randint(3456, 9999)
        print("suijishu:",port)
        result, pid = self.__check_port_is_used(port)
        if result:
            continue
        else:
            # log_info('get port return {}'.format(port))
            return port
    return 3456

def appium_start(self,host, port):
    '''
    启动appium server
    '''
    # bootstrap_port = str(port + 1)
    cmd = 'appium -a {} -p {} --session-override --log-level info'.format(str(host),str(port))
    # cmd = 'appium -a ' + host + ' -p ' + str(port) + ' -bp ' + str(bootstrap_port)
    print('%s at %s' % (cmd, ctime()))
    subprocess.Popen(cmd, shell=True, stdout=open(root_path +'/appium_log/' + str(port) + '.log', 'a'), stderr=subprocess.STDOUT)

def start_appium_action(self,host, port):
    '''
    检测端口是否被占用,如果没有被占用则启动appium服务
    '''
    self.appium_start(host, port)
    return True


def release_port(self,port):
    """释放指定的端口"""
    try:
        if port is not None:
            result, pid = self.__check_port_is_used(port)
            p = platform.system()
            if p == "Windows":
                sys_command = "taskkill /pid %s -t -f" % pid
                info = subprocess.check_output(sys_command)
                print(str(info, encoding='GB2312'))
            elif p == "Darwin" or p == "Linux":
                sys_command = "kill -9 %s" % pid
                os.system(sys_command)
    except Exception as e:
        raise e

def appium_start_sync(self,devices_list):
    '''
    并发启动appium服务
    '''
    print('====appium_start_sync=====')
    devices_dict = {}
    # 构建appium进程组
    appium_process = []
    # 加载appium进程
    for i in range(len(devices_list)):
        port = self.__get_device_port()
        #localhost:4723
        url = str(self.host) + ":" + str(port)
        print("devices_list[i]",devices_list[i])
        devices_dict[devices_list[i]] = url
        appium = multiprocessing.Process(target=self.start_appium_action, args=(self.host, port))
        appium_process.append(appium)
    # 启动appium服务
    for appium in appium_process:
        appium.start()
    for appium in appium_process:
        appium.join()
    sleep(5)
    return devices_dict









class App(BasePage):

def start_appium(self,devices_list):
    self.multiAppium = MultiAppium()
    self.devices_dict = self.multiAppium.appium_start_sync(devices_list)
    return self.devices_dict

def devices_start_sync(self,devices_list):
    '''并发启动设备'''
    devices_dict = self.start_appium(devices_list)
    print('===devices_start_sync===')
    # 定义desired进程组
    desired_process = []
    #进程之间共享变量
    manager = multiprocessing.Manager()
    open_drivers = manager.dict()
    # 加载desired进程
    for i in range(len(devices_dict)):
        url = devices_dict[devices_list[i]]
        desired = multiprocessing.Process(target=self.start, args=(open_drivers,devices_list[i],url))
        desired_process.append(desired)
    # 并发启动App
    for desired in desired_process:
        desired.start()
    for desired in desired_process:
        desired.join()
    return open_drivers,desired_process

def start(self,open_drivers,uid='',url=''):
    caps = parse_yaml(get_yamlpath(__file__))['android']
    for param in caps:
        value = caps[param]
        if "${" in str(value):
            newValue = re.sub('\$\{.*\}', uid, value)
            caps[param] = newValue
    print("caps:",caps)
    conn_url = 'http://{}/wd/hub'.format(url)
    print("conn_url",conn_url)
    self._driver = webdriver.Remote(conn_url, caps)
    tmp_driver = {uid:repr(self._driver)}
    open_drivers.update({'drivers':tmp_driver})
    self._driver.implicitly_wait(15)







class TestFunc(unittest.TestCase):
"""Test TestFunc.py"""

def test_add(self):

    """Test method add(a, b)"""
    devices_list = ["emulator-5554","SALDU17616001182"]
    self.app = App()
    drivers,desired_process = self.app.devices_start_sync(devices_list)
    print("全局变量的drivers",drivers)
    uid = 'emulator-5554'
    driver_emulator = drivers["drivers"][uid]

    d = dill.loads(driver_emulator.encode())

没有实际操作过,不知道你用共享变量的目的是什么,但是像driver这类的对象,普通的字典就可以存了,可以在启动进程的时候把对应的driver通过dict的形式保存下来,里面对应的url这些属性也可以通过嵌套dict的方式来存储,就类似python自带的dir函数。因为垃圾回收机制,有可能会出现浏览器没关闭但是driver找不到了的情况,那么可以用缓存(装饰器)来保存这个字典。个人觉得这种思路会简单一些,进程共享变量感觉复杂了一些