Airtest airtest 增加了观察者函数

zailushang · September 09, 2019 · Last by zailushang replied at September 17, 2019 · 1623 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()
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
共收到 11 条回复 时间 点赞

先收藏,
苦于此久矣

watchman 表人啊

hellohell 回复

什么意思啊?没看懂

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

Jerry li 回复

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

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

直南瓜 回复

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

第一个没搞懂为啥要子进程做循环,我感觉只要特定的时间去做一下是否存在的判断就好了,返回true和false~这个功能我记得原来就有

zlp 回复

特定时间做是否存在的判断:
1、不灵活,也不稳定,特定时间本来就是一个不确定的事情;
2、特定时间去做判断,有时间差,务必需要用循环,每隔一段时间就判断一次,耗时太多,影响主线程的用例执行
3、兼容性,并不需要所有的设备都会触发某些场景,所以不需要每个设备上运行的用例都在主线程中做判断

可以考虑直接到github上面提一个pr呀,测试没问题的话就会被merge到代码主干上面,以后大家就能直接用了

yimelia-github 回复

嗯嗯,已提建议,多谢

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