Spaces:
Running
Running
| """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 | |