Test sessions offline
Replace provider calls with deterministic participants and verify replay without monkeypatching.
Protocol tests should fail because a contract changed, not because a provider timed out. Replace live calls with values and functions passed through the public participant seams:
| Test seam | Use |
|---|---|
ap.replies([...]) | A model-shaped sequence of known messages |
ap.replies(function) | An offer-aware model decision |
ap.Script | A role with a fixed authored sequence |
ap.Machine | A direct request-to-reply handler table |
ap.Policy | A pure rule over View and the current offer |
No environment override or global patch is required.
Fixed and offer-aware replies
import agentsparty as ap
Client, Solver = ap.roles('Client', 'Solver')
class Problem(ap.Message):
"""A problem to solve."""
text: str
class Solution(ap.Message):
"""The proposed solution."""
text: str
protocol = ap.Protocol(
ap.send(Client, Solver, Problem),
ap.send(Solver, Client, Solution),
)
fixed = ap.Session(
protocol,
{Solver: ap.Agent(ap.replies([Solution('42')]), '')},
).ask(Client, Problem('calculate'), expect=Solution)
dynamic = ap.Session(
protocol,
{
Solver: ap.Agent(
ap.replies(lambda offer: Solution('from offer')),
'',
)
},
).ask(Client, Problem('choose'), expect=Solution)Use a sequence when call order is part of the assertion. Use the function form for a branch or loop that must react to the classes currently on offer.
Pick the narrowest participant
Choose Script for a fixture, Machine for a handler contract, and Policy
for a history-based decision. Do not route every deterministic action through
an Agent; doing so hides which boundary the test exercises.
import agentsparty as ap
class Lookup(ap.Message):
"""A lookup request."""
key: str
class Found(ap.Message):
"""A lookup result."""
value: str
machine = ap.Machine({
Lookup: lambda request: Found(request.key),
})
script = ap.Script([Lookup('id')])
policy = ap.Policy(lambda view: Found('from view'))Verify replay
A journal can turn the first run into a replay fixture. Seed a second
MemoryJournal with recorded decisions and supply participants that have no
live answers left. The second run should finish by recalling the decisions.
import agentsparty as ap
User, Worker = ap.roles('User', 'Worker')
class Task(ap.Message):
"""The task."""
text: str
class Answer(ap.Message):
"""The answer."""
text: str
protocol = ap.Protocol(
ap.send(User, Worker, Task),
ap.send(Worker, User, Answer),
)
journal = ap.MemoryJournal()
ap.Session(
protocol,
{
User: ap.Script([Task('replay')]),
Worker: ap.Agent(ap.replies([Answer('ok')]), ''),
},
journal=journal,
).run()
replayed = ap.Session(
protocol,
{
User: ap.Script([]),
Worker: ap.Agent(ap.replies([]), ''),
},
journal=ap.MemoryJournal(journal.script().decisions),
).run()These doubles cover protocol assembly, projection, decoding, cast compatibility, limits, and replay. Keep a separate live integration check for provider authentication, latency, refusals, and real external systems.
Related
- Roles, participants, and models: choose a participant seam.
- Journals, replay, and tracing: understand recorded decisions.
- agentsparty.journal.memory.MemoryJournal: in-memory replay storage.