Spaces:
Running
Running
File size: 3,121 Bytes
a1571b3 8f07721 a1571b3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | """Regression tests for the universal BigMesh Operator."""
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from core import operator
def _amazon_runtime(*, browser=False, session=False, cogs=False):
ops = []
if browser:
ops.append({
"coord": "submesh.bigmesh_e2e.browser.open",
"summary": "Open a visible browser URL",
"method": "EXEC",
})
return {
"sectors": [{
"sector": "amazon_seller",
"capability": {"requires": {
"meshd": {"met": True},
"amazon_session": {"met": session},
"seller_cogs": {"met": cogs},
}},
}],
"devices": [{"connected": True, "advertised_ops": ops}],
}
def test_poorly_phrased_amazon_profit_request_inserts_auth_first():
plan = operator.resolve(
"amazon says im making money but i dont think i actually am",
runtime=_amazon_runtime(browser=False),
)
assert plan["understood"]["id"] == "seller_profit_truth"
assert plan["plan"][0]["action"]["id"] == "amazon.session.connect"
assert plan["plan"][0]["prerequisite"] is True
assert plan["plan"][0]["action"]["status"] == "blocked"
assert "browser" in plan["plan"][0]["action"]["blocked_reason"].lower()
def test_existing_edge_browser_makes_seller_connect_action_real():
plan = operator.resolve(
"just get me into seller central",
runtime=_amazon_runtime(browser=True),
)
action = plan["plan"][0]["action"]
assert action["id"] == "amazon.session.connect"
assert action["status"] == "ready"
assert action["request"]["url"] == "/api/submesh/call"
assert action["request"]["body"]["coord"] == "submesh.bigmesh_e2e.browser.open"
assert action["request"]["body"]["input"] == {
"url": "https://sellercentral.amazon.com/"
}
def test_broken_site_language_resolves_to_continuity_goal():
plan = operator.resolve(
"the damn website keeps having busted docman links can you make it stop",
runtime=_amazon_runtime(),
)
assert plan["understood"]["id"] == "site_continuity"
assert any(s["action"]["id"] == "site.continuity.inspect" for s in plan["plan"])
def test_context_disambiguates_short_follow_up():
plan = operator.resolve(
"fix it",
context={"surface": "atlas", "sector": "amazon_seller",
"selected": "sector.amazon_seller.inventory"},
runtime=_amazon_runtime(),
)
assert plan["understood"]["id"].startswith("seller_")
def test_html_response_gets_operator_loader():
r = HTMLResponse("<html><body><h1>BigMesh</h1></body></html>")
body = r.body.decode("utf-8")
assert "data-signalmesh-operator" in body
assert "/ui/operator/widget.js" in body
def test_signalmesh_fastapi_gets_operator_routes_once():
app = FastAPI(title="SignalMesh test")
paths = [r.path for r in app.routes]
assert paths.count("/api/operator/resolve") == 1
assert paths.count("/api/operator/actions") == 1
assert paths.count("/ui/operator/widget.js") == 1
|