"""Python starter tests, mirroring the TypeScript A1 to A4. Offline: a fake facilitator and a mock registry."""
import base64
import json
import logging
from pathlib import Path

import httpx
import pytest
from eth_account import Account
from eth_account.messages import encode_typed_data
from fastapi.testclient import TestClient

from app import create_app
from evm_pay import exact_payment_header
from list_offer import ListOfferError, list_offer
from x402 import USDC, decode_header, encode_header, to_atomic_usdc

ROOT = Path(__file__).resolve().parents[1]
PAY_TO = "0x3c1A1E01e790Cd326495631B3b38Bd7068F08d6F"
PAYMENT = {"x402Version": 1, "scheme": "exact", "network": "base-sepolia",
           "payload": {"signature": "0xsig", "authorization": {"from": "0x" + "11" * 20}}}
FIVE_LAWS = ["Solidarity Oath", "Renewal Mandate", "Treasury Equilibrium", "Harm Shield", "Transparency Covenant"]


class FakeFacilitator:
    def __init__(self, valid=True, settles=True, verify_raises=False, settle_raises=False):
        self.calls, self.seen = [], []
        self.valid, self.settles, self.verify_raises, self.settle_raises = valid, settles, verify_raises, settle_raises

    async def verify(self, p, r):
        self.calls.append("verify"); self.seen.append((p, r))
        if self.verify_raises:
            raise RuntimeError("down")
        return {"isValid": True, "payer": "0x" + "11" * 20} if self.valid else {"isValid": False, "invalidReason": "invalid_signature"}

    async def settle(self, p, r):
        self.calls.append("settle")
        if self.settle_raises:
            raise RuntimeError("down")
        return {"success": True, "transaction": "0xabc123", "network": r["network"]} if self.settles \
            else {"success": False, "transaction": "", "network": r["network"], "errorReason": "insufficient_funds"}


def client(fac, network="base-sepolia", price="0.01", work=None):
    return TestClient(create_app(price=price, pay_to=PAY_TO, network=network, facilitator=fac, description="test call", work=work),
                      base_url="https://agent.example")


@pytest.mark.parametrize("network", ["base", "base-sepolia"])
def test_a1_unpaid_call_answers_402_with_requirements(network):
    r = client(FakeFacilitator(), network=network, price="0.25").get("/paid?x=1")
    assert r.status_code == 402
    b = r.json()
    assert b["x402Version"] == 1 and isinstance(b["error"], str) and len(b["accepts"]) == 1
    req = b["accepts"][0]
    assert req["scheme"] == "exact" and req["network"] == network
    assert req["maxAmountRequired"] == "250000"
    assert req["resource"] == "https://agent.example/paid"
    assert req["payTo"] == PAY_TO and req["mimeType"] == "application/json"
    assert req["asset"] == USDC[network]["address"]
    assert req["maxTimeoutSeconds"] > 0
    assert client(FakeFacilitator()).get("/health").json() == {"ok": True}


def test_a1_prices_and_bad_configuration():
    assert to_atomic_usdc("0.01") == "10000" and to_atomic_usdc("1") == "1000000" and to_atomic_usdc("0.000001") == "1"
    for bad in ["0", "-1", "0.0000001", "abc", "1e3"]:
        with pytest.raises(ValueError):
            to_atomic_usdc(bad)
    with pytest.raises(ValueError):
        create_app(price="0.01", pay_to="nope", network="base", facilitator=FakeFacilitator())


def test_a2_paid_call_verified_settled_then_200():
    fac = FakeFacilitator()

    async def work(_req):
        fac.calls.append("work")
        return {"ok": True, "answer": 42}

    r = client(fac, work=work).get("/paid", headers={"X-PAYMENT": encode_header(PAYMENT)})
    assert r.status_code == 200 and r.json() == {"ok": True, "answer": 42}
    assert fac.calls == ["verify", "settle", "work"]
    assert fac.seen[0][0] == PAYMENT and fac.seen[0][1]["maxAmountRequired"] == "10000"
    receipt = decode_header(r.headers["X-PAYMENT-RESPONSE"])
    assert receipt["success"] is True and receipt["transaction"] == "0xabc123" and receipt["network"] == "base-sepolia"


@pytest.mark.parametrize("kw,expected", [
    ({"valid": False}, ["verify"]), ({"verify_raises": True}, ["verify"]),
    ({"settles": False}, ["verify", "settle"]), ({"settle_raises": True}, ["verify", "settle"]),
])
def test_a2_failed_verify_or_settle_gets_402_and_no_work(kw, expected):
    fac = FakeFacilitator(**kw)
    worked = []

    async def work(_req):
        worked.append(1)
        return {}

    r = client(fac, work=work).get("/paid", headers={"X-PAYMENT": encode_header(PAYMENT)})
    assert r.status_code == 402 and "X-PAYMENT-RESPONSE" not in r.headers
    assert fac.calls == expected and worked == []


@pytest.mark.parametrize("header", ["%%%", encode_header({**PAYMENT, "network": "base"}), encode_header({**PAYMENT, "scheme": "upto"}), encode_header({"x402Version": 1})])
def test_a2_malformed_header_never_reaches_facilitator(header):
    fac = FakeFacilitator()
    assert client(fac).get("/paid", headers={"X-PAYMENT": header}).status_code == 402
    assert fac.calls == []


def test_a3_offer_passes_lax_rules_with_five_laws():
    o = json.loads((ROOT / "lax-offer.json").read_text())
    assert 3 <= len(o["name"]) <= 80 and 20 <= len(o["description"]) <= 600
    assert o["endpoint"].startswith("https://")
    assert o["agent_did"].startswith("did:")
    for law in FIVE_LAWS:
        assert law in o["constitutional_constraints"]
    cap = o["capabilities"][0]
    assert cap["pricing"] == {"model": "per_call", "amount": "0.01", "currency": "USDC", "rail": "x402"}
    assert set(o["payment_rails"]) <= {"x402", "usdc", "usdc-base"}
    assert "\u2014" not in json.dumps(o, ensure_ascii=False) and "\u2013" not in json.dumps(o, ensure_ascii=False)


def mock_registry(version):
    posts = []
    terms = {"x402Version": version, "error": "pay", "accepts": [{"scheme": "exact", "network": "eip155:8453" if version == 2 else "base",
             "amount": "10000", "maxAmountRequired": "10000", "payTo": "0xf62fb064f9d9d03b8013e5b6023e3501805ba973",
             "asset": USDC["base"]["address"], "maxTimeoutSeconds": 60, "extra": {"name": "USD Coin", "version": "2"}}]}

    def handler(request: httpx.Request):
        posts.append(request)
        if not (request.headers.get("PAYMENT-SIGNATURE") or request.headers.get("X-PAYMENT")):
            headers = {"PAYMENT-REQUIRED": encode_header(terms)} if version == 2 else {}
            return httpx.Response(402, json={"x402Version": 2, "error": "pay"} if version == 2 else terms, headers=headers)
        return httpx.Response(201, json={"ok": True, "id": "lax-ext-test123"})

    return httpx.Client(transport=httpx.MockTransport(handler)), posts, terms


@pytest.mark.parametrize("version,header", [(2, "PAYMENT-SIGNATURE"), (1, "X-PAYMENT")])
def test_a4_list_offer_pays_the_402_and_returns_the_id(version, header):
    c, posts, terms = mock_registry(version)
    offer = json.loads((ROOT / "lax-offer.json").read_text())
    seen = []
    out = list_offer(offer, lambda t: seen.append(t) or "cGF5bWVudA==", registry_url="https://reg.example/registry/", client=c)
    assert out == {"id": "lax-ext-test123"}
    assert len(posts) == 2 and json.loads(posts[1].content) == offer
    assert posts[1].headers[header] == "cGF5bWVudA=="
    assert seen[0]["accepts"] == terms["accepts"]


def test_a4_key_never_logged_and_payment_signature_is_valid(caplog, capsys):
    acct = Account.create()
    key_hex = acct.key.hex().removeprefix("0x")
    c, posts, _ = mock_registry(2)
    offer = json.loads((ROOT / "lax-offer.json").read_text())
    with caplog.at_level(logging.DEBUG):
        list_offer(offer, lambda t: exact_payment_header(t, acct, max_atomic=10000), client=c)
    out = capsys.readouterr()
    assert key_hex not in out.out + out.err + caplog.text
    hdr = posts[1].headers["PAYMENT-SIGNATURE"]
    assert key_hex not in hdr
    p = json.loads(base64.b64decode(hdr))
    a = p["payload"]["authorization"]
    typed = {"types": {"EIP712Domain": [{"name": "name", "type": "string"}, {"name": "version", "type": "string"}, {"name": "chainId", "type": "uint256"}, {"name": "verifyingContract", "type": "address"}],
                       "TransferWithAuthorization": [{"name": n, "type": t} for n, t in [("from", "address"), ("to", "address"), ("value", "uint256"), ("validAfter", "uint256"), ("validBefore", "uint256"), ("nonce", "bytes32")]]},
             "primaryType": "TransferWithAuthorization",
             "domain": {"name": "USD Coin", "version": "2", "chainId": 8453, "verifyingContract": USDC["base"]["address"]},
             "message": {"from": a["from"], "to": a["to"], "value": int(a["value"]), "validAfter": int(a["validAfter"]), "validBefore": int(a["validBefore"]), "nonce": bytes.fromhex(a["nonce"][2:])}}
    assert Account.recover_message(encode_typed_data(full_message=typed), signature=p["payload"]["signature"]) == acct.address
    c2, _, terms = mock_registry(2)
    with pytest.raises(ValueError, match="above the cap"):
        exact_payment_header(terms, acct, max_atomic=9999)


def test_a4_registry_error_is_raised():
    c = httpx.Client(transport=httpx.MockTransport(lambda r: httpx.Response(400, json={"error": "name must be at least 3 characters"})))
    with pytest.raises(ListOfferError, match="at least 3"):
        list_offer({"name": "x"}, lambda t: "x", client=c)


def test_a5_readme_holds_base_sepolia_proof():
    import re
    readme = (Path(__file__).resolve().parent.parent / "README.md").read_text()
    assert re.search(r"Base Sepolia proof[^\n]*\n[\s\S]*?https://sepolia\.basescan\.org/tx/0x[0-9a-fA-F]{64}", readme)

agentgate-source: code mirror on Cloudflare Pages. All repos