m3.testing
Signatures use ... for factory-backed or opaque defaults. Model field tables show required status, defaults, constraints, and descriptions.
ArtifactIntegrityError
m3.testing.ArtifactIntegrityErrorA recorded artifact does not match its content-addressed metadata.
ExpectedCall
m3.testing.ExpectedCall(
method: str,
matcher: _Mapping[str, _Any] = ...,
response: _Any = None,
optional_call: bool = False,
minimum: int = 1,
maximum: int | None = 1,
subset_match: bool = False,
unordered_group: str | None = None,
fallback_call: bool = False,
calls: int = 0,
) -> Noneoptional(
self,
) -> ExpectedCallrepeated(
self,
minimum: int = 0,
maximum: int | None = None,
) -> ExpectedCallsubset(
self,
) -> ExpectedCallunordered(
self,
group: str = 'default',
) -> ExpectedCallfallback(
self,
) -> ExpectedCallreturns(
self,
value: _Any,
) -> ExpectedCallFaultInjector
m3.testing.FaultInjector(
delays: dict[str, float] = ...,
gates: dict[str, Gate] = ...,
disconnect_methods: set[str] = ...,
malformed_methods: set[str] = ...,
partial_methods: set[str] = ...,
invalid_schema_methods: set[str] = ...,
invalid_result_methods: set[str] = ...,
protocol_errors: dict[str, tuple[int, str]] = ...,
oversized_methods: dict[str, int] = ...,
reordered_methods: set[str] = ...,
duplicate_id_methods: set[str] = ...,
cancel_before_methods: set[str] = ...,
process_crash_methods: set[str] = ...,
race_gates: dict[str, Gate] = ...,
reorder_window: float = 0.01,
) -> NoneDeterministic faults for decoded in-process or literal stdio fixtures.
delay(
self,
method: str,
seconds: float,
) -> FaultInjectorhang(
self,
method: str,
gate: Gate | None = None,
) -> Gatedisconnect(
self,
method: str,
) -> FaultInjectormalformed(
self,
method: str,
) -> FaultInjectorpartial_frame(
self,
method: str,
) -> FaultInjectorinvalid_schema(
self,
method: str,
) -> FaultInjectorinvalid_result(
self,
method: str,
) -> FaultInjectorAdvertise an output schema and return a violating structured result.
duplicate_id(
self,
method: str,
) -> FaultInjectorreorder(
self,
method: str,
) -> FaultInjectoroversized(
self,
method: str,
size: int,
) -> FaultInjectorcancel_before_dispatch(
self,
method: str,
) -> FaultInjectorprocess_crash(
self,
method: str,
) -> FaultInjectorExit the dedicated stdio fixture with a non-zero status.
cancellation_race(
self,
method: str,
) -> FaultInjectorrace_gate(
self,
method: str,
) -> GateReturn the gate deciding whether a raced response is released.
protocol_error(
self,
method: str,
code: int = -32000,
message: str = 'mock protocol error',
) -> FaultInjectorstdio_server(
self,
*,
name: str = 'wire-fault',
) -> _StdioServerReturn a safe stdio fixture that applies literal wire faults.
close_fixture(
self,
) -> Nonebefore(
self,
method: str,
) -> Noneresponse(
self,
method: str,
value: _Any,
) -> _AnyApply deterministic response-shape faults after handler output.
Gate
m3.testing.Gate(
*,
open: bool = False,
) -> NoneDeterministic async gate for delay and hang tests.
is_open(property)
open(
self,
) -> Nonerelease(
self,
) -> Noneclose(
self,
) -> Nonewait(
self,
*,
poll_seconds: float = 0.001,
) -> Nonewait_async(
self,
*,
poll_seconds: float = 0.001,
) -> NoneAlias with an explicit name for composing deterministic races.
MockExpectationError
m3.testing.MockExpectationErrorA scripted mock interaction did not match.
MockMCPServer
m3.testing.MockMCPServer(
name: str = 'mock-mcp',
*,
version: str = '1',
strict: bool = True,
redaction_config: _RedactionConfig | None = None,
faults: FaultInjector | None = None,
state: _Mapping[str, _Any] | None = None,
) -> NoneDecorator-defined and stateful official in-process MCP server.
tool(
self,
name: str | _Callable[..., _Any] | None = None,
*,
description: str | None = None,
input_schema: _Mapping[str, _Any] | None = None,
output_schema: _Mapping[str, _Any] | None = None,
) -> _Anyresource(
self,
uri: str,
*,
name: str | None = None,
mime_type: str | None = None,
) -> _Callable[[_Callable[_P, _R]], _Callable[_P, _R]]resource_template(
self,
uri_template: str,
*,
name: str | None = None,
description: str | None = None,
mime_type: str | None = None,
) -> _types.ResourceTemplateRegister a resource template exposed by resources/templates/list.
prompt(
self,
name: str | _Callable[..., _Any] | None = None,
*,
description: str | None = None,
) -> _Anyexpect(
self,
method: str,
*,
optional: bool = False,
repeat: int | tuple[int, int | None] | None = None,
subset: bool = False,
unordered: str | None = None,
fallback: bool = False,
**matcher: _Any,
) -> ExpectedCallexpect_call(
self,
method: str,
*,
optional: bool = False,
repeat: int | tuple[int, int | None] | None = None,
subset: bool = False,
unordered: str | None = None,
fallback: bool = False,
**matcher: _Any,
) -> ExpectedCallexpect_tool_call(
self,
name: str,
arguments: _Mapping[str, _Any] | None = None,
**options: _Any,
) -> ExpectedCallserver(
self,
) -> _Serverin_process(
self,
*,
name: str | None = None,
) -> _InProcessServerReturn a decoded SessionMessage in-process fixture.
verify(
self,
) -> Nonerecording(
self,
) -> Recordingclose(
self,
) -> NoneMockProtocolError
m3.testing.MockProtocolError(
code: int,
message: str,
) -> NoneRecordedArtifact
m3.testing.RecordedArtifact(
artifact_id: str,
sha256: str,
size_bytes: int,
media_type: str | None = None,
) -> NoneContent-addressed artifact metadata carried by a recording.
from_bytes(
cls,
artifact_id: str,
content: bytes,
*,
media_type: str | None = None,
) -> RecordedArtifactvalidate(
self,
content: bytes,
) -> Nonemodel(
self,
) -> dict[str, _JsonValue]RecordedInteraction
m3.testing.RecordedInteraction(
method: str,
params: _Mapping[str, _JsonValue],
response: _JsonValue | None = None,
error: _Mapping[str, _JsonValue] | None = None,
sequence: int = 0,
provenance: tuple[str, ...] = (),
) -> Nonemodel(
self,
) -> dict[str, _JsonValue]Recording
m3.testing.Recording(
interactions: tuple[RecordedInteraction, ...],
redaction_bound: bool,
provenance: tuple[str, ...] = (),
server_name: str = 'mock-mcp',
server_version: str = '1',
initialization: _Mapping[str, _JsonValue] = ...,
artifacts: tuple[RecordedArtifact, ...] = (),
redaction_bindings: tuple[RedactionBinding, ...] = ...,
_runtime: _RedactionRuntime = ...,
) -> Nonewith_redaction_config(
self,
config: _RedactionConfig,
) -> RecordingBind an explicit in-process redaction config without serializing secrets.
validate_artifacts(
self,
contents: _Mapping[str, bytes],
) -> NoneValidate every supplied artifact against the recorded digest/length.
model_dump(
self,
*,
config: _RedactionConfig | None = None,
) -> dict[str, _JsonValue]to_json(
self,
*,
config: _RedactionConfig | None = None,
) -> strredacted(
self,
*,
config: _RedactionConfig | None = None,
) -> RecordingReturn a detached recording whose persisted values are redacted.
from_json(
cls,
value: str,
) -> RecordingRedactionBinding
m3.testing.RedactionBinding(
source: str = 'explicit',
version: str = '1',
wildcard_paths: tuple[str, ...] = (),
secret_references: tuple[str, ...] = (),
) -> NoneSerializable redaction provenance without secret values.
model(
self,
) -> dict[str, _JsonValue]ReplayMismatch
m3.testing.ReplayMismatchA strict replay request differs from the recorded interaction.
ReplayServer
m3.testing.ReplayServer(
recording: Recording,
*,
strict: bool = True,
redaction_config: _RedactionConfig | None = None,
artifacts: _Mapping[str, bytes] | None = None,
expected_server: tuple[str, str] | None = None,
) -> NoneStrict JSON-only replay matcher.
match(
self,
method: str,
params: _Mapping[str, _Any],
) -> RecordedInteractionverify_replay(
self,
) -> Noneprovenance(property)
VirtualClock
m3.testing.VirtualClock(
start: float = 0.0,
) -> NoneA process-local virtual clock with awaitable deterministic sleeps.
now(property)
sleep(
self,
duration: float,
) -> Noneadvance(
self,
duration: float,
) -> float