soev_agents.testing
Check that an agent keeps the determinism and no-hidden-state rules, and that a store keeps its contract.
The checks run an agent with scripted doubles in place of the outside world. The resume-equivalence check stops a
turn after each step and resumes it in a new process. The determinism check rebuilds each model call of a recorded
thread under several server configurations. The HTTP conformance suite is in soev_agents.testing.http.
Modules
Scripted doubles
A scripted double plays a fixed script in place of a model, a tool, an effector or a step, and keeps each call.
ScriptedChatModel
classScriptedChatModel(outputs, fragments=(), *, name='scripted')A chat model that gives the outputs of its script in order, one for each call, and keeps each call.
complete and stream_complete both take the next output of the script. A streamed call yields every fragment
of fragments first, the same fragments for each call, and then the output. An output without a model gets
the name of the model that the call asked for, or name when the call asked for none, as a real adapter names
the model that replied.
Example
from soev_agents.domain import ModelOutput
from soev_agents.ports import ChatRequest, UserMessage
from soev_agents.testing import ScriptedChatModel
model = ScriptedChatModel([ModelOutput(content="Paris.")])
output = await model.complete(ChatRequest(messages=(UserMessage(content="Capital of France?"),)))
assert output.content == "Paris." and output.model == "scripted"Make a chat model that plays outputs.
paramoutputsSequence[ModelOutput]The output of each call, in call order.
paramfragmentsSequence[Fragment]= ()What each streamed call yields before its output.
paramnamestr= 'scripted'The model that a call without a model goes to.
attributerequestslist[ChatRequest]The ChatRequest of each call, in call order.
attributestreamedlist[bool]For each call, whether it came through stream_complete.
attributemodelslist[str | None]The model that each call asked for, None for the adapter's own.
methodresolve(model) -> strReturn the name of the model that a call that asks for model goes to.
parammodelstr | NoneThe model that the call asks for, or None for the adapter's own.
Returns
strmodel, or name when model is None or empty.
methodcompleteasync (request, *, model=None) -> ModelOutputKeep the call and return the next output of the script.
paramrequestChatRequestThe ChatRequest of the call. The model keeps it in requests.
parammodelstr | None= NoneThe model that the call asks for, or None for the adapter's own.
Returns
ModelOutputThe next output, with the name of the called model when the script gives no model.
Raises:
LookupError: Every output of the script was played already.
methodstream_completeasync (request, *, model=None) -> AsyncIterator[Fragment | ModelOutput]Keep the call, then yield the fragments and the next output of the script.
paramrequestChatRequestThe ChatRequest of the call. The model keeps it in requests.
parammodelstr | None= NoneThe model that the call asks for, or None for the adapter's own.
Yields:
Fragment | ModelOutput: Each fragment offragments, then the next output, ascompletereturns it.
Raises:
LookupError: Every output of the script was played already. The first item of the iteration raises it.
ScriptedTool
dataclassScriptedTool(name, results, *, description='', arguments=…)A tool that gives its calls the results of its script in order, and keeps the arguments of each call.
It raises a result that is an exception. It keeps each call's parsed arguments as a dict in calls.
It is a frozen dataclass that holds no mutable field, so ToolStep accepts it.
Make a tool that plays results.
paramnamestrThe name that the model calls the tool by.
paramresultsSequence[Exception | ToolOutputBase]The result of each call, in call order. The tool raises a result that is an exception.
paramdescriptionstr= ''What the model reads about the tool.
paramargumentstype[BaseModel]= …The pydantic model of the arguments. By default it accepts arguments of any shape.
attributecallslist[Mapping[str, Any]]The parsed arguments of each call that the tool ran, as dicts, in call order.
methodrunasync (arguments) -> ToolOutputBaseKeep the call's arguments and return the next result of the script.
paramargumentsBaseModelThe parsed arguments of the call.
Returns
ToolOutputBaseThe next result of the script.
Raises:
LookupError: Every result of the script was played already.Exception: The next result of the script is this exception.
ScriptedEffector
classScriptedEffector(results)An effector that returns the results of its script in order, one for each intent, and keeps each intent.
Make an effector that plays results.
paramresultsSequence[EffectResult]The result of each intent, in call order.
methodperformasync (intent) -> EffectResultKeep intent and return the next result of the script.
paramintentEffectIntentThe recorded intent.
Returns
EffectResultThe next result of the script.
Raises:
LookupError: Every result of the script was played already.
ScriptedStep
classScriptedStep(outcomes)A step that returns the events of its script in order, one batch for each call, and keeps each view.
A call whose outcome is an exception raises it, so the runner records a Failure.
Make a step that plays outcomes.
paramoutcomesSequence[Exception | Sequence[Event]]The outcome of each call, in call order: the events to return, or an exception to raise.
attributecallslist[Trace]The view that each call received, in call order.
method__call__async (trace) -> Sequence[Event]Resume equivalence
A turn that stops after any step and resumes in a new process must record the same trace as a turn that never stopped.
check_resume_equivalence
functioncheck_resume_equivalenceasync (scenario, agent, inputs) -> NoneCheck that a turn that stops after any step and resumes in a new process records the same trace.
The check runs inputs, each a fresh turn of agent, once without a stop. Then, for each step k of that run
except the last, it runs them again: it stops after step k, builds a new process, resumes, and sends the rest.
Last, it runs them with every step on the other of two processes that it built once. Every run must record the
trace of the first run.
A step is one append of a run under its lease. Its events landed before the stop, so a stop after it loses only what the process held. The check releases the lease of the stopped run, as an expired lease is released. Each run uses a new in-memory store whose clock always gives the same instant.
The comparison leaves out the timestamp of each event. It renames the ids of elements and tool calls by their first appearance, because a run makes fresh ids where the world gives none.
Example
from soev_agents.domain import Input, ModelOutput
from soev_agents.runtime import AgentCatalog
from soev_agents.testing import Process, ScriptedChatModel, check_resume_equivalence
# MyAgent is an Agent of your own that calls `model`.
def scenario() -> Process:
model = ScriptedChatModel([ModelOutput(content="Twenty years.")])
return lambda store: AgentCatalog(store, {"chat": MyAgent(model)}, version="test")
await check_resume_equivalence(scenario, "chat", [Input(payload="How long?", model="scripted")])paramagentstrThe name of the agent in the catalog that a process builds.
Raises:
Scenario
type aliasScenario = Callable[[], Process]A function that sets up a fresh outside world and returns the Process that is built in that world.
The world is the doubles behind the adapters, such as a ScriptedChatModel. Every process of one run shares that
world, as real processes share the real one. check_resume_equivalence calls it once for each run, so that each run
starts from the same world.
Process
type aliasProcess = Callable[[TraceStore], AgentCatalog]ResumeMismatchError
errorResumeMismatchErrorDeterminism
What a model call receives must depend only on the code and the trace before the call.
check_determinism
functioncheck_determinismasync (events, configurations) -> NoneRebuild every model call of events under each of configurations and refuse any difference.
Give each configuration adapters that refuse every call (refusing) and vary what a server can vary, so that a
ChatRequest that reads an adapter or a server setting fails here. Every call must rebuild, the ChatRequests
must be equal across configurations, and each must hash as the ChatRequest that its call recorded, when it
recorded one. The check runs replay once for each configuration.
Example
from soev_agents.domain import Event
from soev_agents.runtime import refusing
from soev_agents.testing import built_by, check_determinism
# agent is the Agent under test, and thread is a Thread of that agent with a recorded turn.
trace = await thread.resume()
configurations = [built_by(agent, {"web": None}), built_by(agent, {"web": refusing("the web")})]
await check_determinism(trace.by_type(Event), configurations)paramconfigurationsSequence[GraphFor]One GraphFor for each server configuration. Give at least one.
Raises:
ValueError:configurationsis empty.NondeterministicBuildError: A call does not rebuild under a configuration, two configurations build differentChatRequests, or a rebuiltChatRequestdoes not hash as the recorded one.
built_by
functionbuilt_by(agent, resources) -> GraphForReturn the GraphFor that builds the graphs of agent with resources, as a server builds them.
The function parses the input with the agent's input class, as accepted does. Then it calls the agent's
build with resources, the parsed turn and the model that the input names.
paramagentAgent[Any, Any]The agent under test.
GraphFor
type aliasGraphFor = Callable[[Input], Graph]NondeterministicBuildError
errorNondeterministicBuildErrorA rebuilt ChatRequest depends on something other than the code and the trace.
check_determinism raises it. The message names the model call, counted from 1, and the configuration, counted
from 0.
Store contract
The contract suite holds the tests that every TraceStore must pass.
TraceStoreContract
classTraceStoreContractThe contract suite of TraceStore: the tests that every store must pass.
Subclass it in a test module, in a class whose name starts with Test, so that pytest collects the tests. Give
the subclass make_store. store_type names the class under test: the suite does not read it. The tests are
coroutines, so they need an async pytest plugin, for example pytest-asyncio with asyncio_mode = "auto". Some
tests wait 0.1 seconds for a lease to expire, so the store must date its leases by the real time.
Example
from soev_agents.adapters import InMemoryTraceStore
from soev_agents.ports import TraceStore
from soev_agents.testing import TraceStoreContract
class TestInMemoryTraceStore(TraceStoreContract):
store_type = InMemoryTraceStore
def make_store(self) -> TraceStore:
return InMemoryTraceStore()attributestore_typetypeThe class of the store under test. The suite does not read it.
methodmake_store() -> TraceStoreReturn the store that one test runs on.
Each test calls it once. Return a new, empty store, or a store that an autouse pytest fixture opened for the test.
Returns
TraceStoreThe empty store.
Raises:
NotImplementedError: The subclass does not override it.
methodtest_append_seals_exactly_as_the_sealing_rule_saysasync ()Check that append seals a batch of several streams as seal does, with one timestamp.
methodtest_append_seals_with_seq_and_tsasync ()Check that append returns the events with a seq from 0 and a timestamp.
methodtest_seq_continues_across_appendsasync ()Check that seq continues from one append to the next.
methodtest_seq_is_per_streamasync ()methodtest_any_type_and_payload_read_back_as_appendedasync ()methodtest_a_literal_backslash_u0000_in_text_round_trips_unchangedasync ()Check that the literal text \u0000 in a payload's keys and values reads back unchanged.
methodtest_threads_are_isolatedasync ()methodtest_append_rejects_sealed_entryasync ()methodtest_append_refuses_a_batch_whose_expected_position_movedasync ()Check that append refuses a stale expected position with StreamMovedError and lands nothing.
An expected position of 0 for a new stream lands.
methodtest_failed_append_lands_nothingasync ()methodtest_read_numbers_appends_gaplessly_in_append_orderasync ()Check that read yields the events of all streams in append order, at positions 1, 2, 3 and on.
methodtest_read_after_a_cursor_skips_what_was_seenasync ()Check that read(after=n) yields only the entries after position n.
methodtest_positions_are_per_threadasync ()methodtest_read_of_an_unknown_thread_is_emptyasync ()Check that read of an unknown thread yields nothing, also when it follows.
methodtest_a_following_read_yields_live_appendsasync ()Check that a following read yields the events of the appends that come after it started.
methodtest_a_following_read_waits_rather_than_ending_when_caught_upasync ()Check that a following read waits for more events after the last one, and does not end.
methodtest_read_backwards_yields_the_log_newest_firstasync ()Check that read_backwards yields the entries of read in reverse order.
methodtest_read_backwards_of_an_unknown_thread_is_emptyasync ()Check that read_backwards of an unknown thread yields nothing.
methodtest_read_backwards_reaches_the_first_event_of_a_long_threadasync ()methodtest_read_backwards_stops_where_its_reader_stopsasync ()Check that a reader can close read_backwards after its first entry, which is the newest.
methodtest_delete_removes_the_threadasync ()methodtest_delete_of_an_unknown_thread_is_a_noopasync ()Check that delete of an unknown thread does nothing and raises nothing.
methodtest_a_following_read_ends_when_the_thread_is_deletedasync ()Check that a following read ends when its thread is deleted.
methodtest_a_stale_expectation_on_a_missing_thread_creates_nothingasync ()Check that an append with a stale expected position to an unknown thread creates no thread.
The append raises StreamMovedError. An append with the expected position 0 lands.
methodtest_a_following_read_ends_at_the_delete_even_when_another_thread_takes_the_idasync ()Check that a following read ends at the delete, also when a new thread takes the same id.
methodtest_a_following_read_holding_an_entry_still_ends_after_the_deleteasync ()Check that a following read that yielded an entry before the delete ends after the delete.
It yields no entry that is not an entry of the deleted thread.
methodtest_acquire_gives_a_lease_that_lease_reads_backasync ()methodtest_a_lease_alone_makes_no_threadasync ()methodtest_two_acquisitions_at_once_let_exactly_one_throughasync ()Check that one of two concurrent acquisitions gets the lease, and the other LeaseHeldError.
methodtest_acquire_refuses_a_live_lease_of_another_ownerasync ()Check that acquire raises LeaseHeldError while another owner holds a live lease.
methodtest_the_same_owner_retakes_its_lease_and_the_old_one_is_staleasync ()methodtest_an_expired_lease_is_freeasync ()Check that lease returns None after the expiry, and that another owner can then acquire it.
methodtest_append_under_a_stale_lease_lands_nothingasync ()methodtest_an_append_without_a_lease_is_not_checkedasync ()methodtest_renew_keeps_a_lease_alive_and_refuses_a_stale_oneasync ()methodtest_release_frees_the_lease_and_leaves_another_owners_aloneasync ()methodtest_delete_drops_the_leaseasync ()methodtest_load_restores_a_sealed_log_of_several_streamsasync ()Check that load restores a sealed log of several streams, and that appends continue after it.
methodtest_load_replaces_the_thread_it_namesasync ()methodtest_load_refuses_unsealed_and_gapped_logsasync ()Check that load raises ValueError for an unsealed event or a gap in seq, and loads nothing.
methodtest_threads_lists_each_thread_with_its_first_event_newest_firstasync ()methodtest_threads_pages_follow_one_another_until_the_lastasync ()methodtest_threads_leaves_out_deleted_threads_and_bare_leasesasync ()