Async-safe telemetry testing helpers with per-test event isolation.
Summary
Functions
Assert exactly expected_count telemetry events were received.
Attach a telemetry handler that only receives events matching the current test ID.
Return metadata map containing the current test's telemetry ID.
Merge the current test's telemetry metadata into an existing map.
Detach a telemetry handler attached via attach_isolated/2 and release any
associated buffered state.
Emit a telemetry event with the current test's context.
Flush buffered telemetry messages for a handler created with buffer: true.
Flush telemetry messages from the mailbox.
Retrieve the current test's telemetry ID.
Retrieve the current test's telemetry ID or raise if missing.
Initialize telemetry isolation for the current process.
Initialize telemetry isolation and update the isolation context.
Execute a function and capture emitted telemetry events.
Types
@type assert_opts() :: [{:timeout, pos_integer()}]
@type attach_opts() :: [ filter_key: atom(), passthrough: boolean(), buffer: boolean(), transform: (telemetry_message() -> term()) ]
@type event() :: [atom()]
@type handler_id() :: String.t()
@type measurements() :: map()
@type metadata() :: map()
@type telemetry_message() :: {:telemetry, event(), measurements(), metadata()}
@type test_id() :: integer()
Functions
@spec assert_telemetry_count( event() | (event() -> boolean()), pos_integer(), assert_opts() ) :: [ telemetry_message() ]
Assert exactly expected_count telemetry events were received.
@spec attach_isolated(event() | [event()], attach_opts()) :: {:ok, handler_id()}
Attach a telemetry handler that only receives events matching the current test ID.
@spec current_test_metadata() :: map()
Return metadata map containing the current test's telemetry ID.
Merge the current test's telemetry metadata into an existing map.
@spec detach_isolated(handler_id()) :: :ok
Detach a telemetry handler attached via attach_isolated/2 and release any
associated buffered state.
@spec emit_with_context(event(), measurements(), metadata()) :: :ok
Emit a telemetry event with the current test's context.
@spec flush_buffered_telemetry(handler_id()) :: [telemetry_message()]
Flush buffered telemetry messages for a handler created with buffer: true.
@spec flush_telemetry(:all | event()) :: [telemetry_message()]
Flush telemetry messages from the mailbox.
@spec get_test_id() :: test_id() | nil
Retrieve the current test's telemetry ID.
@spec get_test_id!() :: test_id()
Retrieve the current test's telemetry ID or raise if missing.
@spec setup_telemetry_isolation() :: {:ok, test_id()}
Initialize telemetry isolation for the current process.
@spec setup_telemetry_isolation(Supertester.IsolationContext.t()) :: {:ok, test_id(), Supertester.IsolationContext.t()}
Initialize telemetry isolation and update the isolation context.
@spec with_telemetry(event() | [event()], (-> result), attach_opts()) :: {result, [telemetry_message()]} when result: term()
Execute a function and capture emitted telemetry events.