376 lines
21 KiB
Python
376 lines
21 KiB
Python
# End-to-end test of a brain on a server working on a person's computer, all on this machine and
|
||
# all made up for the test: an account server, a headless brain, a headless "hands" computer (it
|
||
# also plays the person's screen, as a Mac showing the brain does), and a scripted fake model.
|
||
# Never touches a real account, and no Docker box (the bot's own computer is a sealed folder).
|
||
# Build first (cargo build --release; and in relay/), then, from app/:
|
||
# python3 -I deploy/test/remote_e2e.py .
|
||
# On a Mac the terminal needs 輔助使用 (the hands open Calculator through agent-desktop).
|
||
import json, os, shutil, socket, subprocess, sys, tempfile, threading, time, urllib.request, urllib.error
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
|
||
APP = os.path.abspath(sys.argv[1])
|
||
BIN = os.path.join(APP, 'target/release/anybot')
|
||
RELAY = os.path.join(APP, 'relay/target/release/anybot-relay')
|
||
WORK = tempfile.mkdtemp(prefix='anybot-e2e-')
|
||
USER, PW = 'e2euser', 'e2e-pass-12345'
|
||
HANDS = 'E2E手腳'
|
||
results = []
|
||
|
||
def free_port():
|
||
s = socket.socket(); s.bind(('127.0.0.1', 0)); p = s.getsockname()[1]; s.close(); return p
|
||
|
||
def check(name, ok, detail=''):
|
||
results.append((name, bool(ok), detail))
|
||
print(('PASS ' if ok else 'FAIL ') + name + ('' if ok else f' — {detail}'), flush=True)
|
||
|
||
# ---------- the scripted model ----------
|
||
def tool_call(name, args):
|
||
return {'tool': (name, json.dumps(args, ensure_ascii=False))}
|
||
|
||
SCRIPTS = {
|
||
'SC_LIST': [tool_call('list_computers', {})],
|
||
'SC_OPEN': [tool_call('computer_use_open_app', {'app': 'Calculator', 'computer': HANDS})],
|
||
'SC_SHELL': [tool_call('shell', {'command': 'echo e2e-$((6*7))', 'computer': HANDS})],
|
||
'SC_REDIRECT': [tool_call('shell', {'command': 'echo moved-here', 'computer': HANDS})],
|
||
'SC_HANDOFF': [tool_call('hand_over_computer', {'instruction': 'E2E：請按「我完成了」', 'computer': HANDS})],
|
||
'另一台：': [tool_call('hand_over_computer', {'instruction': 'E2E：在另一台', 'computer': 'E2E另一台'})],
|
||
'SC_OWN': [tool_call('shell', {'command': 'echo on-own-$((2+3))'})],
|
||
}
|
||
|
||
class Model(BaseHTTPRequestHandler):
|
||
def log_message(self, *a):
|
||
pass
|
||
def do_GET(self):
|
||
body = json.dumps({'data': [{'id': 'e2e'}]}).encode()
|
||
self.send_response(200); self.send_header('Content-Type', 'application/json'); self.send_header('Content-Length', str(len(body))); self.end_headers(); self.wfile.write(body)
|
||
def do_POST(self):
|
||
n = int(self.headers.get('Content-Length', '0') or 0)
|
||
req = json.loads(self.rfile.read(n) or b'{}')
|
||
msgs = req.get('messages', [])
|
||
last_user = max((i for i, m in enumerate(msgs) if m.get('role') == 'user'), default=-1)
|
||
text = msgs[last_user].get('content', '') if last_user >= 0 else ''
|
||
if isinstance(text, list):
|
||
text = ' '.join(p.get('text', '') for p in text if isinstance(p, dict))
|
||
done = sum(1 for m in msgs[last_user + 1:] if m.get('role') == 'tool')
|
||
last_tool = next((m.get('content', '') for m in reversed(msgs) if m.get('role') == 'tool'), '')
|
||
script = next((v for k, v in SCRIPTS.items() if k in text), [])
|
||
if done < len(script):
|
||
name, args = script[done]['tool']
|
||
chunk = {'choices': [{'delta': {'tool_calls': [{'index': 0, 'id': f'c{int(time.time()*1000)}', 'function': {'name': name, 'arguments': args}}]}}]}
|
||
else:
|
||
chunk = {'choices': [{'delta': {'content': 'E2E 完成：' + (last_tool if isinstance(last_tool, str) else '')[:300]}}]}
|
||
self.send_response(200); self.send_header('Content-Type', 'text/event-stream'); self.end_headers()
|
||
self.wfile.write(('data: ' + json.dumps(chunk, ensure_ascii=False) + '\n\ndata: [DONE]\n\n').encode())
|
||
|
||
# ---------- relay helpers ----------
|
||
RP = free_port(); MP = free_port()
|
||
RURL = f'http://127.0.0.1:{RP}'
|
||
|
||
def http(method, url, body=None, token=None, timeout=60, raw=False):
|
||
data = None if body is None else json.dumps(body).encode()
|
||
req = urllib.request.Request(url, method=method, data=data, headers={'Content-Type': 'application/json', **({'Authorization': 'Bearer ' + token} if token else {})})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||
b = r.read(); return r.status, (b if raw else json.loads(b or b'null'))
|
||
except urllib.error.HTTPError as e:
|
||
b = e.read()
|
||
try:
|
||
return e.code, json.loads(b)
|
||
except Exception:
|
||
return e.code, b
|
||
|
||
procs = {}
|
||
def start(name, args, env):
|
||
log = open(os.path.join(WORK, name + '.log'), 'a')
|
||
e = dict(os.environ); e.update(env)
|
||
procs[name] = subprocess.Popen(args, env=e, stdout=log, stderr=log, cwd=WORK)
|
||
|
||
def stop(name):
|
||
p = procs.pop(name, None)
|
||
if p:
|
||
p.terminate()
|
||
try:
|
||
p.wait(10)
|
||
except Exception:
|
||
p.kill()
|
||
|
||
def template():
|
||
path = os.path.join(WORK, 'sample.json')
|
||
subprocess.run([BIN, '--sample-state', path], check=True)
|
||
t = json.load(open(path))
|
||
for k in ('channels', 'mcp_servers', 'skills', 'pages', 'review_allow_rules', 'review_ask_rules', 'remote_computers', 'folders'):
|
||
if k in t:
|
||
t[k] = []
|
||
t['providers'] = {'custom': {'api_key': '', 'base_url': f'http://127.0.0.1:{MP}/v1', 'default_model': 'e2e', 'known_models': ['e2e']}}
|
||
t['remote'] = {'lan': False, 'password': '', 'devices': [], 'mac_from_phone': False, 'device_link': False, 'relay_url': '', 'relay_user': '', 'host_id': '', 'host_token': '', 'host_name': '', 'backup_key': '', 'auto_backup': False, 'last_backup': 0}
|
||
b = t['bots'][0]
|
||
b.update({'name': 'Master', 'provider': 'Custom', 'model': 'e2e', 'messages': [], 'memories': [], 'allow_rules': [], 'last_turn': None, 'default_computer': 'Bot 的電腦', 'tools_enabled': True, 'instructions': ''})
|
||
b.pop('threads', None)
|
||
t['bots'] = [b]; t['primary_bot'] = b['id']; t['selected'] = b['id']
|
||
t['local_execution'] = False; t['direct_mode'] = False; t['auto_review'] = False
|
||
return t
|
||
|
||
NOBOX = {'DOCKER_HOST': 'unix:///nonexistent/docker.sock', 'PATH': '/usr/bin:/bin:/usr/sbin:/sbin'}
|
||
|
||
def run():
|
||
srv = ThreadingHTTPServer(('127.0.0.1', MP), Model)
|
||
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
||
os.makedirs(os.path.join(WORK, 'relay'))
|
||
start('relay', [RELAY, '--bind', '127.0.0.1', '--port', str(RP), '--data', os.path.join(WORK, 'relay')], {})
|
||
for _ in range(50):
|
||
try:
|
||
if http('GET', RURL + '/api/hello')[0] == 200:
|
||
break
|
||
except Exception:
|
||
time.sleep(0.2)
|
||
# The brain: a seeded bot on the fake model, joins (and makes) the account.
|
||
bdir = os.path.join(WORK, 'brain'); os.makedirs(bdir)
|
||
st = template(); st['remote']['host_name'] = 'E2E大腦'
|
||
json.dump(st, open(os.path.join(bdir, 'state.json'), 'w'), ensure_ascii=False)
|
||
benv = {'ANYBOT_DATA_DIR': bdir, 'ANYBOT_ACCOUNT': f'{RURL}|{USER}', 'ANYBOT_ACCOUNT_PASSWORD': PW, 'ANYBOT_ACCOUNT_CREATE': '1', 'ANYBOT_HOST_NAME': 'E2E大腦', **NOBOX}
|
||
start('brain', [BIN, '--serve'], benv)
|
||
brain_id = None
|
||
for _ in range(60):
|
||
time.sleep(0.5)
|
||
try:
|
||
s = json.load(open(os.path.join(bdir, 'state.json')))
|
||
if s['remote'].get('host_id'):
|
||
brain_id = s['remote']['host_id']; break
|
||
except Exception:
|
||
pass
|
||
check('brain joins the account', brain_id, 'no host id')
|
||
# The hands: a second AnyBot of the account, lending its hands to that brain only.
|
||
_, v = http('POST', RURL + '/relay/login', {'user': USER, 'password': PW, 'device': HANDS})
|
||
_, h = http('POST', RURL + '/relay/hosts', {'name': HANDS}, token=v['token'])
|
||
hdir = os.path.join(WORK, 'hands'); os.makedirs(hdir)
|
||
hs = template(); hs['bots'] = []
|
||
hs['remote'].update({'relay_url': RURL, 'relay_user': USER, 'host_id': h['id'], 'host_token': h['token'], 'host_name': HANDS, 'device_link': True, 'hands_for': brain_id, 'hands_for_name': 'E2E大腦'})
|
||
json.dump(hs, open(os.path.join(hdir, 'state.json'), 'w'), ensure_ascii=False)
|
||
henv = {'ANYBOT_DATA_DIR': hdir, **NOBOX}
|
||
start('hands', [BIN, '--serve'], henv)
|
||
# A second computer showing the same brain (as a desktop AnyBot left on the server does):
|
||
# with two, 「我的電腦」 has to be the one the person typed on, not a guess.
|
||
_, h2 = http('POST', RURL + '/relay/hosts', {'name': 'E2E另一台'}, token=v['token'])
|
||
h2dir = os.path.join(WORK, 'hands2'); os.makedirs(h2dir)
|
||
hs2 = template(); hs2['bots'] = []
|
||
hs2['remote'].update({'relay_url': RURL, 'relay_user': USER, 'host_id': h2['id'], 'host_token': h2['token'], 'host_name': 'E2E另一台', 'device_link': True, 'hands_for': brain_id, 'hands_for_name': 'E2E大腦', 'brain_id': brain_id, 'brain_name': 'E2E大腦'})
|
||
json.dump(hs2, open(os.path.join(h2dir, 'state.json'), 'w'), ensure_ascii=False)
|
||
start('hands2', [BIN, '--serve'], {'ANYBOT_DATA_DIR': h2dir, **NOBOX})
|
||
tok = h['token']
|
||
sync = lambda method, rest='', body=None, q='': http(method, f'{RURL}/relay/hosts/{brain_id}/api/sync{rest}{q}', body, token=tok)
|
||
|
||
def peers():
|
||
return {x['name']: x for x in http('GET', RURL + '/relay/host/peers', token=tok)[1]['hosts']}
|
||
for _ in range(40):
|
||
p = peers()
|
||
if p.get('E2E大腦', {}).get('online') and p.get(HANDS, {}).get('online') and p.get('E2E另一台', {}).get('online'):
|
||
break
|
||
time.sleep(0.5)
|
||
p = peers()
|
||
check('brain and hands are online', p.get('E2E大腦', {}).get('online') and p.get(HANDS, {}).get('online'), json.dumps(p, ensure_ascii=False))
|
||
# Tell the account this computer shows that brain (as the Mac's screen does).
|
||
copy = {'version': 0, 'msgs': {}}
|
||
|
||
def pull(full=False, client='screen'):
|
||
since = 0 if full else copy['version']
|
||
code, v = sync('GET', q=f'?since={since}&client={client}')
|
||
if code != 200:
|
||
raise RuntimeError(f'pull {code} {v}')
|
||
if client == 'screen':
|
||
copy['version'] = v.get('version', 0)
|
||
for d in v.get('bots', []):
|
||
cur = copy['msgs'].setdefault(d['id'], [])
|
||
if len(cur) < d['from']:
|
||
copy['version'] = 0 # the screen asks for everything again
|
||
continue
|
||
del cur[d['from']:]
|
||
cur.extend(d['messages'])
|
||
return v
|
||
|
||
# The brain looks the account's computers up every 15 s: a computer that just came on is
|
||
# there within that.
|
||
for _ in range(25):
|
||
v = pull()
|
||
comps = v.get('computers', {}).get('1', {})
|
||
names = [x['name'] for x in comps.get('names', [])]
|
||
if HANDS in names and 'E2E另一台' in names:
|
||
break
|
||
time.sleep(1)
|
||
check('computers by their own names; no server computer', sorted(names) == sorted(['Master 的電腦', HANDS, 'E2E另一台']), names)
|
||
check('the bot works on its own computer by default', comps.get('current') == 'Master 的電腦', comps)
|
||
|
||
def wait_turn(answer=None, timeout=90, during=None):
|
||
"""Until the turn ends; answers requests with `answer` ("once"/"reject")."""
|
||
t0 = time.time(); seen = []
|
||
while time.time() - t0 < timeout:
|
||
v = pull(client='poll')
|
||
for a in v.get('approvals', []):
|
||
key = (a.get('call_id'), a.get('kind'))
|
||
if key not in [s[:2] for s in seen]:
|
||
seen.append((a.get('call_id'), a.get('kind'), a))
|
||
if during:
|
||
during(a)
|
||
elif answer:
|
||
sync('POST', '/act', {'bot': 1, 'action': 'answer', 'answer': answer})
|
||
if not v.get('live') and not v.get('approvals') and time.time() - t0 > 2.5:
|
||
return seen, v
|
||
time.sleep(0.5)
|
||
return seen, None
|
||
|
||
def last_tool(v):
|
||
msgs = [m for m in v['bots'][0]['messages'] if m['role'] == 'Tool'] if v and v.get('bots') else []
|
||
return msgs[-1]['text'] if msgs else ''
|
||
|
||
def full_messages():
|
||
return pull(full=True, client='check')['bots'][0]['messages']
|
||
|
||
def tool_text_after(marker):
|
||
ms = full_messages()
|
||
i = max((k for k, m in enumerate(ms) if m['role'] == 'User' and marker in m['text']), default=-1)
|
||
return [m['text'] for m in ms[i + 1:] if m['role'] == 'Tool']
|
||
|
||
# 1. 查看電腦: the person's computer is the one they typed on.
|
||
sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '我現在用哪台？SC_LIST'})
|
||
wait_turn('once')
|
||
t = ' '.join(tool_text_after('SC_LIST'))
|
||
check('list_computers names the hands as where the person is', HANDS in t and '使用者現在就是在這台' in t, t[:300])
|
||
# 4. Not asked for: back to the chosen computer, with a word for the bot.
|
||
sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '跑一下 SC_REDIRECT'})
|
||
wait_turn('once')
|
||
t = ' '.join(tool_text_after('SC_REDIRECT'))
|
||
check('a step the person did not ask for goes to the chosen computer, and the bot is told', 'moved-here' in t and '[Master 的電腦]' in t and '這一步你指定了' in t, t[:400])
|
||
# 2. A step on the person's computer, named by them: asks once, runs there (agent-desktop).
|
||
sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '在我的電腦打開計算機 SC_OPEN'})
|
||
seen, _ = wait_turn('once')
|
||
t = ' '.join(tool_text_after('SC_OPEN'))
|
||
check('open_app on the person computer asks first', any(s[1] == 'action' for s in seen), seen)
|
||
check('open_app ran on the hands via agent-desktop', f'[{HANDS}]' in t and ('已打開' in t), t[:400])
|
||
# 3. A command there.
|
||
sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '在我的電腦跑指令 SC_SHELL'})
|
||
wait_turn('once')
|
||
t = ' '.join(tool_text_after('SC_SHELL'))
|
||
check('shell runs on the hands', 'e2e-42' in t and f'[{HANDS}]' in t, t[:300])
|
||
# 5. Its own computer.
|
||
sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '用你自己的 SC_OWN'})
|
||
wait_turn('once')
|
||
t = ' '.join(tool_text_after('SC_OWN'))
|
||
check('own computer by default', 'on-own-5' in t and '[Master 的電腦]' in t, t[:300])
|
||
# 6. 需要處理: 我完成了.
|
||
sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '在我的電腦 SC_HANDOFF'})
|
||
got = {}
|
||
phone = {}
|
||
def on_handoff(a):
|
||
got.update(a)
|
||
# The phone (signed in as the person) sees it too, as 需要處理 with the computer.
|
||
_, ses = http('POST', RURL + '/relay/login', {'user': USER, 'password': PW, 'device': 'e2e-phone'})
|
||
code, view = http('GET', f'{RURL}/h/{brain_id}/api/state?since=0', token=ses.get('token'))
|
||
phone.update(next((x for x in (view or {}).get('approvals', []) if x.get('bot') == 1), {}) if code == 200 else {'code': code, 'body': str(view)[:200]})
|
||
sync('POST', '/act', {'bot': 1, 'action': 'answer', 'answer': 'once'})
|
||
wait_turn(during=on_handoff)
|
||
t = ' '.join(tool_text_after('在我的電腦 SC_HANDOFF'))
|
||
check('hand-off shows as 需要處理 with the instruction and computer', got.get('kind') == 'handoff' and got.get('reason') == 'E2E：請按「我完成了」' and got.get('grant') == HANDS, got)
|
||
check('我完成了 lets the bot carry on', '我完成了' in t, t[:300])
|
||
check('the phone shows the hand-off with its computer', phone.get('kind') == 'handoff' and phone.get('computer') == HANDS and not phone.get('can_always'), phone)
|
||
# 7. 需要處理: 跳過.
|
||
sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '再一次在我的電腦 SC_HANDOFF'})
|
||
wait_turn('reject')
|
||
t = ' '.join(tool_text_after('再一次在我的電腦 SC_HANDOFF'))
|
||
check('跳過 tells the bot the step is skipped', '跳過' in t, t[:300])
|
||
# 8. 需要處理: answered by a message.
|
||
sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '第三次在我的電腦 SC_HANDOFF'})
|
||
def on_reply(a):
|
||
sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '我已經登入好了'})
|
||
seen, v = wait_turn(during=on_reply)
|
||
t = ' '.join(tool_text_after('第三次在我的電腦 SC_HANDOFF'))
|
||
check('a message answers a hand-off', '傳了訊息' in t, t[:300])
|
||
ms = full_messages()
|
||
check('the reply is in the chat once', sum(1 for m in ms if m['role'] == 'User' and m['text'] == '我已經登入好了') == 1, [m['text'] for m in ms if m['role'] == 'User'])
|
||
# 8b. A turn started on one computer; while it waits, the person types on another one:
|
||
# 「我的電腦」 in that message is the computer it was typed on (the turn's own idea was fixed
|
||
# when it started).
|
||
tok2 = h2['token']
|
||
sync2 = lambda method, rest='', body=None, q='': http(method, f'{RURL}/relay/hosts/{brain_id}/api/sync{rest}{q}', body, token=tok2)
|
||
sync2('POST', '/act', {'bot': 1, 'action': 'send', 'text': '另一台：在這台 E2E另一台 請我接手'})
|
||
def steer_from_hands(a):
|
||
if a.get('kind') == 'handoff':
|
||
sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '算了，在我的電腦跑指令 SC_SHELL 插話'})
|
||
else:
|
||
sync('POST', '/act', {'bot': 1, 'action': 'answer', 'answer': 'once'})
|
||
wait_turn(during=steer_from_hands)
|
||
t = ' '.join(tool_text_after('SC_SHELL 插話'))
|
||
check('a message typed on another computer mid-turn: 「我的電腦」 is that one', 'e2e-42' in t and f'[{HANDS}]' in t and '這一步你指定了' not in t, t[:300])
|
||
# 9. The screen's copy, kept from small updates, matches the brain's chat exactly.
|
||
for _ in range(3):
|
||
pull()
|
||
mine = copy['msgs'].get(1, [])
|
||
full = full_messages()
|
||
sig = lambda ms: [(m['role'], m['text'], m.get('at')) for m in ms]
|
||
check('screen copy equals the brain chat (no duplicates)', sig(mine) == sig(full), f'{len(mine)} vs {len(full)}')
|
||
users = [m['text'] for m in full if m['role'] == 'User']
|
||
check('no user message twice', len(users) == len(set(users)), users)
|
||
# 10. Choosing a computer from the screen reaches the brain.
|
||
sync('POST', '/edit', {'client': 'screen', 'seq': 1, 'ops': [{'op': 'bot_set', 'id': 1, 'key': 'default_computer', 'value': HANDS}]})
|
||
time.sleep(1.0)
|
||
c = pull(full=True, client='check2')['computers']['1']
|
||
check('the computer chosen on the screen is the brain\'s choice', c['current'] == HANDS, c)
|
||
sync('POST', '/edit', {'client': 'screen', 'seq': 2, 'ops': [{'op': 'bot_set', 'id': 1, 'key': 'default_computer', 'value': 'Bot 的電腦'}]})
|
||
time.sleep(1.0)
|
||
c = pull(full=True, client='check3')['computers']['1']
|
||
check('and back to its own', c['current'] == 'Master 的電腦', c)
|
||
# 11. The hands go away: plain words, not codes.
|
||
stop('hands')
|
||
time.sleep(10)
|
||
sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '在我的電腦跑指令 SC_SHELL 二'})
|
||
wait_turn('once', timeout=120)
|
||
t = ' '.join(tool_text_after('SC_SHELL 二'))
|
||
check('hands offline: the bot hears it in words', ('連不到' in t or '離線' in t) and 'e2e-42' not in t and '這一步你指定了' not in t and not any(code in t for code in ('503', '502', '404')), t[:300])
|
||
start('hands', [BIN, '--serve'], henv)
|
||
for _ in range(40):
|
||
if peers().get(HANDS, {}).get('online'):
|
||
break
|
||
time.sleep(0.5)
|
||
sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '在我的電腦跑指令 SC_SHELL 三'})
|
||
wait_turn('once')
|
||
t = ' '.join(tool_text_after('SC_SHELL 三'))
|
||
check('hands back: works again', 'e2e-42' in t and f'[{HANDS}]' in t, t[:300])
|
||
# 12. The brain restarts: the screen picks everything up again, and work goes on.
|
||
before = len(full_messages())
|
||
stop('brain')
|
||
time.sleep(3)
|
||
start('brain', [BIN, '--serve'], benv)
|
||
for _ in range(60):
|
||
try:
|
||
if peers().get('E2E大腦', {}).get('online'):
|
||
v = pull()
|
||
break
|
||
except Exception:
|
||
pass
|
||
time.sleep(0.5)
|
||
time.sleep(2)
|
||
pull()
|
||
check('after a brain restart the screen has the whole chat', len(copy['msgs'].get(1, [])) >= before, f'{len(copy["msgs"].get(1, []))} < {before}')
|
||
sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '在我的電腦跑指令 SC_SHELL 四'})
|
||
wait_turn('once')
|
||
t = ' '.join(tool_text_after('SC_SHELL 四'))
|
||
check('after a brain restart, the first 「我的電腦」 is the computer the person typed on', 'e2e-42' in t and f'[{HANDS}]' in t and '這一步你指定了' not in t, t[:300])
|
||
pull(); pull()
|
||
check('screen copy still equals the brain chat', sig(copy['msgs'].get(1, [])) == sig(full_messages()))
|
||
# 13. 清除聊天記錄 from the screen clears the brain's chat.
|
||
sync('POST', '/act', {'bot': 1, 'action': 'clear'})
|
||
time.sleep(1.5)
|
||
check('clear from the screen empties the brain chat', full_messages() == [])
|
||
pull(); pull()
|
||
check('and the screen copy follows', copy['msgs'].get(1, []) == [], len(copy['msgs'].get(1, [])))
|
||
|
||
try:
|
||
run()
|
||
except Exception as e:
|
||
import traceback; traceback.print_exc()
|
||
check('harness ran to the end', False, repr(e))
|
||
finally:
|
||
for n in list(procs):
|
||
stop(n)
|
||
failed = [r for r in results if not r[1]]
|
||
print(f'\n{len(results) - len(failed)}/{len(results)} passed; logs in {WORK}')
|
||
sys.exit(1 if failed else 0)
|