# -*- coding: utf-8 -*- """P6 测试：按名字找设备推送（`_push_by_name`）—— 临时库 + 假探测/假推送，不碰真机。""" import io import os import shutil import sys import tempfile import time sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') TMP = tempfile.mkdtemp(prefix='ac_p6_') os.environ['DATABASE_URL'] = 'sqlite:///' + TMP.replace(chr(92), '/') + '/ac.db' os.environ['DATA_VIDEO_DIR'] = os.path.join(TMP, 'videos') 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 flask import Flask # noqa: E402 from core import db_config, models, device_pool # noqa: E402 from core import video_plan as vp # noqa: E402 from core import account_probe # noqa: E402 from tasks.generic import publish_flow as flow # noqa: E402 from web import video_plan_api as vapi # noqa: E402 DATE = '2026-09-12' 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))) app = Flask("t") uri = db_config.build_db_uri() app.config["SQLALCHEMY_DATABASE_URI"] = uri app.config["SQLALCHEMY_ENGINE_OPTIONS"] = db_config.engine_options(uri) app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False models.db.init_app(app) with app.app_context(): models.db.create_all() vp.init_app(app) device_pool.init_app(app) with app.app_context(): models.db.session.add(models.Device(serial='A:5555', name='A01', enabled=True, created_at=vp._now())) models.db.session.add(models.Device(serial='B:5555', name='A02', enabled=True, created_at=vp._now())) for i, (nm, s) in enumerate([('小王', 1), ('小王', 2), ('小李', 1)]): models.db.session.add(models.VideoPlan( id='p%d' % i, nickname=nm, phone='', device_name='', douyin_id='', serial='', release_date=DATE, seq=s, seq_auto=False, title='t', body='', video_file='x', video_name='x', status=vp.ST_READY, created_at=vp._now(), updated_at=vp._now())) models.db.session.commit() pushed = [] lock = [] def fake_push(serial, plan, opts=None, log=None): pushed.append((serial, plan['nickname'], plan['seq'], (opts or {}).get('device_name'))) return True, "ok" print('=== 1. 名字 → 设备（两台设备各登一个）===') probe = {'A:5555': {'serial': 'A:5555', 'name': '小王', 'error': ''}, 'B:5555': {'serial': 'B:5555', 'name': '小李', 'error': ''}} orig_probe, orig_push = account_probe.probe_online, flow.push_one account_probe.probe_online = lambda serials=None, skip_busy=True, workers=4: probe flow.push_one = fake_push try: with app.app_context(): rows = vp.pushable_plans(DATE) matched, unmatched = vapi._push_by_name(rows) deadline = time.time() + 3 while len(pushed) < 3 and time.time() < deadline: time.sleep(0.05) check('两个名字都找到设备', sorted((m['name'], m['count']) for m in matched), [('小李', 1), ('小王', 2)]) check('小王的设备是 A01', [m['device_name'] for m in matched if m['name'] == '小王'], ['A01']) check('没有 unmatched', unmatched, []) check('推了小王的 2 条 + 小李的 1 条', len(pushed), 3) check('推给 A 的是小王的两条（按编号）', sorted((s, q) for (ser, n, q, dn) in pushed if ser == 'A:5555'), [(1, 1), (1, 2)]) check('推给 B 的是小李', [(n, q) for (ser, n, q, dn) in pushed if ser == 'B:5555'], [('小李', 1)]) check('顺手把设备名字传给了 push_one（写 posted_device_name 用）', {dn for (ser, n, q, dn) in pushed}, {'A01', 'A02'}) with app.app_context(): cached = [(d.serial, d.cur_account_name) for d in models.Device.query.order_by(models.Device.serial)] check('探测结果缓存进了 device 表', cached, [('A:5555', '小王'), ('B:5555', '小李')]) print('=== 2. 没人登着某个名字 → unmatched 里逐条说明 ===') pushed.clear() probe2 = {'A:5555': {'serial': 'A:5555', 'name': '小王', 'error': ''}, 'B:5555': {'serial': 'B:5555', 'name': '别的号', 'error': ''}} account_probe.probe_online = lambda serials=None, skip_busy=True, workers=4: probe2 with app.app_context(): rows = vp.pushable_plans(DATE) matched2, unmatched2 = vapi._push_by_name(rows) deadline = time.time() + 3 while len(pushed) < 2 and time.time() < deadline: time.sleep(0.05) check('只推了小王那 2 条', len(pushed), 2) check('小李进了 unmatched', [(u['name'], u['count']) for u in unmatched2], [('小李', 1)]) check('原因说清了「先切号」', ('没有在线设备登着' in unmatched2[0]['reason'] and '小李' in unmatched2[0]['reason'] and '先到设备上切到它' in unmatched2[0]['reason']), True) check('原因里带上当前探测到的号（排查用，别把整个 dict 打出来）', unmatched2[0]['reason'].endswith('（当前探测到：A:5555(小王)、B:5555(别的号)）'), True) if 'dict' in unmatched2[0]['reason'] or '{' in unmatched2[0]['reason']: print(' 原因里混进了原始 dict：', unmatched2[0]['reason']) print('=== 3. 探测失败（比如选择器没配）→ 也算没设备登着，不炸 ===') pushed.clear() account_probe.probe_online = lambda serials=None, skip_busy=True, workers=4: { 'A:5555': {'serial': 'A:5555', 'name': '', 'error': '还没抓昵称控件'}, 'B:5555': {'serial': 'B:5555', 'name': '', 'error': '连不上设备'}} with app.app_context(): rows = vp.pushable_plans(DATE) matched3, unmatched3 = vapi._push_by_name(rows) check('一条都没推', pushed, []) check('两个名字都 unmatched', sorted(u['name'] for u in unmatched3), ['小李', '小王']) with app.app_context(): keep = models.Device.query.get('A:5555').cur_account_name check('失败的那台不清空旧缓存', keep, '小王') finally: account_probe.probe_online, flow.push_one = orig_probe, orig_push print() shutil.rmtree(TMP, ignore_errors=True) if fails: print('❌ 失败 %d 项：%s' % (len(fails), ' / '.join(fails))) sys.exit(1) print('✅ 全部通过（临时库已清理）')