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

Shared test doubles: make_auth_context() for scoped AuthContexts, MockTraceWriter for recording trace calls, and MockTransport, a configurable event-sequence/LLM-simulation transport mock.

#tests.shared_mocks

#tests.shared_mocks

#make_auth_context

python
def make_auth_context(scopes: frozenset[str]=ALL_SCOPES, *, trust_tier: TrustTier=TrustTier.SYSTEM, consumer_id: UUID | None=None) -> AuthContext

Build an ephemeral AuthContext for tests that dispatch capabilities.

dispatch() enforces authorization at its choke point, so every test that reaches it needs a context carrying the required scopes. This is the one shared builder -- default to ALL_SCOPES (the operator/system posture) so tests opt into scope restriction only when they mean to exercise it.

#MockTraceWriter

Records all calls for verification.

#create_run

python
async def create_run(self, intent: str, config: dict[str, Any], autonomy_level: str, *, run_id: uuid.UUID, created_by: uuid.UUID) -> uuid.UUID

#transition_run

python
async def transition_run(self, run_id: uuid.UUID, new_status: str, reason: str | None=None, *, principal_id: uuid.UUID) -> None

#create_task

python
async def create_task(self, run_id: uuid.UUID, parent_task_id: uuid.UUID | None, name: str, task_type: str, config: dict[str, Any] | None=None) -> uuid.UUID

#transition_task

python
async def transition_task(self, task_id: uuid.UUID, new_status: str, reason: str | None=None, *, principal_id: uuid.UUID) -> None

#create_task_attempt

python
async def create_task_attempt(self, task_id: uuid.UUID, attempt: int) -> uuid.UUID

#complete_task_attempt

python
async def complete_task_attempt(self, attempt_id: uuid.UUID, agent_output: str, structured_output: dict[str, Any] | None, check_result: dict[str, Any] | None, check_verdict: str | None, session_id: uuid.UUID | None, input_tokens: int, output_tokens: int, reasoning_tokens: int, cache_read_tokens: int, cache_write_tokens: int, cost_usd: Decimal, duration_seconds: float) -> None

#fail_task_attempt

python
async def fail_task_attempt(self, attempt_id: uuid.UUID, error: str, session_id: uuid.UUID | None, input_tokens: int, output_tokens: int, reasoning_tokens: int, cache_read_tokens: int, cache_write_tokens: int, cost_usd: Decimal, duration_seconds: float) -> None

#write_event

python
async def write_event(self, run_id: uuid.UUID | None, event_type: str, data: dict[str, Any], task_id: uuid.UUID | None=None, *, principal_id: uuid.UUID, idempotency_key: str | None=None) -> tuple[uuid.UUID, bool]

#write_transcript_entry

python
async def write_transcript_entry(self, session_id: uuid.UUID, run_id: uuid.UUID, turn: int, role: str, content: str, tool_calls: dict[str, Any] | None=None, tokens: dict[str, Any] | None=None) -> None

#write_coherence_summary

python
async def write_coherence_summary(self, run_id: uuid.UUID, summary: str) -> None

#create_inbox_item

python
async def create_inbox_item(self, run_id: uuid.UUID, decision_type: str, question: str, options: list[dict[str, Any]], **kwargs: object) -> uuid.UUID

#write_lesson

python
async def write_lesson(self, **kwargs: object) -> None

#write_constraint

python
async def write_constraint(self, **kwargs: object) -> None

#write_context_diff

python
async def write_context_diff(self, attempt_id: uuid.UUID, pre_refinement: str, refinement_diff: str) -> None

#create_iteration

python
async def create_iteration(self, task_id: uuid.UUID, index: int, item_value: object) -> uuid.UUID

#complete_iteration

python
async def complete_iteration(self, iteration_id: uuid.UUID, output: str | None, structured_output: dict[str, Any] | None, check_results: list[dict[str, Any]] | None) -> None

#fail_iteration

python
async def fail_iteration(self, iteration_id: uuid.UUID, error: str) -> None

#subscribe_run_control

python
async def subscribe_run_control(self, run_id: uuid.UUID, callback: Any) -> None

#unsubscribe_run_control

python
async def unsubscribe_run_control(self, run_id: uuid.UUID) -> None

#get_calls

python
def get_calls(self, method: str) -> list[dict[str, Any]]

#MockTransport

Configurable transport mock for all test scenarios.

Modes:

  • Event sequence mode (session tests): set_events() / set_resume_events()
  • LLM simulation mode (scheduler tests): auto_execute_tools=True
  • No-tools mode: auto_execute_tools=False, no events set

#inject_history

python
def inject_history(self, session_id: str, messages: list[dict[str, Any]]) -> None

Mock of Transport.inject_history for session resume tests.

#set_events

python
def set_events(self, *sequences: list[TransportEvent]) -> None

#set_resume_events

python
def set_resume_events(self, *sequences: list[TransportEvent]) -> None

#send

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

#resume

python
async def resume(self, continuation: Continuation, await_result: str, *, model: str, system_prompt: str, tools: list[Tool]) -> AsyncIterator[TransportEvent]
Search