butubb
beda0f3779
发布计划不再绑设备、不再依赖账号台账：上传文件名 `名字_日期_编号`（标题行 `标题 || 正文_名字_日期_编号`），运行时读本机当前抖音昵称取计划；靠循环块 「一轮把本机待发的号跑完」自动切号，一直发到发完。 · 解析/配对换键：右锚定认名字（不许含下划线、不许纯数字 —— 不合法直接拒收并说清原因）， 删掉三处"手机号必须命中台账"的校验，加索引 ix_video_plan_name_date（启动时补建） · 新步骤三个：读本机账号名字 / 切换账号（**按名字点控件**、切完回读校验、遇登录页停本设备）/ 循环本机待发的号（试过不回头、跑一轮没进展就换号、停止·超时·上限三个干净出口） · 发布骨架改 18 步：读名字 → 校验账号（cmp_source=account_target）→ 推送（**推送前重读昵称核对**） → 抖音流程 → 标记结果；push_release 的 account 默认 name · 账号管理整页下线：删 core/ledger.py（serial_of 迁到 core/device_pool.py）、6 个 /api/ledger*、 「账号」页与 ledger.js、设备池「账号」列、设备端账号块；发布计划升为一级页签； device_account 表与 42 行历史数据保留、仍在备份清单内（**停用不等于删数据**） · 人工入口「按名字找设备」：新增 core/account_probe.py（只读探测、跳过在跑任务的设备） + POST /api/video_plan/push_by_account，没设备登着的号逐条报出来； 推送时把"实际是谁发的"写进 posted_serial / posted_device_name · 通知：新增 task.account.switch_failed / task.account.need_login · schema v12（posted_*）/ v13（device.cur_account_name / cur_name_at） · 文档同步：TASK_DEV / API / DATA_MODEL / ARCHITECTURE / NOTIFY / DEVICE_AGENT / DEPLOY / DEVELOPMENT / README / .env.example（两个抖音控件选择器抓一次填 .env） · 新增 scripts/tests/：6 套用例 + run_all.py（临时库 + 假设备，不碰真实数据、不连真机）
209 lines
7.9 KiB
Python
209 lines
7.9 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
|
||
|
||
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}
|
||
|
||
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)
|
||
check('试了 2 次（max_attempts=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('✅ 全部通过（没连任何真机）')
|