mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-27 20:14:49 +02:00
[eric] browser: unit and end-to-end loop tests for the new browser behaviors
This commit is contained in:
@@ -0,0 +1,64 @@
|
||||
"""Aux-LLM stuck-adjudication for the browser sub-agent."""
|
||||
|
||||
import asyncio
|
||||
|
||||
from backend.apps.agents.browser.browser_validator import adjudicate_stuck, _extract_text
|
||||
|
||||
|
||||
class _Block:
|
||||
def __init__(self, type_, text=""):
|
||||
self.type = type_
|
||||
self.text = text
|
||||
|
||||
|
||||
class _Resp:
|
||||
def __init__(self, blocks):
|
||||
self.content = blocks
|
||||
|
||||
|
||||
class _FakeClient:
|
||||
"""Minimal Anthropic-shaped client: client.messages.create(...)."""
|
||||
|
||||
def __init__(self, resp=None, raise_exc=None):
|
||||
self._resp = resp
|
||||
self._raise = raise_exc
|
||||
self.calls = []
|
||||
self.messages = self
|
||||
|
||||
async def create(self, **kwargs):
|
||||
self.calls.append(kwargs)
|
||||
if self._raise:
|
||||
raise self._raise
|
||||
return self._resp
|
||||
|
||||
|
||||
def test_returns_extracted_guidance_and_assembles_prompt():
|
||||
fc = _FakeClient(resp=_Resp([_Block("text", "Press Tab then Enter to focus the field.")]))
|
||||
out = asyncio.run(adjudicate_stuck(fc, "cheap-model", "share the doc", "- click -> not found", "the page"))
|
||||
assert out == "Press Tab then Enter to focus the field."
|
||||
call = fc.calls[0]
|
||||
assert call["model"] == "cheap-model"
|
||||
assert call["max_tokens"] == 300
|
||||
prompt = call["messages"][0]["content"]
|
||||
assert "share the doc" in prompt
|
||||
assert "not found" in prompt
|
||||
|
||||
|
||||
def test_swallows_provider_error_and_returns_empty():
|
||||
fc = _FakeClient(raise_exc=RuntimeError("429 rate limited"))
|
||||
out = asyncio.run(adjudicate_stuck(fc, "m", "g", "r", "p"))
|
||||
assert out == ""
|
||||
|
||||
|
||||
def test_extract_text_joins_text_blocks_and_ignores_others():
|
||||
resp = _Resp([_Block("text", "First."), _Block("tool_use"), _Block("text", "Second.")])
|
||||
assert _extract_text(resp) == "First. Second."
|
||||
|
||||
|
||||
def test_handles_empty_inputs_without_crashing():
|
||||
fc = _FakeClient(resp=_Resp([_Block("text", "ok")]))
|
||||
out = asyncio.run(adjudicate_stuck(fc, "m", "", "", ""))
|
||||
assert out == "ok"
|
||||
# placeholders keep the prompt well-formed
|
||||
prompt = fc.calls[0]["messages"][0]["content"]
|
||||
assert "(unknown)" in prompt and "(none)" in prompt and "(empty)" in prompt
|
||||
Reference in New Issue
Block a user