orxtra v0.13.0 /tests.conftest
Updated
On this page

Shared scheduler-integration fixtures: AgentTurn-driven mock transports (single- and multi-agent) that execute real tools, plus agent/workflow/scheduler builders and trace_writer/run_id fixtures.

#tests.conftest

#tests.conftest

#AgentTurn

Defines what tool calls an agent makes in a single conversation turn.

Each AgentTurn corresponds to one call to Transport.send(). The transport executes the listed tool calls using the real tool.execute() method, then yields the text_response as the final Result.

#_extract_task_id

python
def _extract_task_id(message: str) -> str | None

Extract the task_id from the scheduler's prompt prefix.

#IntegrationMockTransport

Mock transport that simulates LLM behavior with real tool execution.

Accepts a list of AgentTurns. Each call to send() consumes one turn, executes its tool calls using the actual tool.execute() methods (which call into the Scheduler's lifecycle tools), and yields transport events.

#send

python
async def send(self, message: str, *, model: str, system_prompt: str, tools: list[Tool], session_id: str | None=None) -> AsyncIterator[TransportEvent]

#turns_remaining

python
def turns_remaining(self) -> int

#turns_consumed

python
def turns_consumed(self) -> int

#MultiAgentMockTransport

Mock transport for workflows with multiple agent tasks.

Routes each send() call to the correct turn sequence by matching the system_prompt against registered substrings. This allows different agents in a workflow to have independent scripted behaviors.

#__init__

python
def __init__(self, agent_turns: dict[str, list[AgentTurn]]) -> None

Initialize with agent turn mappings.

Args:

  • agent_turns: Maps agent system_prompt substrings to their

turn sequences. On each send(), the system_prompt is matched against these keys (first match wins).

#_match_agent

python
def _match_agent(self, system_prompt: str) -> str | None

Find the agent key matching this system_prompt.

#send

python
async def send(self, message: str, *, model: str, system_prompt: str, tools: list[Tool], session_id: str | None=None) -> AsyncIterator[TransportEvent]

#turns_remaining

python
def turns_remaining(self, agent_key: str) -> int

#make_agent

python
def make_agent(name: str='test-agent', category: str='default', allow: list[str] | None=None) -> Agent

Create an Agent for testing.

#make_categories

python
def make_categories() -> dict[str, str]

Standard category mapping for tests.

#simple_task

python
def simple_task(name: str='t1', agent: str='test-agent', timeout: int=60, **kwargs: Any) -> TaskSpec

Create a simple agent TaskSpec for testing.

#simple_workflow

python
def simple_workflow(tasks: list[TaskSpec] | None=None, dependencies: dict[str, list[str]] | None=None) -> WorkflowConfig

Create a simple WorkflowConfig for testing.

#make_scheduler

python
def make_scheduler(trace_writer: MockTraceWriter, transport: IntegrationMockTransport | MultiAgentMockTransport, run_id: uuid.UUID, agents: dict[str, Agent] | None=None, categories: dict[str, str] | None=None, read_root: Path | None=None) -> Scheduler

Create a Scheduler with standard test wiring.

The transport is registered under the "anthropic" provider key, matching the default category's "anthropic/claude-sonnet-4-6" model.

#trace_writer

python
def trace_writer() -> MockTraceWriter

#run_id

python
def run_id() -> uuid.UUID
Search