butubb
063818e086
用户实测要求："去掉切完账号下拉刷新这一步，直接改成等待 15 秒"。 原来的做法是"回读不到就再点一次「我」页签"—— 在设备上看着像去点刷新，还可能打断 抖音换号后的整体重载。 · tasks/generic/task.py：切完后按 wait_after（默认 15s，步骤参数可调）静等，等完只回读、 **不再点任何东西**（读不到就多读几次，页面渲染有零点几秒抖动） · 参数落到位：STEP_TYPES / 骨架 build_switch_step / 编辑器面板 / 发布页就地编辑字段表 · 测试：用例里 wait_after=0.2（不然一跑就一分多钟） · 文档：TASK_DEV §3 的 switch_account 行 + research/U2_ELEMENT_SELECTORS §七（记下"只能等"这条实测教训）
214 lines
8.3 KiB
Python
214 lines
8.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""P4 测试：`switch_account`（按名字找控件点击 + 回读校验）—— 假设备，不碰真机。
|
|
|
|
不连 adb、不连真机：`d` 是个模拟对象，只实现 u2 里这一步用到的那几个成员。
|
|
"""
|
|
import io
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
|
|
os.environ['DATABASE_URL'] = 'sqlite:///:memory:'
|
|
os.environ['DEPLOY_ENV'] = 'dev'
|
|
os.environ['DB_ALLOW_SQLITE_FALLBACK'] = '1'
|
|
# 仓库根 = scripts/tests/ 往上三级（不写死绝对路径，换机器也能跑）
|
|
_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
sys.path.insert(0, _ROOT)
|
|
os.chdir(_ROOT)
|
|
|
|
from tasks.generic import task as gt # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
def check(label, got, want):
|
|
ok = got == want
|
|
if not ok:
|
|
fails.append(label)
|
|
print((' ✅ ' if ok else ' ❌ ') + label + ('' if ok else '\n 得到 %r\n 期望 %r' % (got, want)))
|
|
|
|
|
|
class Exists:
|
|
"""u2 的 `exists` 语义：**既可当属性判真假、也可当函数带超时调用**。
|
|
|
|
真实 u2 里 `d.xpath(...).exists` 是纯 bool 属性，而 `d(text=...).exists` 是
|
|
可调用的 Exists 对象（`bool()` 即时判断、`exists(timeout=t)` 带等待）。
|
|
两处写法都要能被这一步的代码用，所以桩里做个两者通吃的。
|
|
"""
|
|
|
|
def __init__(self, flag):
|
|
self.flag = flag
|
|
|
|
def __call__(self, timeout=0):
|
|
return self.flag
|
|
|
|
def __bool__(self):
|
|
return self.flag
|
|
|
|
|
|
class El:
|
|
def __init__(self, text=None, exists=True, on_click=None):
|
|
self._t = text
|
|
self._on = on_click
|
|
self._e = Exists(exists)
|
|
|
|
@property
|
|
def exists(self):
|
|
return self._e
|
|
|
|
def wait(self, timeout=0):
|
|
return self._e.flag
|
|
|
|
def get_text(self, timeout=None):
|
|
return self._t
|
|
|
|
def click(self):
|
|
if self._on:
|
|
self._on()
|
|
|
|
|
|
class StubD:
|
|
"""模拟抖音：能读昵称、能进账号列表、能按名字点号。"""
|
|
|
|
def __init__(self, cur='小李', accounts=('小王', '小李'), login=False):
|
|
self.cur = cur
|
|
self.accounts = list(accounts)
|
|
self.login = login
|
|
self.clicked = []
|
|
self.switch_ok = True # False = 点了但没切过去（模拟没生效）
|
|
self.switch_open = True # False = 进不去账号列表
|
|
|
|
def _click_name(self, name):
|
|
self.clicked.append(name)
|
|
if self.switch_ok:
|
|
self.cur = name
|
|
self.accounts = [] # 真机上点成功后**面板会关掉**（那几行消失）——
|
|
# 步骤就是靠这个判断"点成功了"的：面板还在 = 这一下被吞了，要再点一次
|
|
|
|
LOGIN_HINTS = ('验证码', '其他方式登录', '手机号登录', '一键登录', '密码登录')
|
|
|
|
def xpath(self, v):
|
|
if v == '//name':
|
|
# 登录页上**根本没有昵称这个控件**（元素不存在）—— 这才是真实情况，
|
|
# 所以"读名字"会读到空串（而不是读到一堆登录页的文字）
|
|
return El(None, exists=False) if self.login else El(self.cur)
|
|
if v == '//switch':
|
|
return El('', exists=(self.switch_open and not self.login))
|
|
return El('', exists=False)
|
|
|
|
def __call__(self, **kw):
|
|
if 'descriptionContains' in kw:
|
|
return El('', exists=(kw['descriptionContains'] == '我'))
|
|
if 'textContains' in kw:
|
|
nm = kw['textContains']
|
|
if self.login and nm in self.LOGIN_HINTS:
|
|
return El(nm)
|
|
if nm in self.accounts:
|
|
return El(nm, on_click=lambda n=nm: self._click_name(n))
|
|
return El('', exists=False)
|
|
if 'text' in kw:
|
|
nm = kw['text']
|
|
if nm in self.accounts:
|
|
return El(nm, on_click=lambda n=nm: self._click_name(n))
|
|
return El('', exists=False)
|
|
return El('', exists=False)
|
|
|
|
|
|
PARAMS = {'target': 'var:target_name', 'switch_selector_type': 'xpath',
|
|
'switch_selector_value': '//switch', 'name_selector_type': 'xpath',
|
|
'name_selector_value': '//name', 'timeout': 1, 'max_attempts': 2,
|
|
# 切完后等它重载的秒数（真机默认 15s）—— 用例里调小，不然一跑就是一分多钟
|
|
'wait_after': 0.2}
|
|
|
|
notes = []
|
|
orig_notify = gt.notifier.notify
|
|
gt.notifier.notify = lambda ev, **kw: notes.append((ev, kw.get('want') or ''))
|
|
try:
|
|
print('=== 1. 已经是目标 → 直接成功，不点任何东西 ===')
|
|
w = gt.GenericStepsWorker("F1", {})
|
|
w._vars['target_name'] = '小李'
|
|
d = StubD(cur='小李')
|
|
check('返回 True', w._exec_switch_account(d, PARAMS), True)
|
|
check('一个号都没点', d.clicked, [])
|
|
|
|
print('=== 2. 切到另一个号（在列表里按名字点）===')
|
|
w = gt.GenericStepsWorker("F2", {})
|
|
w._vars['target_name'] = '小王'
|
|
d = StubD(cur='小李')
|
|
check('返回 True', w._exec_switch_account(d, PARAMS), True)
|
|
check('点是按名字点的', d.clicked, ['小王'])
|
|
check('当前号变了', d.cur, '小王')
|
|
check('顺手把运行期变量同步成新名字', w._vars.get('account_name'), '小王')
|
|
|
|
print('=== 3. 列表里没有这个名字（本机没登）→ 立刻 False、不重试 ===')
|
|
notes.clear()
|
|
w = gt.GenericStepsWorker("F3", {})
|
|
w._vars['target_name'] = '不存在的号'
|
|
d = StubD(cur='小李')
|
|
check('返回 False', w._exec_switch_account(d, PARAMS), False)
|
|
check('只找了一次（没重试）', d.clicked, [])
|
|
check('发了切号失败通知', [n[0] for n in notes], ['task.account.switch_failed'])
|
|
check('通知里带上想切的目标', notes[0][1], '不存在的号')
|
|
|
|
print('=== 4. 点了但没切过去 → 按 max_attempts 重试后放弃 + 通知 ===')
|
|
notes.clear()
|
|
w = gt.GenericStepsWorker("F4", {})
|
|
w._vars['target_name'] = '小王'
|
|
d = StubD(cur='小李')
|
|
d.switch_ok = False # 抖音没反应
|
|
check('返回 False', w._exec_switch_account(d, PARAMS), False)
|
|
# 面板一直没关（点一次吞一次）→ 每次尝试点两下、共 2 次尝试
|
|
check('试了 2 次尝试、每次点两下', d.clicked, ['小王', '小王', '小王', '小王'])
|
|
check('发了切号失败通知', [n[0] for n in notes], ['task.account.switch_failed'])
|
|
|
|
print('=== 5. 卡在登录/验证码页 → 通知 + 停本设备 ===')
|
|
notes.clear()
|
|
w = gt.GenericStepsWorker("F5", {})
|
|
w._vars['target_name'] = '小王'
|
|
d = StubD(cur='小李', login=True)
|
|
check('返回 False', w._exec_switch_account(d, PARAMS), False)
|
|
check('发了 need_login 通知', [n[0] for n in notes], ['task.account.need_login'])
|
|
check('本设备已停止', w.stopped(), True)
|
|
|
|
print('=== 6. on_login_page=continue → 不停设备（只记一句）===')
|
|
notes.clear()
|
|
w = gt.GenericStepsWorker("F6", {})
|
|
w._vars['target_name'] = '小王'
|
|
d = StubD(cur='小李', login=True)
|
|
p = dict(PARAMS, on_login_page='continue')
|
|
check('返回 False', w._exec_switch_account(d, p), False)
|
|
check('还是发了 need_login 通知', [n[0] for n in notes], ['task.account.need_login'])
|
|
check('设备没被停', w.stopped(), False)
|
|
|
|
print('=== 7. 进账号列表的选择器没配 → 明确报错、不乱点 ===')
|
|
notes.clear()
|
|
w = gt.GenericStepsWorker("F7", {})
|
|
w._vars['target_name'] = '小王'
|
|
d = StubD(cur='小李')
|
|
p = dict(PARAMS, switch_selector_value='')
|
|
check('返回 False', w._exec_switch_account(d, p), False)
|
|
check('没点任何东西', d.clicked, [])
|
|
check('没瞎发通知（这是配置问题，不是切号失败）', notes, [])
|
|
|
|
print('=== 8. 没给目标名字 → 跳过 ===')
|
|
w = gt.GenericStepsWorker("F8", {})
|
|
d = StubD(cur='小李')
|
|
check('返回 False', w._exec_switch_account(d, PARAMS), False)
|
|
check('没点任何东西', d.clicked, [])
|
|
|
|
print('=== 9. verify=False → 点了就算过（不回读）===')
|
|
w = gt.GenericStepsWorker("F9", {})
|
|
w._vars['target_name'] = '小王'
|
|
d = StubD(cur='小李')
|
|
d.switch_ok = False
|
|
check('返回 True', w._exec_switch_account(d, dict(PARAMS, verify=False)), True)
|
|
finally:
|
|
gt.notifier.notify = orig_notify
|
|
|
|
print()
|
|
if fails:
|
|
print('❌ 失败 %d 项：%s' % (len(fails), ' / '.join(fails)))
|
|
sys.exit(1)
|
|
print('✅ 全部通过（没连任何真机）')
|