You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
88 lines
2.8 KiB
88 lines
2.8 KiB
6 days ago
|
# -*- coding: utf-8 -*-
|
||
|
|
"""把 accept-selftest.py 的 wait helpers 从 wait_for_function（浏览器端求值，subBox 为动态元素时
|
||
|
|
静默 TypeError 致等待失效）改为 Python 侧轮询（行为可预测）。"""
|
||
|
|
import io
|
||
|
|
import sys
|
||
|
|
|
||
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
||
|
|
|
||
|
|
P = r'd:\code\crm-backend-matt\.scratch\opportunity-acceptance\accept-selftest.py'
|
||
|
|
src = open(P, encoding='utf-8').read()
|
||
|
|
|
||
|
|
start = src.index('def wait_ready(page')
|
||
|
|
end = src.index('def last_toast(page)')
|
||
|
|
new_helpers = '''def poll(page, fn, timeout):
|
||
|
|
"""Python 侧轮询：fn(page) 返回 truthy 即通过；超时静默（由后续断言判 FAIL）。"""
|
||
|
|
end = time.time() + timeout / 1000.0
|
||
|
|
while time.time() < end:
|
||
|
|
try:
|
||
|
|
if fn(page):
|
||
|
|
return
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
page.wait_for_timeout(300)
|
||
|
|
|
||
|
|
|
||
|
|
def wait_ready(page, timeout=9000):
|
||
|
|
"""等 listBody/subBox/detailBox/ruleListBox 无「加载中」且 boardBox 无 loading 态。"""
|
||
|
|
def fn(pg):
|
||
|
|
for i in ('listBody', 'subBox', 'detailBox', 'ruleListBox'):
|
||
|
|
el = pg.locator('#' + i)
|
||
|
|
if el.count() and '加载中' in (el.text_content() or ''):
|
||
|
|
return False
|
||
|
|
return pg.locator('#boardBox .loading').count() == 0
|
||
|
|
poll(page, fn, timeout)
|
||
|
|
|
||
|
|
|
||
|
|
def wait_stage(page, timeout=9000):
|
||
|
|
poll(page, lambda pg: pg.locator('#stageBar .node').count() > 0, timeout)
|
||
|
|
|
||
|
|
|
||
|
|
def wait_logged(page, timeout=9000):
|
||
|
|
poll(page, lambda pg: '已登录' in (pg.locator('#whoBox').text_content() or ''), timeout)
|
||
|
|
|
||
|
|
|
||
|
|
def wait_expr(page, js_expr, timeout=12000):
|
||
|
|
def fn(pg):
|
||
|
|
try:
|
||
|
|
return bool(pg.evaluate(js_expr))
|
||
|
|
except Exception:
|
||
|
|
return False
|
||
|
|
poll(page, fn, timeout)
|
||
|
|
|
||
|
|
|
||
|
|
def wait_sub_has(page, text, timeout=12000):
|
||
|
|
poll(page, lambda pg: pg.locator('#subBox').count() > 0 and text in (pg.locator('#subBox').text_content() or ''), timeout)
|
||
|
|
|
||
|
|
|
||
|
|
def wait_sub_not(page, text, timeout=12000):
|
||
|
|
poll(page, lambda pg: pg.locator('#subBox').count() > 0 and text not in (pg.locator('#subBox').text_content() or ''), timeout)
|
||
|
|
|
||
|
|
|
||
|
|
def wait_badge(page, text, timeout=12000):
|
||
|
|
def fn(pg):
|
||
|
|
b = pg.locator('#detailBox .badge')
|
||
|
|
return b.count() > 0 and (b.first.text_content() or '').strip() == text
|
||
|
|
poll(page, fn, timeout)
|
||
|
|
|
||
|
|
|
||
|
|
'''
|
||
|
|
|
||
|
|
src = src[:start] + new_helpers + src[end:]
|
||
|
|
|
||
|
|
# import time
|
||
|
|
if '\nimport time\n' not in src:
|
||
|
|
src = src.replace('\nimport sys\n', '\nimport sys\nimport time\n', 1)
|
||
|
|
|
||
|
|
# wait_text 调用改名为 wait_expr（23 段两处）
|
||
|
|
n = src.count('wait_text(page,')
|
||
|
|
src = src.replace('wait_text(page,', 'wait_expr(page,')
|
||
|
|
print('wait_text 调用改名:', n, '处')
|
||
|
|
|
||
|
|
open(P, 'w', encoding='utf-8').write(src)
|
||
|
|
print('helpers 已替换为 Python 轮询实现')
|
||
|
|
|
||
|
|
import ast
|
||
|
|
ast.parse(src)
|
||
|
|
print('syntax OK, lines:', src.count('\n') + 1)
|