auto_control/scripts/tests/test_switch_account.py

# -*- 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}

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('✅ 全部通过（没连任何真机）')