多线程启动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())