agent-market/tests/e2e_http.py

#!/usr/bin/env python3
"""End-to-end HTTP checks against a running Agent Market. Usage: e2e_http.py http://127.0.0.1:8795
Typed messages always go to a model (hosted free models, then the local model). With REQUIRE_LLM=1 they must be answered
by one; otherwise an honest busy reply is accepted for them. Clicks, forms and payments are deterministic and always checked."""
import json, os, sys, urllib.request, urllib.error

BASE = (sys.argv[1] if len(sys.argv) > 1 else "http://127.0.0.1:8795").rstrip("/")
results = []


def call(method, path, body=None, headers=None, raw=False):
    data = json.dumps(body).encode() if body is not None else None
    h = {"Content-Type": "application/json", **(headers or {})}
    req = urllib.request.Request(BASE + path, data=data, method=method, headers=h)
    try:
        with urllib.request.urlopen(req, timeout=90) as r:
            b = r.read().decode()
            return r.status, (b if raw else json.loads(b))
    except urllib.error.HTTPError as e:
        b = e.read().decode()
        try:
            return e.code, json.loads(b)
        except Exception:
            return e.code, b


def check(name, cond, detail=""):
    results.append((name, bool(cond)))
    print(("PASS " if cond else "FAIL ") + name + (f"  [{detail}]" if detail else ""))


s, html = call("GET", "/", raw=True)
import re
check("landing page loads (React SPA shell)", s == 200 and "Agent Market" in html and '<div id="root">' in html)
assets = re.findall(r'(?:src|href)="(/static/dist/assets/[^"]+)"', html)
js = [a for a in assets if a.endswith(".js")]
bundle = call("GET", js[0], raw=True) if js else (0, "")
check("built JS bundle served", bundle[0] == 200 and "chatLauncher" in bundle[1] and "show_checkout" in bundle[1], f"assets={len(assets)}")
s, st = call("GET", "/api/status")
check("status endpoint", s == 200 and st["payments"]["mode"] == "test", f"agent={st['agent']['mode']} fee_bps={st['fee_bps']}")
fee_bps = st["fee_bps"]

REQUIRE_LLM = os.environ.get("REQUIRE_LLM") == "1"
BUSY = "The assistant is busy, please try again in a moment."


def typed(name, out, ok_fn, detail=""):
    """A typed message: answered by a model (then ok_fn must hold) or, unless REQUIRE_LLM, the honest busy reply."""
    if out.get("agent") == "llm":
        check(name + f" (model {out.get('model')}{', local' if out.get('local') else ''})", ok_fn(out), detail)
    else:
        check(name + " -> busy reply (no model available)", not REQUIRE_LLM and out.get("agent") == "unavailable" and
              out.get("reply") == BUSY and out.get("actions") == [] and out.get("retry") is True, f"agent={out.get('agent')}")


s, out = call("POST", "/api/chat", {"event": {"type": "landing"}})
check("landing view (deterministic featured listings)", s == 200 and out["agent"] == "ui-event" and out["actions"][0]["type"] == "show_cards"
      and len(out["actions"][0]["payload"]["listings"]) >= 3)
s, out = call("POST", "/api/chat", {"message": "show me bikes"})
cards = [a for a in out["actions"] if a["type"] == "show_cards"]
titles = [l["title"] for l in cards[0]["payload"]["listings"]] if cards else []
typed("typed search renders hydrated cards", out, lambda o: cards and any("bike" in t.lower() for t in titles), ", ".join(titles))
s, r = call("POST", "/api/actions", {"action": "search", "args": {"q": "bike"}}, {"X-Agent-Id": "test-buyer-e2e"})
bike = next((l for l in r.get("listings", []) if l["kind"] == "product" and "online" in l["payment_options"] and l["status"] == "active"), None)
if not bike:  # earlier runs may have sold the seed bike; list a fresh test-only one
    s, r = call("POST", "/api/actions", {"action": "create_listing", "args": {"title": "Test-only commuter bike", "kind": "product", "price": 18000,
                "description": "Fake bike for automated tests", "payment_options": ["online", "cash"]}}, {"X-Agent-Id": "test-seller-e2e"})
    bike = r["listing"]

s, out = call("POST", "/api/chat", {"message": "I want to sell my old desk lamp for $20, cash only"})
typed("typed sell request opens a prefilled listing form", out, lambda o: any(a["type"] == "show_listing_form" for a in o["actions"]))
s, out = call("POST", "/api/chat", {"event": {"type": "sell"}})
check("Sell button opens the listing form (deterministic)", out["agent"] == "ui-event" and out["actions"][0]["type"] == "show_listing_form")
s, out = call("POST", "/api/chat", {"event": {"type": "submit_listing", "fields": {"title": "Test-only desk lamp", "kind": "product", "price": "20", "description": "Fake test listing", "payment_options": ["online", "cash"]}}, "actor": "guest-e2e"})
new = [a for a in out["actions"] if a["type"] == "show_cards"]
newl = new[0]["payload"]["listings"][0] if new else {}
check("creating a listing via the agent works", newl.get("title") == "Test-only desk lamp" and newl.get("price") == 2000 and newl.get("seller") == "guest-e2e")

# Online purchase through the agent: checkout panel shows fee, test card confirms.
s, out = call("POST", "/api/chat", {"event": {"type": "buy_online", "listing_id": bike["id"]}, "actor": "guest-e2e"})
co = [a for a in out["actions"] if a["type"] == "show_checkout"]
it = co[0]["payload"]["intent"] if co else {}
exp_fee = round(bike["price"] * fee_bps / 10000)
check("checkout panel shows fee", it.get("fee_cents") == exp_fee and it.get("total_cents") == bike["price"] + exp_fee and it.get("mode") == "test",
      f"price={bike['price']} fee={it.get('fee_cents')} total={it.get('total_cents')}")
s, r = call("POST", f"/api/payments/{it['id']}/confirm", {"card_number": "1234 5678 9012 3456"})
check("non-test card refused", r.get("status") == "failed" and r.get("error") == "not_a_test_card")
s, r = call("POST", f"/api/payments/{it['id']}/confirm", {"card_number": "4000 0000 0000 0002"})
check("decline test card declines", r.get("status") == "failed" and r.get("error") == "card_declined")
s, r = call("POST", f"/api/payments/{it['id']}/confirm", {"card_number": "4242 4242 4242 4242"})
order = r.get("order") or {}
check("test card 4242 confirms via signed webhook", r.get("status") == "succeeded" and order.get("status") == "paid" and order.get("fee_cents") == exp_fee,
      f"order={order.get('id')} fee={order.get('fee_cents')} last4={order.get('card_last4')}")
s, out = call("POST", "/api/chat", {"event": {"type": "payment_result", "intent_id": it["id"], "status": "succeeded", "order_id": order.get("id")}})
check("agent renders receipt after payment", any(a["type"] == "show_receipt" for a in out["actions"]))
s, l = call("GET", f"/api/listings/{bike['id']}")
check("sold product marked sold", l["listing"]["status"] == "sold")
s, r = call("POST", "/api/payments/webhook", {"type": "payment_intent.succeeded", "data": {"intent_id": it["id"]}}, {"X-Mock-Signature": "forged"})
check("forged webhook rejected", s == 401)

# Cash in person: zero fee.
s, r = call("POST", "/api/actions", {"action": "search", "args": {"q": "lamp"}}, {"X-Agent-Id": "test-buyer-e2e"})
lamps = [l for l in r.get("listings", []) if "cash" in l["payment_options"] and l["kind"] == "product" and l["status"] == "active"]
if not lamps:
    s, r = call("POST", "/api/actions", {"action": "create_listing", "args": {"title": "Test-only lamp", "kind": "product", "price": 3000,
                "description": "Fake lamp for automated tests", "payment_options": ["online", "cash"]}}, {"X-Agent-Id": "test-seller-e2e"})
    lamps = [r["listing"]]
lamp = lamps[0]
s, out = call("POST", "/api/chat", {"event": {"type": "buy_cash", "listing_id": lamp["id"]}, "actor": "guest-e2e"})
cm = [a for a in out["actions"] if a["type"] == "show_cash_meetup"]
deal = cm[0]["payload"]["deal"] if cm else {}
check("cash meetup panel, fee $0", deal.get("fee_cents") == 0 and deal.get("status") == "arranged")
s, out = call("POST", "/api/chat", {"event": {"type": "cash_completed", "deal_id": deal["id"]}})
rc = [a for a in out["actions"] if a["type"] == "show_receipt"]
check("cash deal completion records no fee", rc and rc[0]["payload"]["order"]["fee_cents"] == 0 and rc[0]["payload"]["order"]["method"] == "cash_in_person")

# Agent API
s, man = call("GET", "/agent.json")
check("agent.json manifest", s == 200 and "start_checkout" in man["actions"] and man["test_mode"] is True)
s, oa = call("GET", "/openapi.json")
check("openapi.json", s == 200 and oa["openapi"].startswith("3.") and "/api/actions" in oa["paths"])
A = {"X-Agent-Id": "agent:e2e-seller-bot"}
B = {"X-Agent-Id": "agent:e2e-buyer-bot"}
s, r = call("POST", "/api/actions", {"action": "create_listing", "args": {"title": "Test-only data labelling (agent service)", "kind": "service", "price": 1500, "description": "Fake service listed by a test agent"}}, A)
al = r.get("listing", {})
check("agent API: agent lists a service", s == 201 and al.get("seller") == "agent:e2e-seller-bot" and al.get("seller_type") == "agent")
s, r = call("POST", "/api/actions", {"action": "search", "args": {"q": "labelling", "limit": 50}}, B)   # staging accumulates test listings
check("agent API: search finds it", any(x["id"] == al.get("id") for x in r.get("listings", [])))
s, r = call("POST", "/api/actions", {"action": "quote", "args": {"listing_id": al["id"]}}, B)
check("agent API: quote online vs cash", r["online"]["fee_cents"] == round(1500 * fee_bps / 10000) and r["cash_in_person"]["fee_cents"] == 0)
s, r = call("POST", "/api/actions", {"action": "start_checkout", "args": {"listing_id": al["id"]}}, B)
pi = r.get("intent", {})
s, r = call("POST", "/api/actions", {"action": "confirm_payment", "args": {"intent_id": pi.get("id"), "card_number": "4242424242424242"}}, B)
check("agent API: agent buys with test card", r.get("status") == "succeeded" and r["order"]["buyer"] == "agent:e2e-buyer-bot" and r["order"]["seller"] == "agent:e2e-seller-bot",
      f"total={r.get('order', {}).get('total_cents')} fee={r.get('order', {}).get('fee_cents')}")
s, r = call("POST", "/api/actions", {"action": "list_orders"}, B)
check("agent API: buyer sees its order", any(o["listing_id"] == al["id"] for o in r.get("orders", [])))
s, r = call("POST", "/api/actions", {"action": "start_cash_deal", "args": {"listing_id": lamp["id"]}}, B)
check("agent API: sold product cannot be bought again", s == 400 and r.get("error") == "listing_not_active")

passed = sum(1 for _, ok in results if ok)
print(f"\n{passed}/{len(results)} passed")
sys.exit(0 if passed == len(results) else 1)