Airtest airtest 增加了观察者函数

zailushang · September 09, 2019 · Last by zailushang replied at September 12, 2019 · 758 hits

一直困扰于Airtest没有观察者函数,所以自己实现了一个,原理简单,python子线程实现

利用子线程的方式实现:

def loop_watcher(find_element, timeout):
"""
每隔一秒,循环查找元素是否存在.
如果元素存在,click操作
:param find_element: 要查找元素,需要是poco对象
:param timeout: 超时时间,单位:秒
:return:
"""

log('thread %s is running...' % threading.current_thread().name)
start_time = time.time()
while True:
if find_element.exists():
find_element.click()
log("观察者线程发现")
break
elif (time.time() - start_time) < timeout:
log("观察者线程等待1秒")
time.sleep(1)
else:
log("观察者线程超时未发现")
break
log('thread %s ended.' % threading.current_thread().name)


def watcher(text=None, textMatches=None, timeout=5, poco=None):
"""
观察者函数: 根据text或textMatches定位元素,用守护线程的方式,增加子线程循环查找元素,直到超时
:param text: 元素的text
:param textMatches: 元素的textMatches,正则表达式
:param timeout: 超时时间
:param poco: poco实例
:return:
"""

# 目标元素
find_element = None
if poco is None:
raise Exception("poco is None")
if text or textMatches:
if text:
find_element = poco(text=text)
elif textMatches:
find_element = poco(textMatches=textMatches)

# 定义子线程: 循环查找目标元素
t = threading.Thread(target=loop_watcher, name='LoopWatcherThread', args=(find_element, timeout))
# 增加守护线程,主线程结束,子线程也结束
t.setDaemon(True)
t.start()
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
共收到 7 条回复 时间 点赞

先收藏,
苦于此久矣

watchman 表人啊

hellohell 回复

什么意思啊?没看懂

有个问题啊,如果元素没找到,timeout 就减一秒。这里是不是没考虑找元素find element方法执行的耗时?

Jerry li 回复

是的,这个时间没有考虑,严格来看应该要考虑,我优化一下,多谢

你会不会出现两条线程同时写log文件,最后生成的报告会有点乱的情况

直南瓜 回复

不会,这边每次生成的报告,都生成到了指定文件内,递增的顺序

需要 Sign In 后方可回复, 如果你还没有账号请点击这里 Sign Up