auto_control/scripts/tests/test_read_text.py

# -*- coding: utf-8 -*-
"""「读取文字」+ 选择器里的 `{变量}` +「条件判断→变量」用例。

假设备，不连真机。验三件事：
  ① 选择器值里写 `{本次要发的名字}` / `{本机账号名字}` → 真的换成本轮的值；
     变量是空的 → 换成空串、这一步找不到元素（**不能留着字面量去瞎点**）；
  ② 「读取文字」把元素文字读出来、清洗、存进指定变量（变量名叫 account_name 时顺手记下昵称控件）；
  ③ 「条件判断」候选值来源选「变量」→ 候选就是那个变量的值，空变量 = 不命中。
"""
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
from core import task_draft                                            # 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, text=""):
        self._flag = flag
        self._text = text
        self.clicked = False

    @property
    def exists(self):
        return _Ex(self._flag)

    def wait(self, timeout=0):
        return self._flag

    def get_text(self, timeout=None):
        return self._text

    def click(self):
        self.clicked = True


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:
    """`kw` 里记最后一次 d(**kw) 的参数；`texts` 给元素文本。"""

    def __init__(self, present=(), texts=None):
        self.present = set(present)
        self.texts = texts or {}
        self.last_kw = None
        self.clicked = []

    def xpath(self, v):
        return El(v in self.present, self.texts.get(v, ""))

    def __call__(self, **kw):
        self.last_kw = kw
        v = list(kw.values())[0]
        if v in self.present:
            self.clicked.append(v)
        return El(v in self.present, self.texts.get(v, ""))


print('=== 1. 选择器值里的 {本次要发的名字} 换成本轮目标 ===')
w = gt.GenericStepsWorker("F1", {})
w._vars['target_name'] = '锋哥苏州寻宝藏小店'
d = StubD(present=['锋哥苏州寻宝藏小店'])
check('点击命中（说明选择器已代入）',
      w._exec_click(d, {"selector_type": "textContains",
                        "selector_value": "{本次要发的名字}", "wait_timeout": 1}, 0), True)
check('实际用的选择器 = 名字本身', d.last_kw, {'textContains': '锋哥苏州寻宝藏小店'})

print('=== 2. 变量还是空的 → 选择器变空 → 不许瞎点 ===')
w = gt.GenericStepsWorker("F2", {})
d = StubD(present=[''])
check('返回 False（没找到）',
      w._exec_click(d, {"selector_type": "textContains",
                        "selector_value": "prefix-{本次要发的名字}", "wait_timeout": 1}, 0), False)
check('空变量 → 字符串里不留 {…} 字面量', d.last_kw, {'textContains': 'prefix-'})

print('=== 3. {本机账号名字} 用「读取文字」读到的值 ===')
w = gt.GenericStepsWorker("F3", {})
w._vars['account_name'] = '小王'
d = StubD(present=['小王'])
check('等待元素命中', w._exec_wait_el(d, {"selector_type": "text",
                                          "selector_value": "{本机账号名字}", "timeout": 1}, 0), True)
check('用的是本机账号名', d.last_kw, {'text': '小王'})

print('=== 4. 「读取文字」读出来、清洗、存变量 ===')
w = gt.GenericStepsWorker("F4", {})
d = StubD(present=['//*[@text="nick"]'],
          texts={'//*[@text="nick"]': '  锋哥苏州康养探店  \n编辑资料  \n抖音号：123'})
check('返回 True', w._exec_read_text(d, {"selector_type": "xpath",
                                        "selector_value": '//*[@text="nick"]',
                                        "var": "account_name", "clean": "account",
                                        "timeout": 1}, 0), True)
check('存进 account_name（按账号名清洗：取第一行非标签）',
      w._vars.get('account_name'), '锋哥苏州康养探店')
check('顺手记下昵称控件（切号/推送核对复用）',
      w._vars.get('name_selector'), ('xpath', '//*[@text="nick"]'))

print('=== 5. clean=strip：原样（只去首尾空白） ===')
w = gt.GenericStepsWorker("F5", {})
d = StubD(present=['//*[@text="x"]'],
          texts={'//*[@text="x"]': '  小王  \n编辑资料'})
w._exec_read_text(d, {"selector_type": "xpath", "selector_value": '//*[@text="x"]',
                      "var": "raw_name", "clean": "strip", "timeout": 1}, 0)
check('只去首尾空白、不按行取', w._vars.get('raw_name'), '小王  \n编辑资料')
check('变量名不是 account_name → 不记昵称控件', w._vars.get('name_selector'), None)

print('=== 6. 读不到：默认 skip（只记一句、不拦任务） ===')
w = gt.GenericStepsWorker("F6", {})
d = StubD(present=[])
check('返回 None（skip 不算失败）',
      w._exec_read_text(d, {"selector_type": "xpath", "selector_value": '//*[@text="没有"]',
                            "var": "account_name", "timeout": 1}, 0), None)
check('变量被置空', w._vars.get('account_name'), '')

print('=== 7. 读不到 + on_missing=fail → 算这一步失败 ===')
w = gt.GenericStepsWorker("F7", {})
d = StubD(present=[])
check('返回 False',
      w._exec_read_text(d, {"selector_type": "xpath", "selector_value": '//*[@text="没有"]',
                            "var": "v", "timeout": 1, "on_missing": "fail"}, 0), False)

print('=== 8. 「条件判断」候选值来源=变量 ===')
w = gt.GenericStepsWorker("F8", {})
w._vars['account_name'] = '小王'
val, note = w._resolve_cmp_value({"cmp_op": "等于", "cmp_source": "var",
                                  "cmp_var": "account_name"}, "")
check('候选值 = 变量的值', val, '小王')
check('来源说明写清了', 'account_name' in note and '小王' in note, True)

print('=== 9. 变量是空的 → 候选为空（永远不命中，走 else） ===')
w = gt.GenericStepsWorker("F9", {})
val, note = w._resolve_cmp_value({"cmp_op": "等于", "cmp_source": "var",
                                  "cmp_var": "account_name"}, "手填兜底")
check('只剩手填值', val, '手填兜底')
check('说明里点出变量是空的', '空' in note, True)

print('=== 10. 草稿校验认这两个新能力 ===')
ok_steps = [{"type": "read_text", "label": "读取文字",
             "params": {"selector_type": "xpath", "selector_value": '//*[@id="nick"]',
                        "var": "account_name"}},
            {"type": "if_el", "label": "条件判断",
             "params": {"selector_type": "xpath", "selector_value": '//*[@id="nick"]',
                        "cmp_op": "包含", "cmp_source": "var", "cmp_var": "account_name",
                        "then": [{"type": "notify", "label": "命中",
                                  "params": {"title": "ok", "message": "ok"}}],
                        "else": []}}]
errs, warns = [], []
task_draft.validate_steps(ok_steps, errors=errs, warnings=warns)
check('合法草稿没有 error', errs, [])

bad_steps = [{"type": "read_text", "label": "读取文字",
              "params": {"selector_type": "xpath", "selector_value": ""}}]
errs2 = []
task_draft.validate_steps(bad_steps, errors=errs2, warnings=[])
check('「读取文字」缺选择器 → 报错',
      any('selector_value' in e for e in errs2), True)

bad2 = [{"type": "if_el", "label": "条件判断",
         "params": {"selector_type": "xpath", "selector_value": '//*[@id="x"]',
                    "cmp_op": "包含", "cmp_source": "var", "then": [], "else": []}}]
errs3 = []
task_draft.validate_steps(bad2, errors=errs3, warnings=[])
check('来源=变量但没填变量名 → 报错',
      any('cmp_var' in e for e in errs3), True)

print()
if fails:
    print('❌ 失败 %d 项：%s' % (len(fails), ' / '.join(fails)))
    sys.exit(1)
print('✅ 全部通过（没连任何真机）')