feat(发布计划): 新增「校验页面」步骤(expect_el)—— 每一步先确认上一步真的完成了

用户要求:"进行每一步的时候先校验上一步有没有完成,比如相册界面肯定有特征,如果没有到相册界面
就出现执行,你看看怎么处理" → 方案 A:新增只做校验、不带副作用的步骤。

· tasks/generic/task.py:新步骤 expect_el
  - 参数:特征元素选择器 / timeout(10) / what("在等什么") / on_missing
  - on_missing:skip_rest(默认,**跳过本轮剩余步骤** —— 这一条不发、换下一个号) / notify / stop
  - 为此加了 worker 级的 `_skip_rest`:_exec_steps 每步前检查,foreach_account 每轮复位
· 骨架:关键点击前各插一个 —— 相册页(//*[@text='所有照片']) → 素材已选中(desc 选择/预览)
  → 剪辑页(desc 文字/话题) → 发布页(含'添加标题'/'添加作品描述') → 发布按钮
· 通知:新增 task.page.mismatch(页面不对,本轮剩余步骤已跳过)
· 昵称选择器改**两版通吃**(实测 A06=40.7.0):xpath 联合
  `//*[@resource-id="…:id/tkv"] | //*[@resource-id="…:id/tk+"]`
  —— 40.7 是 tkv、40.6 是 tk+,联合在 40.7 上实测能读到名字;.env(dev)已改,切号入口两版同为 desc=切换账号
· 前端:编辑器面板 + 参数表单 + 发布页就地编辑字段表都认 expect_el
· 文档:TASK_DEV §3(27→28 种)+ §4.7("每步先校验上一步")、NOTIFY、research §七·补(40.7.0 对照表
  + 联合 xpath + 各页特征 + "别按坐标点页签"的坑)、.env.example
· 测试:新增 scripts/tests/test_expect_el.py(特征在/不在、三种 on_missing、选择器为空),全套 7 个用例通过
This commit is contained in:
2026-10-12 11:49:48 +08:00
parent 50255af899
commit 10d2eb8334
12 changed files with 328 additions and 14 deletions
+145
View File
@@ -0,0 +1,145 @@
# -*- coding: utf-8 -*-
"""「校验页面」(expect_el) 用例:看到就过;没看到按 on_missing 处置(默认跳过本轮剩余步骤)。
假设备,不连真机。重点验两件事:
① 特征元素在 → True,且**不影响**后面的步骤;
② 特征元素不在 → 默认 `skip_rest`:**本轮剩下的步骤一步都不执行**(不盲点后面的按钮)。
"""
import io
import os
import sqlite3
import sys
import tempfile
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
os.environ['DATABASE_URL'] = 'sqlite:///' + os.path.join(tempfile.mkdtemp(), 'ac.db').replace('\\', '/')
os.environ['DATA_VIDEO_DIR'] = tempfile.mkdtemp()
os.environ['DEPLOY_ENV'] = 'dev'
os.environ['DB_ALLOW_SQLITE_FALLBACK'] = '1'
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
os.chdir(sys.path[0])
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 El:
def __init__(self, flag):
self._flag = flag
@property
def exists(self):
return _Ex(self._flag)
def wait(self, timeout=0):
return self._flag
def get_text(self, timeout=None):
return ""
class _Ex:
def __init__(self, f):
self.f = f
def __call__(self, timeout=0):
return self.f
def __bool__(self):
return self.f
class StubD:
"""`present` 里有的 xpath 才算存在(用来模拟"这一页的特征元素")。"""
def __init__(self, present=()):
self.present = set(present)
self.calls = []
def xpath(self, v):
self.calls.append(v)
return El(v in self.present)
def __call__(self, **kw):
v = kw.get('selector_value') or ''
self.calls.append('kw:%s' % kw)
return El(v in self.present)
notes = []
orig = gt.notifier.notify
gt.notifier.notify = lambda ev, **kw: notes.append((ev, kw.get('want') or ''))
try:
print('=== 1. 特征在 → True,后面的步骤照跑 ===')
P = {'selector_type': 'xpath', 'selector_value': '//*[@text="所有照片"]',
'timeout': 1, 'on_missing': 'skip_rest', 'what': '相册页'}
w = gt.GenericStepsWorker("F1", {})
d = StubD(present=['//*[@text="所有照片"]'])
check('返回 True', w._exec_expect_el(d, P), True)
check('没置"跳过剩余"', w._skip_rest, False)
check('没发通知', notes, [])
print('=== 2. 特征不在 → 默认 skip_rest:**本轮剩余步骤一步都不执行** ===')
notes.clear()
w = gt.GenericStepsWorker("F2", {})
d = StubD(present=[]) # 特征不在(没到相册页)
ran = []
real_one = gt.GenericStepsWorker._exec_one
def fake_one(dd, step, depth=0):
"""真跑 expect_el 那一步;其它步骤只记一笔(被跳过就不该记)。"""
if step.get('type') == 'expect_el':
return real_one(w, dd, step, depth)
ran.append(step.get('label'))
w._exec_one = fake_one
steps = [{"type": "expect_el", "label": "校验:应在相册页", "params": P},
{"type": "click", "label": "点第一个素材", "params": {}},
{"type": "click", "label": "点下一步", "params": {}}]
w._exec_steps(d, steps)
check('后面的点击一步都没执行', ran, [])
check('置了"跳过剩余"', w._skip_rest, True)
check('发了页面不对的通知', [n[0] for n in notes], ['task.page.mismatch'])
check('通知里带上"在等什么"', notes[0][1], '相册页')
print('=== 3. on_missing=stop → 停本设备 ===')
notes.clear()
w = gt.GenericStepsWorker("F3", {})
d = StubD(present=[])
check('返回 False', w._exec_expect_el(d, dict(P, on_missing='stop')), False)
check('本设备已停', w.stopped(), True)
check('发了通知', [n[0] for n in notes], ['task.page.mismatch'])
check('没置"跳过剩余"(停设备就够了)', w._skip_rest, False)
print('=== 4. on_missing=notify → 只通知、继续 ===')
notes.clear()
w = gt.GenericStepsWorker("F4", {})
d = StubD(present=[])
check('返回 False', w._exec_expect_el(d, dict(P, on_missing='notify')), False)
check('设备没停', w.stopped(), False)
check('没置"跳过剩余"', w._skip_rest, False)
check('发了通知', [n[0] for n in notes], ['task.page.mismatch'])
print('=== 5. 选择器为空 → 明确报错,不乱判 ===')
notes.clear()
w = gt.GenericStepsWorker("F5", {})
d = StubD(present=[])
check('返回 None', w._exec_expect_el(d, dict(P, selector_value='')), None)
check('没置"跳过剩余"', w._skip_rest, False)
check('没发通知', notes, [])
finally:
gt.notifier.notify = orig
print()
if fails:
print('❌ 失败 %d 项:%s' % (len(fails), ' / '.join(fails)))
sys.exit(1)
print('✅ 全部通过(没连任何真机)')