Metadata-Version: 2.5
Name: uagent-testkit
Version: 0.1.2
Summary: Test Fetch.ai uAgents offline — handlers, agent-to-agent flows, and BNB Chain reads.
Project-URL: Homepage, https://github.com/pgalyen1987/uagent-testkit
Project-URL: Repository, https://github.com/pgalyen1987/uagent-testkit
Project-URL: Issues, https://github.com/pgalyen1987/uagent-testkit/issues
Project-URL: Changelog, https://github.com/pgalyen1987/uagent-testkit/blob/main/CHANGELOG.md
Author-email: Patrick Galyen <pgalyen87@gmail.com>
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: agent-launch,agentverse,asi,bnb-chain,bsc,fetchai,pytest,testing,uagents,web3
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: Pytest
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Testing
Requires-Python: >=3.10
Requires-Dist: uagents>=0.23
Provides-Extra: test
Requires-Dist: aiohttp>=3.9; extra == 'test'
Requires-Dist: httpx>=0.27; extra == 'test'
Requires-Dist: pytest-asyncio>=0.24; extra == 'test'
Requires-Dist: pytest>=8.0; extra == 'test'
Requires-Dist: requests>=2.31; extra == 'test'
Requires-Dist: web3>=7; extra == 'test'
Description-Content-Type: text/markdown

# uagent-testkit

[![PyPI](https://img.shields.io/pypi/v/uagent-testkit.svg)](https://pypi.org/project/uagent-testkit/)
[![CI](https://github.com/pgalyen1987/uagent-testkit/actions/workflows/ci.yml/badge.svg)](https://github.com/pgalyen1987/uagent-testkit/actions/workflows/ci.yml)

Test [Fetch.ai uAgents](https://github.com/fetchai/uAgents) without a network, a wallet, or a `sleep()`.

```python
from uagent_testkit import harness

async def test_echo():
    h = harness(my_agent)
    result = await h.deliver(Ping(text="hi"))

    assert result.reply(Pong).text == "pong:hi"
    assert h.storage["seen"] == 1
```

## Why

The uAgents framework has no testing story. To exercise a handler today you generally
have to stand the agent up for real — which means Almanac registration, endpoint
resolution, a funded wallet on some paths, and `asyncio.sleep()` calls sprinkled through
the test to wait for messages that may never arrive. That is slow, flaky, and awkward to
run in CI.

There is also a subtler problem. `Agent._handle_message` catches every exception a
handler raises and writes it to a log. That is the right call at runtime — an agent
processes messages from peers it does not control, and one malformed message must not
take the process down. But the same containment applies under pytest, and there is no
strict mode to turn it off, so a handler that crashes on every message still looks
like a green test.

`uagent-testkit` routes messages straight to the registered handler with a context that
records instead of transmitting. No sockets, no registration, no waiting.

- **Handler exceptions are raised** rather than contained into a log.
- **Storage is in-memory**, so state doesn't leak between tests via the JSON file the
  stock `KeyValueStore` writes to your working directory.
- **Reply contracts are checked** — if `@on_message(replies=Pong)` sends neither a
  `Pong` nor an `ErrorMessage`, `assert_reply_contract()` fails, and the error the
  framework would log lands in `result.logs`, where `assert_no_errors()` sees it.
- **Interval and lifecycle handlers run on demand**, so an `@on_interval(period=3600)`
  is testable without waiting an hour.
- **Multi-agent conversations replay deterministically**, with a loop guard instead of
  a hung test.
- **Agents that read BNB Chain are testable offline** — `ChainDouble` answers BEP-20
  and balance queries from state your test sets, and blocks any other request made
  through requests, httpx, aiohttp or urllib.

## Install

```bash
pip install uagent-testkit
```

Requires Python 3.10+ and `uagents>=0.23`. The pytest fixtures register automatically.

## Testing one agent

```python
from uagents import Agent, Context, Model
from uagent_testkit import harness

class Ping(Model):
    text: str

class Pong(Model):
    text: str

agent = Agent(name="echo", seed="echo seed")

@agent.on_message(model=Ping, replies=Pong)
async def on_ping(ctx: Context, sender: str, msg: Ping):
    ctx.storage.set("seen", (ctx.storage.get("seen") or 0) + 1)
    await ctx.send(sender, Pong(text=f"pong:{msg.text}"))
```

```python
async def test_ping():
    h = harness(agent)
    result = await h.deliver(Ping(text="hi"), sender="agent1qexample")

    result.assert_replied_with(Pong)
    result.assert_reply_contract()          # honoured its replies= declaration
    assert result.reply(Pong).text == "pong:hi"
    assert result.sent[0].destination == "agent1qexample"
    assert h.storage["seen"] == 1
```

### Asserting on failure

```python
async def test_bad_input_is_handled():
    result = await h.deliver(Ping(text=""), raise_errors=False)
    assert isinstance(result.error, ValueError)
    assert "empty" in str(result.error)
```

By default `deliver()` re-raises whatever the handler raised, so a broken handler fails
the test loudly. Pass `raise_errors=False` when the failure is the thing under test.
Cancellation and `KeyboardInterrupt` are not handler failures, so they always propagate.

### Sender verification

Handlers without `allow_unverified=True` are only reachable from a verified agent
address. The harness enforces that, so a test can't pass for a flow production
rejects:

```python
from uagent_testkit import UnverifiedSender, user_sender

async def test_strict_handler_refuses_users():
    with pytest.raises(UnverifiedSender):
        await h.deliver(Ping(text="hi"), sender=user_sender())
```

### Intervals and lifecycle

```python
await h.startup()                  # runs @on_event("startup") handlers
result = await h.tick()            # runs @on_interval handlers once, ignoring period
                                   # (and @on_schedule ones, on uagents 0.26+)
result = await h.tick(only="heartbeat")   # just one, by function name
await h.shutdown()
```

## Testing agents against each other

```python
from uagent_testkit import AgentNetwork

async def test_conversation():
    net = AgentNetwork(alice, bob)
    transcript = await net.send(Ping(text="hello"), to=bob, sender=alice)

    transcript.assert_delivered(Pong, to=alice.address)
    assert transcript.of(Pong)[0].text == "pong:hello"
    assert net.harness(alice).storage["last_pong"] == "pong:hello"
```

Every reply addressed to another agent in the network is delivered in turn until the
conversation goes quiet. As on a live network, a reply carries the session of the message
it answers, so the whole conversation runs in one `ctx.session` (pass `session=` to choose
it). Sending to an agent you never added raises `KeyError`. Messages a handler sends to
addresses outside the network are collected on `transcript.undelivered` rather than
silently dropped:

```python
transcript.assert_all_delivered()   # fails if anything went to an unknown address
```

Agents that answer each other forever raise `ConversationTooLong` after `max_rounds`
deliveries instead of hanging the suite. This catches echo loops, which are easy to
write and expensive to discover in production:

```python
# a keyword responder whose answer contains its own trigger word
if "menu" in msg.text().lower():
    await ctx.send(sender, chat("here is the menu"))   # -> contains "menu"
```

Two agents like that will talk to each other indefinitely. On a live network that is
two agents spamming each other; here it fails in milliseconds with the transcript
so far attached.

## Protocols and the chat protocol

Agents assembled from `Protocol` objects work the same way, including the shared chat
protocol from `uagents_core` that Agentverse and Agent Launch agents use:

```python
chat = Protocol(spec=chat_protocol_spec)

@chat.on_message(ChatMessage)
async def on_chat(ctx: Context, sender: str, msg: ChatMessage):
    ...

agent.include(chat)
```

```python
async def test_chat():
    h = harness(agent)
    result = await h.deliver(
        ChatMessage(
            timestamp=datetime.now(timezone.utc),
            msg_id=uuid4(),
            content=[TextContent(type="text", text="show me the menu")],
        )
    )
    assert result.replies(ChatAcknowledgement)
    assert result.replies(ChatMessage)[0].text() == "here is what we have"
```

## Agents that read BNB Chain

The harness intercepts `ctx.send`, which covers agent-to-agent messaging. It does not
cover a handler that reaches out to an RPC endpoint — an agent reading a BEP-20 balance
still talks to the network during a test. `ChainDouble` closes that gap:

```python
from uagent_testkit import ChainDouble, FET_BSC, harness

async def test_treasury_agent():
    chain = ChainDouble()
    chain.add_token(FET_BSC, symbol="FET", decimals=18,
                    balances={wallet: 5_500_000_000_000_000_000})

    h = harness(treasury_agent)
    with chain.install():
        result = await h.deliver(CheckBalance(wallet=wallet))

    assert result.reply(BalanceReport).amount == 5.5
    chain.assert_called("eth_call")
```

It patches `requests`, `httpx` and `aiohttp` where they are importable, plus `urllib`,
at the call every request through them passes, so it works whether the agent uses
web3.py or raw HTTP; the test suite drives web3.py's `HTTPProvider` and
`AsyncHTTPProvider` through it. It understands web3.py's JSON-encoded `data=` body as
well as `json=`. It does not cover `http.client` or `urllib3` used directly, or sockets
a handler opens itself.

Supported: `eth_call` (BEP-20 `balanceOf`, `symbol`, `name`, `decimals`, `totalSupply`),
`eth_getBalance`, `eth_getCode`, `eth_blockNumber`, `eth_chainId`, `net_version`,
`eth_getTransactionCount`, batched requests. Two of these answer without being told:
`eth_getCode` returns `0x` (no code) for any address not registered with `add_token()`,
and `eth_getTransactionCount` always returns 0.

Two deliberate refusals, both of which turn a silent pass into a loud failure:

```python
# reading state the test never set up
with pytest.raises(UnstubbedCall):
    ...                       # rather than quietly returning zero

# any other HTTP from inside a handler: a price API, an LLM, a webhook
with pytest.raises(NetworkCallBlocked):
    ...                       # rather than hitting a real API in CI
```

Agents wrap RPC reads in `try/except` so one bad call can't crash them, and that habit
would swallow a refusal and turn it back into a silent pass. So `deliver()` (and
`tick()`, `startup()`, `shutdown()`) re-raises any refusal the handler's own work made,
caught or not: in the handler, in the tasks it starts, and in `asyncio.to_thread()`.
Work the handler did not start is not charged to it, such as a delivery running
alongside it or the manifest upload that `include(proto, publish_manifest=True)` queues.
A refusal from a thread that carries no context (`loop.run_in_executor()`, a plain
`Thread`) fails every delivery running at that moment. `chain.refusals` keeps every
refusal either way.

There's also a `bsc` fixture that gives you an installed double directly.

## pytest fixtures

Installing the package registers two factory fixtures:

```python
def test_with_fixture(agent_harness, agent_network):
    h = agent_harness(my_agent)
    net = agent_network(alice, bob)
```

## API

| | |
|---|---|
| `harness(agent)` | Wrap an agent. Swaps in in-memory storage and merges the internal protocol. |
| `h.deliver(msg, sender=…, raise_errors=True)` | Route a message to its handler. Returns `Delivery`. |
| `h.tick(only=None)` | Run `@on_interval` (and `@on_schedule`) handlers once. |
| `h.startup()` / `h.shutdown()` | Run `@on_event` handlers. |
| `h.storage` | In-memory store, subscriptable. |
| `h.handles(Model)` | Whether a handler is registered for a type. |
| `Delivery.sent` | List of `Sent(destination, schema_digest, body)`. |
| `Delivery.reply(Model)` | The single reply of that type; raises otherwise. |
| `Delivery.replies(Model, to=…)` | Replies of that type to the sender; `to=ANY` for any destination. |
| `user_sender()` | A user (non-agent) address, for `allow_unverified` paths. |
| `Delivery.error` / `.logs` | Handler exception (or a refusal it caught) and captured log records. |
| `Delivery.assert_replied_with` / `assert_silent` / `assert_no_errors` / `assert_reply_contract` | Chainable assertions. |
| `AgentNetwork(*agents)` | Multi-agent network. |
| `net.send(msg, to=…, sender=…, max_rounds=20, session=None)` | Play out a conversation in one session. Returns `Transcript`. |
| `Transcript.of(Model)` / `.to(addr)` / `.undelivered` | Inspect what was exchanged. |

## Scope

This exercises your handler logic: routing, replies, storage, protocol contracts, and
inter-agent flow. It deliberately does not simulate Almanac resolution, envelope
signing, or on-chain settlement — those are integration concerns and mocking them would
give you false confidence. Keep a thin integration test on a real testnet for those, and
use this for everything else.

`ctx.send_and_receive()` is not supported. Sends are recorded rather than delivered, so
no answer can reach the handler that is waiting for one; the harness raises
`NotImplementedError` rather than hand it `None`, and a handler that catches it still
fails its delivery.

The harness can't stop work an agent schedules on its own event loop.
`include(proto, publish_manifest=True)` queues an upload of the protocol's manifest to
Agentverse, and it runs whenever that loop runs, with or without the harness: for an
agent built inside an async test, as soon as the test yields to the loop. Pass
`publish_manifest=False` when you build agents for tests.

The [changelog](https://github.com/pgalyen1987/uagent-testkit/blob/main/CHANGELOG.md)
lists what each release fixed, and which fixes can fail a test that used to pass.

## Running the tests

```bash
python -m venv .venv && .venv/bin/pip install -e ".[test]"
.venv/bin/python -m pytest
```

## License

Apache-2.0
