mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-08-22 20:52:23 +02:00
99 lines
4.5 KiB
Python
99 lines
4.5 KiB
Python
"""Send-script mechanism, verified without a live webview: real LinkedIn-shaped
|
|
interactives fixtures driven through a mock executor exercise the exact path the
|
|
live rig would (opener -> composer -> fill -> commit-check -> send -> clear-check),
|
|
plus every abort/honesty branch. The wall-clock a live run measures is not here;
|
|
the correctness the live run would prove is."""
|
|
import pytest
|
|
|
|
from backend.apps.agents.browser import browser_send_script as ss
|
|
from backend.apps.agents.browser.browser_agent import send_index_in_state, payload_in_textbox
|
|
|
|
PROFILE = '[22]*<link "Tyler Chen Premium 1st">\n[50]*<link "Message">\n[51]<button "Follow">'
|
|
COMPOSER_EMPTY = '[2]<textbox "Write a message">\n[9]<button "Attach">'
|
|
COMPOSER_FILLED = '[2]<textbox "Write a message" value="[test] hello world r9-os">\n[14]<button "Send">'
|
|
COMPOSER_SENT = '[2]<textbox "Write a message">\n[9]<button "Attach">' # cleared, Send gone
|
|
|
|
TASK = "go to tyler chen's linkedin hes in entrepreneurs first and text him '[test] hello world r9-os'"
|
|
|
|
|
|
def make_exec(list_script):
|
|
"""execute_tool mock: each BrowserListInteractives call returns the next
|
|
scripted state; clicks/fills succeed and are recorded."""
|
|
calls = {"list": 0, "clicks": []}
|
|
states = list(list_script)
|
|
|
|
async def execute(tool, params, bid, tid):
|
|
if tool == "BrowserListInteractives":
|
|
i = min(calls["list"], len(states) - 1)
|
|
calls["list"] += 1
|
|
return {"text": states[i]}
|
|
calls["clicks"].append((tool, params))
|
|
return {"ok": True}
|
|
return execute, calls
|
|
|
|
|
|
async def run(task, state0, list_script):
|
|
ex, calls = make_exec(list_script)
|
|
r = await ss.run_send_script(task, "b1", "", state0, ex, send_index_in_state, payload_in_textbox)
|
|
return r, calls
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_opener_hop_full_success():
|
|
"""Prestage stopped on the profile: script opens the composer, fills,
|
|
sees it commit, finds the late Send, clicks, sees it clear -> receipt passes."""
|
|
r, calls = await run(TASK, PROFILE, [COMPOSER_EMPTY, COMPOSER_FILLED, COMPOSER_FILLED, COMPOSER_SENT])
|
|
assert r is not None and r["sent"] is True
|
|
assert r["payload"] == "[test] hello world r9-os"
|
|
# opener click (50), fill into composer (2 w/ text), solo send click (14)
|
|
idxs = [c[1].get("index") for c in calls["clicks"]]
|
|
assert 50 in idxs and 2 in idxs and 14 in idxs
|
|
fill = next(c for c in calls["clicks"] if c[1].get("text"))
|
|
assert fill[1]["text"] == "[test] hello world r9-os"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_composer_already_open_skips_opener():
|
|
"""Prestage left the composer open: no opener click, straight to fill+send."""
|
|
r, calls = await run(TASK, COMPOSER_EMPTY, [COMPOSER_FILLED, COMPOSER_FILLED, COMPOSER_SENT])
|
|
assert r is not None and r["sent"] is True
|
|
assert 50 not in [c[1].get("index") for c in calls["clicks"]] # never clicked an opener
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_abort_when_no_payload():
|
|
"""No quoted payload = the model's judgment call, never the script's."""
|
|
r, _ = await run("open tyler chen's linkedin and message him something nice", PROFILE,
|
|
[COMPOSER_EMPTY])
|
|
assert r is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_abort_when_fill_not_seen_committed():
|
|
"""Fill click ran but the textbox never shows the payload -> abort PRE-click,
|
|
the irreversible send never fires (no false 'sent')."""
|
|
r, calls = await run(TASK, COMPOSER_EMPTY, [COMPOSER_EMPTY, COMPOSER_EMPTY, COMPOSER_EMPTY])
|
|
assert r is None
|
|
# a Send-class click (index 14) must NEVER have been issued
|
|
assert all(c[1].get("index") != 14 for c in calls["clicks"])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_post_click_unverified_yields_honest_note_not_resend():
|
|
"""Send clicked but the composer never verifiably clears -> the run returns
|
|
sent=False with a do-not-resend note, never a silent retry."""
|
|
r, calls = await run(TASK, COMPOSER_EMPTY, [COMPOSER_FILLED, COMPOSER_FILLED,
|
|
COMPOSER_FILLED, COMPOSER_FILLED, COMPOSER_FILLED])
|
|
assert r is not None and r["sent"] is False
|
|
assert "do NOT send again" in r["note"] or "not send again" in r["note"].lower()
|
|
# exactly one send-class click was issued (no blind re-fire)
|
|
assert sum(1 for c in calls["clicks"] if c[1].get("index") == 14) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ambiguous_opener_aborts():
|
|
"""Two 'Message' openers = ambiguous = hands to the model, no guess."""
|
|
two = '[50]*<link "Message">\n[70]*<link "Message">'
|
|
r, _ = await run(TASK, two, [COMPOSER_EMPTY])
|
|
assert r is None
|