diff --git a/lib/sentry/test.ex b/lib/sentry/test.ex
index e383fd21..4ac09c67 100644
--- a/lib/sentry/test.ex
+++ b/lib/sentry/test.ex
@@ -65,7 +65,7 @@ defmodule Sentry.Test do
   Opens a Bypass on a random port, configures the DSN to point to it,
   wires up `before_send` / `before_send_log` callbacks to capture structs
   in an isolated ETS table, and starts a per-test `Sentry.TelemetryProcessor`
-  (via `setup_telemetry_processor/0`) so that assertions work for events
+  (via `setup_telemetry_processor/1`) so that assertions work for events
   that travel through the TelemetryProcessor pipeline (logs, metrics, or
   `send_result: :none`).
 
@@ -79,6 +79,23 @@ defmodule Sentry.Test do
   Any extra Sentry config options (e.g., `dedup_events: false`, `traces_sample_rate: 1.0`)
   will be forwarded to the test config.
 
+  The reserved `:telemetry_processor` option is *not* forwarded to the test
+  config. Instead, its value (a keyword list) is passed to the per-test
+  `Sentry.TelemetryProcessor` (e.g. `buffer_configs`, `buffer_capacities`,
+  `scheduler_weights`, `transport_capacity`). This replaces the need to
+  manually `stop_supervised!/1` and re-`start_supervised!/2` the processor.
+
+  The reserved `:collect_envelopes` option is *not* forwarded to the test
+  config either. When set, a Bypass envelope collector is wired up
+  automatically and its reference is returned under the `:ref` key:
+
+    * `true` — set up the collector with no options;
+    * a keyword list — forwarded to `setup_bypass_envelope_collector/2`
+      (e.g. `[type: "check_in"]` to only collect a given item type).
+
+  This collapses the common `bypass = setup_sentry(...); ref =
+  setup_bypass_envelope_collector(bypass)` two-step into one call.
+
   ## Examples
 
       setup do
@@ -89,28 +106,38 @@ defmodule Sentry.Test do
         Sentry.Test.setup_sentry(dedup_events: false)
       end
 
-  Replacing the auto-started processor with a custom-configured one:
+  Configuring the per-test processor (e.g. a smaller log batch size):
 
       setup do
-        %{telemetry_processor: name} = ctx = Sentry.Test.setup_sentry()
-        stop_supervised!(name)
-
-        start_supervised!(
-          {Sentry.TelemetryProcessor,
-           name: name, buffer_configs: %{log: %{batch_size: 1}}},
-          id: name
+        Sentry.Test.setup_sentry(
+          telemetry_processor: [buffer_configs: %{log: %{batch_size: 1}}]
         )
+      end
 
-        ctx
+  Collecting envelopes directly as the ExUnit setup return:
+
+      setup do
+        Sentry.Test.setup_sentry(collect_envelopes: true, traces_sample_rate: 1.0)
+      end
+
+      test "...", %{ref: ref} do
+        # ...
       end
 
   """
   @doc since: "13.0.0"
-  @spec setup_sentry(keyword()) :: %{bypass: term(), telemetry_processor: atom()}
+  @spec setup_sentry(keyword()) :: %{
+          :bypass => term(),
+          :telemetry_processor => atom(),
+          optional(:ref) => reference()
+        }
   def setup_sentry(extra_config \\ []) do
     ensure_bypass_loaded!()
     ensure_nimble_ownership_loaded!()
 
+    {tp_opts, extra_config} = Keyword.pop(extra_config, :telemetry_processor, [])
+    {collect_envelopes, extra_config} = Keyword.pop(extra_config, :collect_envelopes, false)
+
     # Open a per-test Bypass and stub the envelope endpoint
     bypass = Bypass.open()
 
@@ -120,13 +147,25 @@ defmodule Sentry.Test do
 
     # Start a per-test TelemetryProcessor before setup_collector/1 so that
     # the collector wires this test's scheduler into its registry.
-    processor_name = setup_telemetry_processor()
+    processor_name = setup_telemetry_processor(tp_opts)
 
     # Set up collector with DSN pointing to this test's Bypass
     bypass_config = [dsn: "http://public:secret@localhost:#{bypass.port}/1"]
     setup_collector(bypass_config ++ extra_config)
 
-    %{bypass: bypass, telemetry_processor: processor_name}
+    case collect_envelopes do
+      false ->
+        %{bypass: bypass, telemetry_processor: processor_name}
+
+      collect ->
+        collector_opts = if is_list(collect), do: collect, else: []
+
+        %{
+          bypass: bypass,
+          telemetry_processor: processor_name,
+          ref: setup_bypass_envelope_collector(bypass, collector_opts)
+        }
+    end
   end
 
   @doc """
@@ -153,34 +192,69 @@ defmodule Sentry.Test do
   Must be called from within an ExUnit test because it uses
   `ExUnit.Callbacks.start_supervised!/2` for automatic cleanup.
 
-  If a per-test processor is already registered for this test (for example
-  when using `Sentry.Case`), this function is idempotent and returns the
-  existing processor name instead of starting a new one.
+  ## Options
+
+  `tp_opts` is a keyword list forwarded to the per-test
+  `Sentry.TelemetryProcessor` child spec (e.g. `buffer_configs`,
+  `buffer_capacities`, `scheduler_weights`, `transport_capacity`).
+
+  Idempotency depends on `tp_opts`:
+
+    * with no `tp_opts`, an already-registered live processor (for example
+      one started by `Sentry.Case`) is reused and its name returned;
+    * with `tp_opts`, an already-registered live processor is stopped and
+      restarted under the same name with the given options, so callers no
+      longer need to `stop_supervised!/1` + `start_supervised!/2` manually.
   """
   @doc since: "13.0.0"
-  @spec setup_telemetry_processor() :: atom()
-  def setup_telemetry_processor do
+  @spec setup_telemetry_processor(keyword()) :: atom()
+  def setup_telemetry_processor(tp_opts \\ []) do
     case Process.get(:sentry_telemetry_processor) do
       name when is_atom(name) and not is_nil(name) ->
-        if processor_alive?(name), do: name, else: start_telemetry_processor()
+        cond do
+          not processor_alive?(name) -> start_telemetry_processor(tp_opts)
+          tp_opts == [] -> name
+          true -> restart_telemetry_processor(name, tp_opts)
+        end
 
       _ ->
-        start_telemetry_processor()
+        start_telemetry_processor(tp_opts)
     end
   end
 
-  defp start_telemetry_processor do
+  defp start_telemetry_processor(tp_opts) do
     uid = System.unique_integer([:positive])
     processor_name = :"test_telemetry_processor_#{uid}"
 
-    ExUnit.Callbacks.start_supervised!(
-      {Sentry.TelemetryProcessor,
-       name: processor_name, processor_resolver: &Sentry.Test.Registry.lookup_processor_for/1},
-      id: processor_name
-    )
+    start_processor_child(processor_name, tp_opts)
 
+    # Must be set before tag_scheduler/1, which reads
+    # `:sentry_telemetry_processor` from this process's dictionary via
+    # `fetch_owner_processor/1`. Tagging would otherwise be a silent no-op.
     Process.put(:sentry_telemetry_processor, processor_name)
 
+    tag_scheduler(processor_name)
+    processor_name
+  end
+
+  defp restart_telemetry_processor(name, tp_opts) do
+    ExUnit.Callbacks.stop_supervised!(name)
+    start_processor_child(name, tp_opts)
+    # The process dictionary already holds `name`; the new scheduler pid
+    # must be re-tagged since the old one died with the old supervisor.
+    tag_scheduler(name)
+    name
+  end
+
+  defp start_processor_child(name, tp_opts) do
+    opts =
+      [name: name, processor_resolver: &Sentry.Test.Registry.lookup_processor_for/1]
+      |> Keyword.merge(tp_opts)
+
+    ExUnit.Callbacks.start_supervised!({Sentry.TelemetryProcessor, opts}, id: name)
+  end
+
+  defp tag_scheduler(processor_name) do
     scheduler_pid = Sentry.TelemetryProcessor.get_scheduler(processor_name)
 
     if scheduler_pid do
@@ -191,7 +265,7 @@ defmodule Sentry.Test do
       tag_processor_for_allowed_pid(self(), scheduler_pid)
     end
 
-    processor_name
+    :ok
   end
 
   defp processor_alive?(name) do
@@ -366,9 +440,9 @@ defmodule Sentry.Test do
   # callback drops the event.
   #
   # The owner's processor name is looked up from its process
-  # dictionary; tests set it in `setup_telemetry_processor/0`. If the
+  # dictionary; tests set it in `setup_telemetry_processor/1`. If the
   # owner has no per-test processor (e.g. legacy
-  # `start_collecting/1` without `setup_telemetry_processor/0`), the
+  # `start_collecting/1` without `setup_telemetry_processor/1`), the
   # tag is skipped and the buffered event still falls back to the
   # global processor — the same behaviour as before this change.
   defp tag_processor_for_allowed_pid(owner_pid, allowed_pid) do
diff --git a/lib/sentry/test/assertions.ex b/lib/sentry/test/assertions.ex
index 56ae1476..e2ef721f 100644
--- a/lib/sentry/test/assertions.ex
+++ b/lib/sentry/test/assertions.ex
@@ -57,6 +57,7 @@ defmodule Sentry.Test.Assertions do
   import ExUnit.Assertions, only: [flunk: 1]
 
   @default_timeout 1000
+  @refute_timeout 100
   @max_poll_interval 50
 
   @type_to_pop %{
@@ -246,6 +247,112 @@ defmodule Sentry.Test.Assertions do
     find_item!(items, criteria, "report")
   end
 
+  @doc """
+  Asserts that exactly one transaction envelope is collected via `ref` and
+  matches the given `criteria`.
+
+  This is shorthand for:
+
+      assert_sentry_report(collect_sentry_transactions(ref, 1), criteria)
+
+  Reserved option in `criteria`:
+
+    * `:timeout` — ms to wait for the envelope (default: `#{1000}`)
+
+  Use with a collector created via
+  `Sentry.Test.setup_sentry(collect_envelopes: true)`.
+
+  Returns the matched transaction map.
+
+  ## Examples
+
+      test "GET /transaction", %{conn: conn, ref: ref} do
+        get(conn, ~p"/transaction")
+
+        assert_sentry_transaction(ref,
+          transaction: "test_span",
+          contexts: %{trace: %{op: "test_span"}}
+        )
+      end
+
+  """
+  @doc since: "13.0.2"
+  @spec assert_sentry_transaction(reference(), keyword()) :: map()
+  def assert_sentry_transaction(ref, criteria \\ []) when is_reference(ref) do
+    {timeout, criteria} = Keyword.pop(criteria, :timeout, @default_timeout)
+    transactions = Sentry.Test.collect_sentry_transactions(ref, 1, timeout: timeout)
+    tx = unwrap_single!(transactions, "transaction", timeout)
+    assert_fields!(tx, criteria, "transaction")
+    tx
+  end
+
+  @doc """
+  Collects up to `:count` transaction envelopes via `ref` and returns the
+  first one matching `criteria`. Raises if no transaction matches.
+
+  This is shorthand for:
+
+      find_sentry_report!(collect_sentry_transactions(ref, count, timeout: timeout), criteria)
+
+  Reserved options in `criteria`:
+
+    * `:count`   — max number of envelopes to collect (default: `1`)
+    * `:timeout` — ms to wait for each envelope (default: `#{1000}`)
+
+  ## Examples
+
+      find_sentry_transaction!(ref,
+        count: 10,
+        timeout: 2000,
+        transaction: "PhoenixAppWeb.UserLive.Index.handle_event#save",
+        contexts: %{trace: %{origin: "opentelemetry_phoenix"}}
+      )
+
+  """
+  @doc since: "13.0.2"
+  @spec find_sentry_transaction!(reference(), keyword()) :: map()
+  def find_sentry_transaction!(ref, criteria) when is_reference(ref) do
+    {count, criteria} = Keyword.pop(criteria, :count, 1)
+    {timeout, criteria} = Keyword.pop(criteria, :timeout, @default_timeout)
+    transactions = Sentry.Test.collect_sentry_transactions(ref, count, timeout: timeout)
+    find_item!(transactions, criteria, "transaction")
+  end
+
+  @doc """
+  Asserts that **no** Sentry check-in envelope reaches the Bypass envelope
+  collector identified by `ref` within `timeout` ms (default
+  `#{@refute_timeout}`).
+
+  The `Sentry.TelemetryProcessor` pipeline is flushed first, so a check-in
+  that was buffered (rather than sent synchronously) is still detected
+  rather than silently slipping past the timeout window.
+
+  Use with a collector created via
+  `Sentry.Test.setup_sentry(collect_envelopes: [type: "check_in"])` (or
+  `Sentry.Test.setup_bypass_envelope_collector/2` with `type: "check_in"`),
+  so that only check-in envelopes are forwarded to the test process.
+
+  ## Examples
+
+      test "ignores non-cron jobs", %{ref: ref} do
+        :telemetry.execute([:oban, :job, :start], %{}, %{job: %Oban.Job{}})
+        refute_sentry_check_in(ref)
+      end
+
+  """
+  @doc since: "13.0.2"
+  @spec refute_sentry_check_in(reference(), timeout()) :: :ok
+  def refute_sentry_check_in(ref, timeout \\ @refute_timeout) when is_reference(ref) do
+    maybe_flush(timeout)
+
+    receive do
+      {:bypass_envelope, ^ref, body} ->
+        flunk("expected no check-in to be sent, but received envelope: #{body}")
+    after
+      timeout -> :ok
+    end
+  end
+
   # --- Private helpers ---
 
   defp pop_for_type(type) do
diff --git a/test/sentry/integrations/oban/cron_test.exs b/test/sentry/integrations/oban/cron_test.exs
index c465dc3f..d3e369e5 100644
--- a/test/sentry/integrations/oban/cron_test.exs
+++ b/test/sentry/integrations/oban/cron_test.exs
@@ -15,59 +15,61 @@ defmodule Sentry.Integrations.Oban.CronTest do
   end
 
   setup do
-    SentryTest.setup_sentry(dedup_events: false, environment_name: "test")
+    SentryTest.setup_sentry(
+      dedup_events: false,
+      environment_name: "test",
+      collect_envelopes: [type: "check_in"]
+    )
   end
 
   for event_type <- [:start, :stop, :exception] do
-    test "ignores #{event_type} events without a cron meta", %{bypass: bypass} do
-      Bypass.down(bypass)
+    test "ignores #{event_type} events without a cron meta", %{ref: ref} do
       :telemetry.execute([:oban, :job, unquote(event_type)], %{}, %{job: %Oban.Job{}})
+      refute_sentry_check_in(ref)
     end
 
-    test "ignores #{event_type} events without a cron_expr meta", %{bypass: bypass} do
-      Bypass.down(bypass)
-
+    test "ignores #{event_type} events without a cron_expr meta", %{ref: ref} do
       :telemetry.execute([:oban, :job, unquote(event_type)], %{}, %{
         job: %Oban.Job{meta: %{"cron" => true}}
       })
-    end
 
-    test "ignores #{event_type} events with a cron expr of @reboot", %{bypass: bypass} do
-      Bypass.down(bypass)
+      refute_sentry_check_in(ref)
+    end
 
+    test "ignores #{event_type} events with a cron expr of @reboot", %{ref: ref} do
       :telemetry.execute([:oban, :job, unquote(event_type)], %{}, %{
         job: %Oban.Job{
           worker: "Sentry.MyWorker",
           meta: %{"cron" => true, "cron_expr" => "@reboot"}
         }
       })
+
+      refute_sentry_check_in(ref)
     end
 
     test "ignores #{event_type} events with a cron expr of @reboot even with timezone", %{
-      bypass: bypass
+      ref: ref
     } do
-      Bypass.down(bypass)
-
       :telemetry.execute([:oban, :job, unquote(event_type)], %{}, %{
         job: %Oban.Job{
           worker: "Sentry.MyWorker",
           meta: %{"cron" => true, "cron_expr" => "@reboot", "cron_tz" => "Etc/UTC"}
         }
       })
-    end
 
-    test "ignores #{event_type} events with a cron expr that is not a string", %{bypass: bypass} do
-      Bypass.down(bypass)
+      refute_sentry_check_in(ref)
+    end
 
+    test "ignores #{event_type} events with a cron expr that is not a string", %{ref: ref} do
       :telemetry.execute([:oban, :job, unquote(event_type)], %{}, %{
         job: %Oban.Job{worker: "Sentry.MyWorker", meta: %{"cron" => true, "cron_expr" => 123}}
       })
+
+      refute_sentry_check_in(ref)
     end
   end
 
-  test "captures start events with monitor config", %{bypass: bypass} do
-    ref = SentryTest.setup_bypass_envelope_collector(bypass, type: "check_in")
-
+  test "captures start events with monitor config", %{ref: ref} do
     :telemetry.execute([:oban, :job, :start], %{}, %{
       job: %Oban.Job{
         worker: "Sentry.MyWorker",
@@ -111,9 +113,7 @@ defmodule Sentry.Integrations.Oban.CronTest do
         {"@annually", "year"}
       ] do
     test "captures stop events with monitor config and state of #{inspect(oban_state)} and frequency of #{frequency}",
-         %{bypass: bypass} do
-      ref = SentryTest.setup_bypass_envelope_collector(bypass, type: "check_in")
-
+         %{ref: ref} do
       duration = System.convert_time_unit(12_099, :millisecond, :native)
 
       :telemetry.execute([:oban, :job, :stop], %{duration: duration}, %{
@@ -146,9 +146,7 @@ defmodule Sentry.Integrations.Oban.CronTest do
     end
   end
 
-  test "captures exception events with monitor config", %{bypass: bypass} do
-    ref = SentryTest.setup_bypass_envelope_collector(bypass, type: "check_in")
-
+  test "captures exception events with monitor config", %{ref: ref} do
     duration = System.convert_time_unit(12_099, :millisecond, :native)
 
     :telemetry.execute([:oban, :job, :exception], %{duration: duration}, %{
@@ -179,7 +177,7 @@ defmodule Sentry.Integrations.Oban.CronTest do
     )
   end
 
-  test "uses default monitor configuration in Sentry's config if present", %{bypass: bypass} do
+  test "uses default monitor configuration in Sentry's config if present", %{ref: ref} do
     put_test_config(
       integrations: [
         monitor_config_defaults: [
@@ -190,8 +188,6 @@ defmodule Sentry.Integrations.Oban.CronTest do
       ]
     )
 
-    ref = SentryTest.setup_bypass_envelope_collector(bypass, type: "check_in")
-
     :telemetry.execute([:oban, :job, :exception], %{duration: 0}, %{
       state: :success,
       job: %Oban.Job{
@@ -219,9 +215,7 @@ defmodule Sentry.Integrations.Oban.CronTest do
 
   @tag attach_opts: [monitor_slug_generator: {__MODULE__, :custom_name_generator}]
   test "monitor_slug is not affected if the custom monitor_name_generator does not target the worker",
-       %{bypass: bypass} do
-    ref = SentryTest.setup_bypass_envelope_collector(bypass, type: "check_in")
-
+       %{ref: ref} do
     :telemetry.execute([:oban, :job, :start], %{}, %{
       job: %Oban.Job{
         worker: "Sentry.MyWorker",
@@ -236,9 +230,8 @@ defmodule Sentry.Integrations.Oban.CronTest do
 
   @tag attach_opts: [monitor_slug_generator: {__MODULE__, :custom_name_generator}]
   test "monitor_slug is set based on the custom monitor_name_generator if it targets the worker",
-       %{bypass: bypass} do
+       %{ref: ref} do
     client_name = "my-client"
-    ref = SentryTest.setup_bypass_envelope_collector(bypass, type: "check_in")
 
     :telemetry.execute([:oban, :job, :start], %{}, %{
       job: %Oban.Job{
@@ -253,9 +246,7 @@ defmodule Sentry.Integrations.Oban.CronTest do
     assert_sentry_report(check_in_body, monitor_slug: "sentry-client-worker-my-client")
   end
 
-  test "custom options overide job metadata", %{bypass: bypass} do
-    ref = SentryTest.setup_bypass_envelope_collector(bypass, type: "check_in")
-
+  test "custom options overide job metadata", %{ref: ref} do
     defmodule WorkerWithCustomOptions do
       use Oban.Worker
 
diff --git a/test/sentry/integrations/quantum/cron_test.exs b/test/sentry/integrations/quantum/cron_test.exs
index c977e519..61b4510e 100644
--- a/test/sentry/integrations/quantum/cron_test.exs
+++ b/test/sentry/integrations/quantum/cron_test.exs
@@ -16,21 +16,23 @@ defmodule Sentry.Integrations.Quantum.CronTest do
   end
 
   setup do
-    SentryTest.setup_sentry(dedup_events: false, environment_name: "test")
+    SentryTest.setup_sentry(
+      dedup_events: false,
+      environment_name: "test",
+      collect_envelopes: [type: "check_in"]
+    )
   end
 
   for event_type <- [:start, :stop, :exception] do
-    test "ignores #{event_type} events without a cron meta", %{bypass: bypass} do
-      Bypass.down(bypass)
-
+    test "ignores #{event_type} events without a cron meta", %{ref: ref} do
       :telemetry.execute([:quantum, :job, unquote(event_type)], %{}, %{
         job: Scheduler.new_job(name: :test_job)
       })
-    end
 
-    test "ignores #{event_type} events with a cron expr of @reboot", %{bypass: bypass} do
-      Bypass.down(bypass)
+      refute_sentry_check_in(ref)
+    end
 
+    test "ignores #{event_type} events with a cron expr of @reboot", %{ref: ref} do
       :telemetry.execute([:quantum, :job, unquote(event_type)], %{}, %{
         job:
           Scheduler.new_job(
@@ -38,11 +40,12 @@ defmodule Sentry.Integrations.Quantum.CronTest do
             schedule: Crontab.CronExpression.Parser.parse!("@reboot")
           )
       })
+
+      refute_sentry_check_in(ref)
     end
   end
 
-  test "captures start events with monitor config", %{bypass: bypass} do
-    ref = SentryTest.setup_bypass_envelope_collector(bypass, type: "check_in")
+  test "captures start events with monitor config", %{ref: ref} do
     span_ref = make_ref()
 
     :telemetry.execute([:quantum, :job, :start], %{}, %{
@@ -73,8 +76,7 @@ defmodule Sentry.Integrations.Quantum.CronTest do
     )
   end
 
-  test "captures exception events with monitor config", %{bypass: bypass} do
-    ref = SentryTest.setup_bypass_envelope_collector(bypass, type: "check_in")
+  test "captures exception events with monitor config", %{ref: ref} do
     span_ref = make_ref()
 
     duration = System.convert_time_unit(12_099, :millisecond, :native)
@@ -108,8 +110,7 @@ defmodule Sentry.Integrations.Quantum.CronTest do
     )
   end
 
-  test "captures stop events with monitor config", %{bypass: bypass} do
-    ref = SentryTest.setup_bypass_envelope_collector(bypass, type: "check_in")
+  test "captures stop events with monitor config", %{ref: ref} do
     span_ref = make_ref()
 
     duration = System.convert_time_unit(12_099, :millisecond, :native)
@@ -146,8 +147,7 @@ defmodule Sentry.Integrations.Quantum.CronTest do
         {:some_job, "quantum-some-job"},
         {MyApp.MyJob, "quantum-my-app-my-job"}
       ] do
-    test "works for a job named #{inspect(job_name)}", %{bypass: bypass} do
-      ref = SentryTest.setup_bypass_envelope_collector(bypass, type: "check_in")
+    test "works for a job named #{inspect(job_name)}", %{ref: ref} do
       span_ref = make_ref()
 
       duration = System.convert_time_unit(12_099, :millisecond, :native)
@@ -166,8 +166,7 @@ defmodule Sentry.Integrations.Quantum.CronTest do
     end
   end
 
-  test "works for a job without the name", %{bypass: bypass} do
-    ref = SentryTest.setup_bypass_envelope_collector(bypass, type: "check_in")
+  test "works for a job without the name", %{ref: ref} do
     span_ref = make_ref()
 
     duration = System.convert_time_unit(12_099, :millisecond, :native)
@@ -181,8 +180,7 @@ defmodule Sentry.Integrations.Quantum.CronTest do
     assert_sentry_report(check_in_body, monitor_slug: "quantum-generic-job")
   end
 
-  test "start event and same ref stop event have same check-in id", %{bypass: bypass} do
-    ref = SentryTest.setup_bypass_envelope_collector(bypass, type: "check_in")
+  test "start event and same ref stop event have same check-in id", %{ref: ref} do
     span_ref = make_ref()
     id = CheckInIDMappings.lookup_or_insert_new("quantum-#{:erlang.phash2(span_ref)}")
 
diff --git a/test/sentry/metrics_integration_test.exs b/test/sentry/metrics_integration_test.exs
index 02290ae6..46f9a3df 100644
--- a/test/sentry/metrics_integration_test.exs
+++ b/test/sentry/metrics_integration_test.exs
@@ -7,23 +7,13 @@ defmodule Sentry.MetricsIntegrationTest do
   alias Sentry.{Metrics, TelemetryProcessor}
   alias Sentry.Telemetry.Buffer
 
-  setup context do
-    bypass = Bypass.open()
-
-    ref = setup_bypass_envelope_collector(bypass)
-
-    stop_supervised!(context.telemetry_processor)
-
-    uid = System.unique_integer([:positive])
-    processor_name = :"test_metric_integration_#{uid}"
-
-    start_supervised!(
-      {TelemetryProcessor, name: processor_name, buffer_configs: %{metric: %{batch_size: 1}}},
-      id: processor_name
-    )
-
-    Process.put(:sentry_telemetry_processor, processor_name)
-    put_test_config(dsn: "http://public:secret@localhost:#{bypass.port}/1", enable_metrics: true)
+  setup do
+    %{bypass: bypass, telemetry_processor: processor_name, ref: ref} =
+      Sentry.Test.setup_sentry(
+        collect_envelopes: true,
+        enable_metrics: true,
+        telemetry_processor: [buffer_configs: %{metric: %{batch_size: 1}}]
+      )
 
     %{processor: processor_name, ref: ref, bypass: bypass}
   end
@@ -64,7 +54,7 @@ defmodule Sentry.MetricsIntegrationTest do
       assert length(batches) == 3
     end
 
-    test "applies before_send_metric callback", ctx do
+    test "applies before_send_metric callback" do
       put_test_config(
         before_send_metric: fn metric ->
           if metric.value < 10, do: nil, else: metric
@@ -74,15 +64,13 @@ defmodule Sentry.MetricsIntegrationTest do
       Metrics.count("keep.me", 15)
       Metrics.count("drop.me", 5)
 
-      [%{"items" => [metric]}] = collect_sentry_metric_items(ctx.ref, 1)
-      assert_sentry_report(metric, name: "keep.me", value: 15)
+      assert_sentry_metric(:counter, name: "keep.me", value: 15)
 
-      # The dropped metric should not produce an envelope
-      ref = ctx.ref
-      refute_receive {:bypass_envelope, ^ref, _body}, 200
+      # The dropped metric never reaches the collector
+      assert [] == Sentry.Test.pop_sentry_metrics()
     end
 
-    test "callback can modify metrics before sending", ctx do
+    test "callback can modify metrics before sending" do
       put_test_config(
         before_send_metric: fn metric ->
           %{metric | value: metric.value * 2}
@@ -91,8 +79,7 @@ defmodule Sentry.MetricsIntegrationTest do
 
       Metrics.count("test.metric", 5)
 
-      [%{"items" => [metric]}] = collect_sentry_metric_items(ctx.ref, 1)
-      assert_sentry_report(metric, value: 10)
+      assert_sentry_metric(:counter, name: "test.metric", value: 10)
     end
   end
 
@@ -117,32 +104,28 @@ defmodule Sentry.MetricsIntegrationTest do
       assert payload["items"] |> hd() |> Map.get("timestamp")
     end
 
-    test "metrics include environment and release", ctx do
+    test "metrics include environment and release" do
       put_test_config(environment_name: "production", release: "1.0.0")
 
       Metrics.gauge("memory.usage", 1024)
 
-      [%{"items" => [metric]}] = collect_sentry_metric_items(ctx.ref, 1)
-
-      assert_sentry_report(metric,
+      assert_sentry_metric(:gauge,
+        name: "memory.usage",
         attributes: %{
-          :"sentry.environment" => %{value: "production"},
-          :"sentry.release" => %{value: "1.0.0"}
+          "sentry.environment" => "production",
+          "sentry.release" => "1.0.0"
         }
       )
     end
 
-    test "all three metric types are supported", ctx do
+    test "all three metric types are supported" do
       Metrics.count("counter.metric", 1)
       Metrics.gauge("gauge.metric", 100)
       Metrics.distribution("distribution.metric", 3.14)
 
-      batches = collect_sentry_metric_items(ctx.ref, 3)
-      metrics = Enum.flat_map(batches, fn %{"items" => items} -> items end)
-
-      find_sentry_report!(metrics, type: "counter")
-      find_sentry_report!(metrics, type: "gauge")
-      find_sentry_report!(metrics, type: "distribution")
+      assert_sentry_metric(:counter, name: "counter.metric")
+      assert_sentry_metric(:gauge, name: "gauge.metric")
+      assert_sentry_metric(:distribution, name: "distribution.metric")
     end
   end
 end
diff --git a/test/sentry/opentelemetry/span_processor_test.exs b/test/sentry/opentelemetry/span_processor_test.exs
index a5333181..56c6f139 100644
--- a/test/sentry/opentelemetry/span_processor_test.exs
+++ b/test/sentry/opentelemetry/span_processor_test.exs
@@ -489,7 +489,7 @@ defmodule Sentry.Opentelemetry.SpanProcessorTest do
         Process.sleep(1)
       end
 
-      assert_sentry_report(collect_sentry_transactions(ref, 1),
+      assert_sentry_transaction(ref,
         contexts: %{"trace" => %{"op" => "http.server", "description" => "GET /api/users"}}
       )
     end
@@ -509,7 +509,7 @@ defmodule Sentry.Opentelemetry.SpanProcessorTest do
         Process.sleep(1)
       end
 
-      assert_sentry_report(collect_sentry_transactions(ref, 1),
+      assert_sentry_transaction(ref,
         contexts: %{"trace" => %{"op" => "http.server", "description" => "GET"}}
       )
     end
@@ -530,7 +530,7 @@ defmodule Sentry.Opentelemetry.SpanProcessorTest do
         Process.sleep(1)
       end
 
-      assert_sentry_report(collect_sentry_transactions(ref, 1),
+      assert_sentry_transaction(ref,
         contexts: %{"trace" => %{"op" => "http.client", "description" => "GET /external/api"}}
       )
     end
@@ -552,7 +552,7 @@ defmodule Sentry.Opentelemetry.SpanProcessorTest do
         Process.sleep(1)
       end
 
-      assert_sentry_report(collect_sentry_transactions(ref, 1),
+      assert_sentry_transaction(ref,
         contexts: %{
           "trace" => %{
             "op" => "http.server",
@@ -578,7 +578,7 @@ defmodule Sentry.Opentelemetry.SpanProcessorTest do
         Process.sleep(1)
       end
 
-      assert_sentry_report(collect_sentry_transactions(ref, 1),
+      assert_sentry_transaction(ref,
         contexts: %{
           "trace" => %{"op" => "db", "description" => "SELECT * FROM users WHERE id = $1"}
         }
@@ -600,7 +600,7 @@ defmodule Sentry.Opentelemetry.SpanProcessorTest do
         Process.sleep(1)
       end
 
-      assert_sentry_report(collect_sentry_transactions(ref, 1),
+      assert_sentry_transaction(ref,
         contexts: %{"trace" => %{"op" => "db", "description" => nil}}
       )
     end
@@ -621,7 +621,7 @@ defmodule Sentry.Opentelemetry.SpanProcessorTest do
         Process.sleep(1)
       end
 
-      assert_sentry_report(collect_sentry_transactions(ref, 1),
+      assert_sentry_transaction(ref,
         contexts: %{
           "trace" => %{"op" => "queue.process", "description" => "MyApp.Workers.EmailWorker"}
         },
@@ -639,7 +639,7 @@ defmodule Sentry.Opentelemetry.SpanProcessorTest do
         Process.sleep(1)
       end
 
-      assert_sentry_report(collect_sentry_transactions(ref, 1),
+      assert_sentry_transaction(ref,
         contexts: %{"trace" => %{"op" => "custom_operation", "description" => "custom_operation"}}
       )
     end
diff --git a/test/sentry/telemetry_processor_integration_test.exs b/test/sentry/telemetry_processor_integration_test.exs
index 40f296a9..c3f07541 100644
--- a/test/sentry/telemetry_processor_integration_test.exs
+++ b/test/sentry/telemetry_processor_integration_test.exs
@@ -7,30 +7,14 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
   alias Sentry.Telemetry.Buffer
   alias Sentry.{LogEvent, Metric, Transaction}
 
-  setup context do
-    %{bypass: bypass} = setup_bypass()
-    test_pid = self()
-    ref = make_ref()
-
-    Bypass.expect(bypass, "POST", "/api/1/envelope/", fn conn ->
-      {:ok, body, conn} = Plug.Conn.read_body(conn)
-      send(test_pid, {ref, body})
-      Plug.Conn.resp(conn, 200, ~s<{"id": "340"}>)
-    end)
-
-    stop_supervised!(context.telemetry_processor)
-
-    uid = System.unique_integer([:positive])
-    processor_name = :"test_integration_#{uid}"
-
-    start_supervised!(
-      {TelemetryProcessor, name: processor_name, buffer_configs: %{log: %{batch_size: 1}}},
-      id: processor_name
-    )
-
-    Process.put(:sentry_telemetry_processor, processor_name)
+  setup _context do
+    %{bypass: bypass, telemetry_processor: name, ref: ref} =
+      Sentry.Test.setup_sentry(
+        collect_envelopes: true,
+        telemetry_processor: [buffer_configs: %{log: %{batch_size: 1}}]
+      )
 
-    %{processor: processor_name, ref: ref, bypass: bypass}
+    %{processor: name, ref: ref, bypass: bypass}
   end
 
   describe "error events with telemetry_processor_categories" do
@@ -50,11 +34,10 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
 
       :sys.resume(scheduler)
 
-      bodies = collect_envelope_bodies(ctx.ref, 1)
-      assert length(bodies) == 1
+      envelopes = collect_envelopes(ctx.ref, 1, timeout: 2000)
+      assert length(envelopes) == 1
 
-      [items] = Enum.map(bodies, &decode_envelope!/1)
-      assert [{%{"type" => "event"}, event}] = items
+      [[{%{"type" => "event"}, event}]] = envelopes
       assert event["message"]["formatted"] == "integration test error"
     end
 
@@ -77,9 +60,8 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
 
       :sys.resume(scheduler)
 
-      bodies = collect_envelope_bodies(ctx.ref, 5)
-      items = Enum.map(bodies, &decode_envelope!/1)
-      categories = Enum.map(items, &decoded_envelope_category/1)
+      envelopes = collect_envelopes(ctx.ref, 5, timeout: 2000)
+      categories = Enum.map(envelopes, &decoded_envelope_category/1)
 
       error_count = Enum.count(categories, &(&1 == :error))
       assert error_count == 3
@@ -104,10 +86,9 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
 
       assert Buffer.size(error_buffer) == 0
 
-      bodies = collect_envelope_bodies(ctx.ref, 5)
-      items = Enum.map(bodies, &decode_envelope!/1)
-      assert length(items) == 5
-      assert Enum.all?(items, fn [{%{"type" => type}, _}] -> type == "event" end)
+      envelopes = collect_envelopes(ctx.ref, 5, timeout: 2000)
+      assert length(envelopes) == 5
+      assert Enum.all?(envelopes, fn [{%{"type" => type}, _}] -> type == "event" end)
     end
   end
 
@@ -128,11 +109,10 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
 
       :sys.resume(scheduler)
 
-      bodies = collect_envelope_bodies(ctx.ref, 1)
-      assert length(bodies) == 1
+      envelopes = collect_envelopes(ctx.ref, 1, timeout: 2000)
+      assert length(envelopes) == 1
 
-      [items] = Enum.map(bodies, &decode_envelope!/1)
-      assert [{%{"type" => "check_in"}, check_in}] = items
+      [[{%{"type" => "check_in"}, check_in}]] = envelopes
       assert check_in["monitor_slug"] == "test-job"
       assert check_in["status"] == "ok"
     end
@@ -153,10 +133,9 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
 
       assert Buffer.size(check_in_buffer) == 0
 
-      bodies = collect_envelope_bodies(ctx.ref, 3)
-      items = Enum.map(bodies, &decode_envelope!/1)
-      assert length(items) == 3
-      assert Enum.all?(items, fn [{%{"type" => type}, _}] -> type == "check_in" end)
+      envelopes = collect_envelopes(ctx.ref, 3, timeout: 2000)
+      assert length(envelopes) == 3
+      assert Enum.all?(envelopes, fn [{%{"type" => type}, _}] -> type == "check_in" end)
     end
   end
 
@@ -178,11 +157,10 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
 
       :sys.resume(scheduler)
 
-      bodies = collect_envelope_bodies(ctx.ref, 1)
-      assert length(bodies) == 1
+      envelopes = collect_envelopes(ctx.ref, 1, timeout: 2000)
+      assert length(envelopes) == 1
 
-      [items] = Enum.map(bodies, &decode_envelope!/1)
-      assert [{%{"type" => "transaction"}, transaction_data}] = items
+      [[{%{"type" => "transaction"}, transaction_data}]] = envelopes
       assert is_binary(transaction_data["event_id"])
       assert is_number(transaction_data["start_timestamp"])
       assert is_number(transaction_data["timestamp"])
@@ -204,10 +182,9 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
 
       assert Buffer.size(transaction_buffer) == 0
 
-      bodies = collect_envelope_bodies(ctx.ref, 3)
-      items = Enum.map(bodies, &decode_envelope!/1)
-      assert length(items) == 3
-      assert Enum.all?(items, fn [{%{"type" => type}, _}] -> type == "transaction" end)
+      envelopes = collect_envelopes(ctx.ref, 3, timeout: 2000)
+      assert length(envelopes) == 3
+      assert Enum.all?(envelopes, fn [{%{"type" => type}, _}] -> type == "transaction" end)
     end
   end
 
@@ -216,11 +193,10 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
       TelemetryProcessor.add(ctx.processor, make_log_event("log-1"))
       TelemetryProcessor.add(ctx.processor, make_log_event("log-2"))
 
-      bodies = collect_envelope_bodies(ctx.ref, 2)
-      items = Enum.map(bodies, &decode_envelope!/1)
-      assert length(items) == 2
+      envelopes = collect_envelopes(ctx.ref, 2, timeout: 2000)
+      assert length(envelopes) == 2
 
-      for [{header, payload}] <- items do
+      for [{header, payload}] <- envelopes do
         assert header["type"] == "log"
         assert %{"items" => [%{"body" => _}]} = payload
       end
@@ -242,8 +218,8 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
 
       assert Buffer.size(buffer) == 0
 
-      bodies = collect_envelope_bodies(ctx.ref, 3)
-      assert length(bodies) == 3
+      envelopes = collect_envelopes(ctx.ref, 3, timeout: 2000)
+      assert length(envelopes) == 3
     end
 
     test "applies before_send_log callback", ctx do
@@ -256,37 +232,25 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
       TelemetryProcessor.add(ctx.processor, make_log_event("keep me"))
       TelemetryProcessor.add(ctx.processor, make_log_event("drop me"))
 
-      bodies = collect_envelope_bodies(ctx.ref, 1)
-      assert length(bodies) == 1
+      envelopes = collect_envelopes(ctx.ref, 1, timeout: 2000)
+      assert length(envelopes) == 1
 
-      [items] = Enum.map(bodies, &decode_envelope!/1)
-      assert [{%{"type" => "log"}, %{"items" => [%{"body" => "keep me"}]}}] = items
+      [[{%{"type" => "log"}, %{"items" => [%{"body" => "keep me"}]}}]] = envelopes
 
       # The dropped event should not produce an envelope
       ref = ctx.ref
-      refute_receive {^ref, _body}, 200
+      refute_receive {:bypass_envelope, ^ref, _body}, 200
     end
   end
 
   describe "buffer overflow client reports" do
     setup ctx do
-      stop_supervised!(ctx.processor)
-
-      uid = System.unique_integer([:positive])
-      processor_name = :"test_overflow_#{uid}"
-
-      start_supervised!(
-        {TelemetryProcessor,
-         name: processor_name, buffer_configs: %{log: %{capacity: 2, batch_size: 1}}},
-        id: processor_name
-      )
-
-      Process.put(:sentry_telemetry_processor, processor_name)
+      Sentry.Test.setup_telemetry_processor(buffer_configs: %{log: %{capacity: 2, batch_size: 1}})
 
       Sentry.ClientReport.Sender.flush()
       flush_ref_messages(ctx.ref)
 
-      %{processor: processor_name}
+      :ok
     end
 
     test "sends cache_overflow client report when log buffer overflows", ctx do
@@ -303,7 +267,7 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
       Sentry.ClientReport.Sender.flush()
 
       ref = ctx.ref
-      assert_receive {^ref, body}, 2000
+      assert_receive {:bypass_envelope, ^ref, body}, 2000
 
       items = decode_envelope!(body)
       assert [{%{"type" => "client_report"}, client_report}] = items
@@ -449,7 +413,7 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
       Sentry.ClientReport.Sender.flush()
 
       ref = ctx.ref
-      assert_receive {^ref, body}, 2000
+      assert_receive {:bypass_envelope, ^ref, body}, 2000
 
       items = decode_envelope!(body)
       assert [{%{"type" => "client_report"}, client_report}] = items
@@ -476,7 +440,7 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
       Sentry.ClientReport.Sender.flush()
 
       ref = ctx.ref
-      assert_receive {^ref, body}, 2000
+      assert_receive {:bypass_envelope, ^ref, body}, 2000
 
       items = decode_envelope!(body)
       assert [{%{"type" => "client_report"}, client_report}] = items
@@ -548,7 +512,7 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
 
   defp flush_ref_messages(ref) do
     receive do
-      {^ref, _body} -> flush_ref_messages(ref)
+      {:bypass_envelope, ^ref, _body} -> flush_ref_messages(ref)
     after
       100 -> :ok
     end
@@ -562,20 +526,6 @@ defmodule Sentry.TelemetryProcessorIntegrationTest do
     }
   end
 
-  defp collect_envelope_bodies(ref, expected_count) do
-    collect_envelope_bodies(ref, expected_count, [])
-  end
-
-  defp collect_envelope_bodies(_ref, 0, acc), do: Enum.reverse(acc)
-
-  defp collect_envelope_bodies(ref, remaining, acc) do
-    receive do
-      {^ref, body} -> collect_envelope_bodies(ref, remaining - 1, [body | acc])
-    after
-      2000 -> Enum.reverse(acc)
-    end
-  end
-
   defp decoded_envelope_category([{%{"type" => "event"}, _} | _]), do: :error
   defp decoded_envelope_category([{%{"type" => "check_in"}, _} | _]), do: :check_in
   defp decoded_envelope_category([{%{"type" => "transaction"}, _} | _]), do: :transaction
diff --git a/test/sentry/test/assertions_test.exs b/test/sentry/test/assertions_test.exs
index 8895d0b3..9c0c573e 100644
--- a/test/sentry/test/assertions_test.exs
+++ b/test/sentry/test/assertions_test.exs
@@ -388,6 +388,45 @@ defmodule Sentry.Test.AssertionsTest do
     end
   end
 
+  describe "refute_sentry_check_in/2 (message-level)" do
+    test "returns :ok when no envelope is received for the ref" do
+      assert :ok = refute_sentry_check_in(make_ref(), 10)
+    end
+
+    test "ignores envelopes addressed to a different ref" do
+      send(self(), {:bypass_envelope, make_ref(), "unrelated"})
+
+      assert :ok = refute_sentry_check_in(make_ref(), 10)
+    end
+
+    test "flunks when an envelope is received for the ref" do
+      ref = make_ref()
+      send(self(), {:bypass_envelope, ref, ~s({"type":"check_in"})})
+
+      assert_raise ExUnit.AssertionError, ~r/expected no check-in/, fn ->
+        refute_sentry_check_in(ref, 10)
+      end
+    end
+  end
+
+  describe "refute_sentry_check_in/2 with real pipeline" do
+    setup do
+      SentryTest.setup_sentry(collect_envelopes: [type: "check_in"])
+    end
+
+    test "returns :ok when no check-in was captured", %{ref: ref} do
+      assert :ok = refute_sentry_check_in(ref, 50)
+    end
+
+    test "flunks (after flushing the pipeline) when a check-in was captured", %{ref: ref} do
+      {:ok, _id} = Sentry.capture_check_in(status: :ok, monitor_slug: "refute-test")
+
+      assert_raise ExUnit.AssertionError, ~r/expected no check-in/, fn ->
+        refute_sentry_check_in(ref)
+      end
+    end
+  end
+
   # Test helpers — direct ETS insertion for isolation
 
   defp msg(text) do
diff --git a/test/sentry/test_test.exs b/test/sentry/test_test.exs
index cbf2b188..2cf1147c 100644
--- a/test/sentry/test_test.exs
+++ b/test/sentry/test_test.exs
@@ -27,6 +27,70 @@ defmodule Sentry.TestTest do
 
       assert Sentry.Test.Registry.lookup_processor_for(scheduler_pid) == processor_name
     end
+
+    test ":telemetry_processor option configures the per-test processor" do
+      %{telemetry_processor: name} =
+        SentryTest.setup_sentry(telemetry_processor: [buffer_configs: %{log: %{batch_size: 1}}])
+
+      log_buffer = Sentry.TelemetryProcessor.get_buffer(name, :log)
+
+      assert :sys.get_state(log_buffer).batch_size == 1
+    end
+
+    test ":telemetry_processor option coexists with sibling config options" do
+      SentryTest.setup_sentry(
+        dedup_events: false,
+        telemetry_processor: [buffer_configs: %{log: %{batch_size: 1}}]
+      )
+
+      assert Sentry.Config.dedup_events?() == false
+    end
+
+    test "re-tags the scheduler after restarting with :telemetry_processor opts" do
+      %{telemetry_processor: name} =
+        SentryTest.setup_sentry(telemetry_processor: [buffer_configs: %{log: %{batch_size: 1}}])
+
+      scheduler_pid = Sentry.TelemetryProcessor.get_scheduler(name)
+
+      assert is_pid(scheduler_pid)
+      assert Sentry.Test.Registry.lookup_processor_for(scheduler_pid) == name
+    end
+
+    test "does not return a :ref by default" do
+      refute Map.has_key?(SentryTest.setup_sentry(), :ref)
+    end
+
+    test "collect_envelopes: true returns a ref that captures sent envelopes" do
+      %{ref: ref} = SentryTest.setup_sentry(collect_envelopes: true)
+
+      assert {:ok, _} = Sentry.capture_message("collected", result: :sync)
+
+      assert [[{%{"type" => "event"}, event}]] = SentryTest.collect_envelopes(ref, 1)
+      assert event["message"]["formatted"] == "collected"
+    end
+
+    test "collect_envelopes accepts collector options forwarded to the collector" do
+      %{ref: ref} = SentryTest.setup_sentry(collect_envelopes: [type: "event"])
+
+      assert is_reference(ref)
+      assert {:ok, _} = Sentry.capture_message("typed", result: :sync)
+      assert [[{%{"type" => "event"}, _}]] = SentryTest.collect_envelopes(ref, 1)
+    end
+
+    test "collect_envelopes coexists with :telemetry_processor and config options" do
+      %{ref: ref, telemetry_processor: name} =
+        SentryTest.setup_sentry(
+          dedup_events: false,
+          collect_envelopes: true,
+          telemetry_processor: [buffer_configs: %{log: %{batch_size: 1}}]
+        )
+
+      assert is_reference(ref)
+      assert Sentry.Config.dedup_events?() == false
+
+      log_buffer = Sentry.TelemetryProcessor.get_buffer(name, :log)
+      assert :sys.get_state(log_buffer).batch_size == 1
+    end
   end
 
   describe "start_collecting_sentry_reports/0" do
diff --git a/test_integrations/phoenix_app/test/phoenix_app/oban_test.exs b/test_integrations/phoenix_app/test/phoenix_app/oban_test.exs
index 2f633ba6..92e032ca 100644
--- a/test_integrations/phoenix_app/test/phoenix_app/oban_test.exs
+++ b/test_integrations/phoenix_app/test/phoenix_app/oban_test.exs
@@ -3,6 +3,7 @@ defmodule Sentry.Integrations.Phoenix.ObanTest do
   use Oban.Testing, repo: PhoenixApp.Repo
 
   import ExUnit.CaptureLog
+  import Sentry.Test.Assertions
   import Sentry.TestHelpers
 
   require OpenTelemetry.Tracer
@@ -10,10 +11,7 @@ defmodule Sentry.Integrations.Phoenix.ObanTest do
   alias Sentry.Integrations.Oban.ErrorReporter
 
   setup do
-    %{bypass: bypass} = setup_bypass(traces_sample_rate: 1.0)
-    ref = setup_bypass_envelope_collector(bypass)
-
-    %{bypass: bypass, ref: ref}
+    Sentry.Test.setup_sentry(collect_envelopes: true, traces_sample_rate: 1.0)
   end
 
   defmodule TestWorker do
@@ -47,43 +45,29 @@ defmodule Sentry.Integrations.Phoenix.ObanTest do
     end
   end
 
-  test "captures Oban worker execution as transaction", %{ref: ref} do
+  test "captures Oban worker execution as transaction" do
     :ok = perform_job(TestWorker, %{test: "args"})
 
-    envelopes = collect_envelopes(ref, 1)
-    transactions = extract_transactions(envelopes)
-    assert length(transactions) == 1
-
-    [tx] = transactions
+    tx =
+      assert_sentry_report(:transaction,
+        transaction: "Sentry.Integrations.Phoenix.ObanTest.TestWorker",
+        transaction_info: %{source: :custom}
+      )
 
-    assert tx["transaction"] == "Sentry.Integrations.Phoenix.ObanTest.TestWorker"
-    assert tx["transaction_info"] == %{"source" => "custom"}
+    trace = tx.contexts.trace
+    assert trace.origin == "opentelemetry_oban"
+    assert trace.op == "queue.process"
+    assert trace.description == "Sentry.Integrations.Phoenix.ObanTest.TestWorker"
+    assert trace.data["oban.job.job_id"]
+    assert trace.data["messaging.destination.name"] == "default"
+    assert trace.data["oban.job.attempt"] == 1
 
-    trace = tx["contexts"]["trace"]
-    assert trace["origin"] == "opentelemetry_oban"
-    assert trace["op"] == "queue.process"
-    assert trace["description"] == "Sentry.Integrations.Phoenix.ObanTest.TestWorker"
-    assert trace["data"]["oban.job.job_id"]
-    assert trace["data"]["messaging.destination.name"] == "default"
-    assert trace["data"]["oban.job.attempt"] == 1
-
-    assert tx["spans"] == []
+    assert tx.spans == []
   end
 
-  test "captures Oban worker with trace links", %{bypass: bypass} do
+  test "captures Oban worker with trace links", %{ref: ref} do
     # This test verifies that when an Oban job is inserted within an active trace,
     # the consumer span has links back to the producer span.
-    test_pid = self()
-
-    Bypass.stub(bypass, "POST", "/api/1/envelope/", fn conn ->
-      {:ok, body, conn} = Plug.Conn.read_body(conn)
-
-      for {headers, body_json} <- decode_envelope!(body) do
-        send(test_pid, {headers["type"], body_json})
-      end
-
-      Plug.Conn.send_resp(conn, 200, ~s<{"id": "340"}>)
-    end)
 
     # Insert within an active span so trace context is propagated into job metadata
     OpenTelemetry.Tracer.with_span "test.request" do
@@ -96,9 +80,10 @@ defmodule Sentry.Integrations.Phoenix.ObanTest do
     # Drain the queue to execute the job
     Oban.drain_queue(queue: :default)
 
-    # Multiple transactions are sent (test.request, DB queries, Oban consumer),
-    # so we need to find the Oban consumer transaction specifically
-    oban_tx = receive_oban_transaction()
+    transactions = collect_sentry_transactions(ref, 100, timeout: 500)
+
+    oban_tx =
+      find_sentry_report!(transactions, contexts: %{trace: %{origin: "opentelemetry_oban"}})
 
     trace = oban_tx["contexts"]["trace"]
     assert trace["op"] == "queue.process"
@@ -116,70 +101,34 @@ defmodule Sentry.Integrations.Phoenix.ObanTest do
     end
   end
 
-  test "captures Oban worker with child spans", %{ref: ref} do
+  test "captures Oban worker with child spans" do
     :ok = perform_job(WorkerWithDatabaseQuery, %{})
 
-    envelopes = collect_envelopes(ref, 1)
-    transactions = extract_transactions(envelopes)
-    assert length(transactions) == 1
-
-    [tx] = transactions
-
-    assert tx["transaction"] ==
-             "Sentry.Integrations.Phoenix.ObanTest.WorkerWithDatabaseQuery"
+    tx =
+      assert_sentry_report(:transaction,
+        transaction: "Sentry.Integrations.Phoenix.ObanTest.WorkerWithDatabaseQuery"
+      )
 
-    # Should have child spans from the database query
-    assert length(tx["spans"]) > 0
+    assert length(tx.spans) > 0
 
-    # Verify at least one db span exists
-    assert Enum.any?(tx["spans"], fn span ->
-             span["op"] == "db"
-           end)
-  end
-
-  defp receive_oban_transaction do
-    receive do
-      {"transaction", tx} ->
-        if tx["contexts"]["trace"]["origin"] == "opentelemetry_oban" do
-          tx
-        else
-          receive_oban_transaction()
-        end
-    after
-      2000 -> flunk("expected an Oban consumer transaction")
-    end
+    assert Enum.any?(tx.spans, fn span -> span.op == "db" end)
   end
 
   describe "should_report_error_callback config" do
-    setup %{bypass: bypass} do
+    setup do
       :telemetry.detach(ErrorReporter)
 
-      ref = setup_bypass_envelope_collector(bypass, type: "event")
-
       on_exit(fn ->
         _ = :telemetry.detach(ErrorReporter)
         ErrorReporter.attach([])
       end)
 
-        %{ref: ref}
-      end
+      :ok
+    end
 
-    test "skips error reporting when callback returns false", %{bypass: bypass, ref: ref} do
+    test "skips error reporting when callback returns false" do
       test_pid = self()
 
-      # Allow transaction envelopes through but assert no error events are sent
-      Bypass.stub(bypass, "POST", "/api/1/envelope/", fn conn ->
-        {:ok, body, conn} = Plug.Conn.read_body(conn)
-        items = decode_envelope!(body)
-
-        for {headers, _body} <- items do
-          assert headers["type"] != "event",
-                 "Should not send error events when callback returns false"
-        end
-
-        Plug.Conn.send_resp(conn, 200, ~s<{"id": "340"}>)
-      end)
-
       ErrorReporter.attach(
         should_report_error_callback: fn worker, job ->
           send(test_pid, {:callback_invoked, worker, job})
@@ -199,11 +148,10 @@ defmodule Sentry.Integrations.Phoenix.ObanTest do
       assert %Oban.Job{} = received_job
       assert received_job.args == %{"should_fail" => true}
 
-      envelopes = collect_envelopes(ref, 10, timeout: 500)
-      assert extract_events(envelopes) == []
+      assert [] == Sentry.Test.pop_sentry_reports()
     end
 
-    test "reports error when callback returns true", %{ref: ref} do
+    test "reports error when callback returns true" do
       test_pid = self()
 
       ErrorReporter.attach(
@@ -222,18 +170,22 @@ defmodule Sentry.Integrations.Phoenix.ObanTest do
 
       assert_receive {:callback_invoked, _worker, _job}
 
-      envelopes = collect_envelopes(ref, 10, timeout: 500)
-      events = extract_events(envelopes)
-      assert [event] = events
-
-      assert [%{"type" => "RuntimeError", "value" => "intentional failure for testing"} | _] =
-               event["exception"]
+      event =
+        find_sentry_report!(Sentry.Test.pop_sentry_reports(),
+          tags: %{oban_worker: "Sentry.Integrations.Phoenix.ObanTest.FailingWorker"}
+        )
 
-      assert event["tags"]["oban_worker"] ==
-               "Sentry.Integrations.Phoenix.ObanTest.FailingWorker"
+      assert [
+               %Sentry.Interfaces.Exception{
+                 type: "RuntimeError",
+                 value: "intentional failure for testing"
+               }
+               | _
+             ] =
+               event.exception
     end
 
-    test "callback receives worker module and full job struct", %{ref: ref} do
+    test "callback receives worker module and full job struct" do
       test_pid = self()
 
       ErrorReporter.attach(
@@ -262,27 +214,11 @@ defmodule Sentry.Integrations.Phoenix.ObanTest do
       assert job.max_attempts == 3
       assert is_integer(job.attempt)
       assert is_integer(job.id)
-
-      # Drain envelopes to avoid Bypass errors
-      collect_envelopes(ref, 10, timeout: 500)
     end
 
-    test "callback can make decisions based on attempt number", %{bypass: bypass, ref: ref} do
+    test "callback can make decisions based on attempt number" do
       test_pid = self()
 
-      # Allow transaction envelopes through but assert no error events are sent
-      Bypass.stub(bypass, "POST", "/api/1/envelope/", fn conn ->
-        {:ok, body, conn} = Plug.Conn.read_body(conn)
-        items = decode_envelope!(body)
-
-        for {headers, _body} <- items do
-          assert headers["type"] != "event",
-                 "Should not send error events when callback returns false"
-        end
-
-        Plug.Conn.send_resp(conn, 200, ~s<{"id": "340"}>)
-      end)
-
       ErrorReporter.attach(
         should_report_error_callback: fn _worker, job ->
           should_report = job.attempt >= job.max_attempts
@@ -303,11 +239,10 @@ defmodule Sentry.Integrations.Phoenix.ObanTest do
       assert max_attempts == 3
       assert should_report == false
 
-      envelopes = collect_envelopes(ref, 10, timeout: 500)
-      assert extract_events(envelopes) == []
+      assert [] == Sentry.Test.pop_sentry_reports()
     end
 
-    test "handles callback errors gracefully and defaults to reporting", %{ref: ref} do
+    test "handles callback errors gracefully and defaults to reporting" do
       log =
         capture_log(fn ->
           ErrorReporter.attach(
@@ -328,15 +263,22 @@ defmodule Sentry.Integrations.Phoenix.ObanTest do
       assert log =~ "FailingWorker"
       assert log =~ "callback crashed!"
 
-      envelopes = collect_envelopes(ref, 10, timeout: 500)
-      events = extract_events(envelopes)
-      assert [event] = events
+      event =
+        find_sentry_report!(Sentry.Test.pop_sentry_reports(),
+          tags: %{oban_worker: "Sentry.Integrations.Phoenix.ObanTest.FailingWorker"}
+        )
 
-      assert [%{"type" => "RuntimeError", "value" => "intentional failure for testing"} | _] =
-               event["exception"]
+      assert [
+               %Sentry.Interfaces.Exception{
+                 type: "RuntimeError",
+                 value: "intentional failure for testing"
+               }
+               | _
+             ] =
+               event.exception
     end
 
-    test "reports error when no callback is configured", %{ref: ref} do
+    test "reports error when no callback is configured" do
       ErrorReporter.attach([])
 
       {:ok, _job} =
@@ -346,30 +288,24 @@ defmodule Sentry.Integrations.Phoenix.ObanTest do
 
       Oban.drain_queue(queue: :default)
 
-      envelopes = collect_envelopes(ref, 10, timeout: 500)
-      events = extract_events(envelopes)
-      assert [event] = events
+      event =
+        find_sentry_report!(Sentry.Test.pop_sentry_reports(),
+          tags: %{oban_worker: "Sentry.Integrations.Phoenix.ObanTest.FailingWorker"}
+        )
 
-      assert [%{"type" => "RuntimeError", "value" => "intentional failure for testing"} | _] =
-               event["exception"]
+      assert [
+               %Sentry.Interfaces.Exception{
+                 type: "RuntimeError",
+                 value: "intentional failure for testing"
+               }
+               | _
+             ] =
+               event.exception
     end
 
-    test "callback can filter based on worker type", %{bypass: bypass, ref: ref} do
+    test "callback can filter based on worker type" do
       test_pid = self()
 
-      # Allow transaction envelopes through but assert no error events are sent
-      Bypass.stub(bypass, "POST", "/api/1/envelope/", fn conn ->
-        {:ok, body, conn} = Plug.Conn.read_body(conn)
-        items = decode_envelope!(body)
-
-        for {headers, _body} <- items do
-          assert headers["type"] != "event",
-                 "Should not send error events when callback returns false"
-        end
-
-        Plug.Conn.send_resp(conn, 200, ~s<{"id": "340"}>)
-      end)
-
       ErrorReporter.attach(
         should_report_error_callback: fn worker, _job ->
           should_report = worker != FailingWorker
@@ -387,11 +323,10 @@ defmodule Sentry.Integrations.Phoenix.ObanTest do
 
       assert_receive {:worker_check, FailingWorker, false}
 
-      envelopes = collect_envelopes(ref, 10, timeout: 500)
-      assert extract_events(envelopes) == []
+      assert [] == Sentry.Test.pop_sentry_reports()
     end
 
-    test "callback receives nil and logs warning for non-existent worker module", %{ref: ref} do
+    test "callback receives nil and logs warning for non-existent worker module" do
       test_pid = self()
 
       log =
@@ -432,10 +367,12 @@ defmodule Sentry.Integrations.Phoenix.ObanTest do
       assert worker == nil
       assert received_job.worker == "NonExistent.Worker.Module"
 
-      envelopes = collect_envelopes(ref, 1)
-      events = extract_events(envelopes)
-      assert [event] = events
-      assert event["tags"]["oban_worker"] == "NonExistent.Worker.Module"
+      event =
+        assert_sentry_report(:event,
+          tags: %{oban_worker: "NonExistent.Worker.Module"}
+        )
+
+      assert event.tags[:oban_worker] == "NonExistent.Worker.Module"
     end
   end
 
diff --git a/test_integrations/phoenix_app/test/phoenix_app/repo_test.exs b/test_integrations/phoenix_app/test/phoenix_app/repo_test.exs
index 97a05f51..f885b3d8 100644
--- a/test_integrations/phoenix_app/test/phoenix_app/repo_test.exs
+++ b/test_integrations/phoenix_app/test/phoenix_app/repo_test.exs
@@ -4,30 +4,36 @@ defmodule PhoenixApp.RepoTest do
   alias PhoenixApp.{Repo, Accounts.User}
 
   import Sentry.TestHelpers
+  import Sentry.Test.Assertions
 
   setup do
-    setup_bypass(traces_sample_rate: 1.0)
+    Sentry.Test.setup_sentry(collect_envelopes: true, traces_sample_rate: 1.0)
   end
 
-  test "instrumented top-level ecto transaction span", %{bypass: bypass} do
-    ref = setup_bypass_envelope_collector(bypass)
-
+  test "instrumented top-level ecto transaction span", %{ref: ref} do
     Repo.all(User) |> Enum.map(& &1.id)
 
-    assert [tx] = collect_sentry_transactions(ref, 1)
-
-    assert tx["transaction_info"] == %{"source" => "custom"}
-
-    assert tx["contexts"]["trace"]["op"] == "db"
-
-    assert tx["contexts"]["trace"]["data"]["db.system"] == "sqlite"
-    assert tx["contexts"]["trace"]["data"]["db.type"] == "sql"
-    assert tx["contexts"]["trace"]["data"]["db.instance"] == "db/test.sqlite3"
-    assert tx["contexts"]["trace"]["data"]["db.name"] == "db/test.sqlite3"
-
-    assert String.starts_with?(tx["contexts"]["trace"]["description"], "SELECT")
-    assert String.starts_with?(tx["contexts"]["trace"]["data"]["db.statement"], "SELECT")
-
-    refute Map.has_key?(tx["contexts"]["trace"]["data"], "db.url")
+    tx =
+      assert_sentry_transaction(ref,
+        transaction_info: %{"source" => "custom"},
+        contexts: %{
+          trace: %{
+            op: "db",
+            data: %{
+              "db.system" => "sqlite",
+              "db.type" => "sql",
+              "db.instance" => "db/test.sqlite3",
+              "db.name" => "db/test.sqlite3"
+            }
+          }
+        }
+      )
+
+    trace = tx["contexts"]["trace"]
+
+    assert String.starts_with?(trace["description"], "SELECT")
+    assert String.starts_with?(trace["data"]["db.statement"], "SELECT")
+
+    refute Map.has_key?(trace["data"], "db.url")
   end
 end
diff --git a/test_integrations/phoenix_app/test/phoenix_app_web/controllers/transaction_test.exs b/test_integrations/phoenix_app/test/phoenix_app_web/controllers/transaction_test.exs
index 23640c78..83a50ca4 100644
--- a/test_integrations/phoenix_app/test/phoenix_app_web/controllers/transaction_test.exs
+++ b/test_integrations/phoenix_app/test/phoenix_app_web/controllers/transaction_test.exs
@@ -3,49 +3,39 @@ defmodule Sentry.Integrations.Phoenix.TransactionTest do
 
   import Phoenix.LiveViewTest
   import Sentry.TestHelpers
+  import Sentry.Test.Assertions
 
   setup do
-    setup_bypass(traces_sample_rate: 1.0)
+    Sentry.Test.setup_sentry(collect_envelopes: true, traces_sample_rate: 1.0)
   end
 
-  test "GET /transaction", %{conn: conn, bypass: bypass} do
-    ref = setup_bypass_envelope_collector(bypass)
-
+  test "GET /transaction", %{conn: conn, ref: ref} do
     get(conn, ~p"/transaction")
 
-    transactions = collect_envelopes(ref, 1) |> extract_transactions()
-
-    assert length(transactions) == 1
-
-    assert [tx] = transactions
-
-    assert tx["transaction"] == "test_span"
-    assert tx["transaction_info"] == %{"source" => "custom"}
+    tx =
+      assert_sentry_transaction(ref,
+        transaction: "test_span",
+        transaction_info: %{"source" => "custom"},
+        contexts: %{trace: %{origin: "phoenix_app", op: "test_span"}}
+      )
 
-    trace = tx["contexts"]["trace"]
-    assert trace["origin"] == "phoenix_app"
-    assert trace["op"] == "test_span"
-    assert trace["data"] == %{}
+    assert tx["contexts"]["trace"]["data"] == %{}
   end
 
-  test "GET /users", %{conn: conn, bypass: bypass} do
-    ref = setup_bypass_envelope_collector(bypass)
-
+  test "GET /users", %{conn: conn, ref: ref} do
     get(conn, ~p"/users")
 
-    transactions = collect_envelopes(ref, 2) |> extract_transactions()
-
-    assert length(transactions) == 2
-
-    assert [mount_tx, handle_params_tx] = transactions
+    assert [mount_tx, handle_params_tx] = collect_sentry_transactions(ref, 2)
 
-    assert mount_tx["transaction"] == "PhoenixAppWeb.UserLive.Index.mount"
-    assert mount_tx["transaction_info"] == %{"source" => "custom"}
+    assert_sentry_report(mount_tx,
+      transaction: "PhoenixAppWeb.UserLive.Index.mount",
+      transaction_info: %{"source" => "custom"},
+      contexts: %{
+        trace: %{origin: "opentelemetry_phoenix", op: "PhoenixAppWeb.UserLive.Index.mount"}
+      }
+    )
 
-    trace = mount_tx["contexts"]["trace"]
-    assert trace["origin"] == "opentelemetry_phoenix"
-    assert trace["op"] == "PhoenixAppWeb.UserLive.Index.mount"
-    assert trace["data"] == %{}
+    assert mount_tx["contexts"]["trace"]["data"] == %{}
 
     assert [span_ecto] = mount_tx["spans"]
 
@@ -54,33 +44,29 @@ defmodule Sentry.Integrations.Phoenix.TransactionTest do
     assert span_ecto["description"] ==
              "SELECT u0.\"id\", u0.\"name\", u0.\"age\", u0.\"inserted_at\", u0.\"updated_at\" FROM \"users\" AS u0"
 
-    assert handle_params_tx["transaction"] ==
-             "PhoenixAppWeb.UserLive.Index.handle_params"
-
-    assert handle_params_tx["transaction_info"] == %{"source" => "custom"}
-
-    trace = handle_params_tx["contexts"]["trace"]
-    assert trace["origin"] == "opentelemetry_phoenix"
-    assert trace["op"] == "PhoenixAppWeb.UserLive.Index.handle_params"
-    assert trace["data"] == %{}
+    assert_sentry_report(handle_params_tx,
+      transaction: "PhoenixAppWeb.UserLive.Index.handle_params",
+      transaction_info: %{"source" => "custom"},
+      contexts: %{
+        trace: %{
+          origin: "opentelemetry_phoenix",
+          op: "PhoenixAppWeb.UserLive.Index.handle_params"
+        }
+      }
+    )
+
+    assert handle_params_tx["contexts"]["trace"]["data"] == %{}
   end
 
-  test "GET /nested-spans includes grand-child spans", %{conn: conn, bypass: bypass} do
-    ref = setup_bypass_envelope_collector(bypass)
-
+  test "GET /nested-spans includes grand-child spans", %{conn: conn, ref: ref} do
     get(conn, ~p"/nested-spans")
 
-    transactions = collect_envelopes(ref, 1) |> extract_transactions()
-
-    assert length(transactions) == 1
-    assert [tx] = transactions
-
-    assert tx["transaction"] == "root_span"
-    assert tx["transaction_info"] == %{"source" => "custom"}
-
-    trace = tx["contexts"]["trace"]
-    assert trace["origin"] == "phoenix_app"
-    assert trace["op"] == "root_span"
+    tx =
+      assert_sentry_transaction(ref,
+        transaction: "root_span",
+        transaction_info: %{"source" => "custom"},
+        contexts: %{trace: %{origin: "phoenix_app", op: "root_span"}}
+      )
 
     assert length(tx["spans"]) == 6
 
@@ -115,16 +101,11 @@ defmodule Sentry.Integrations.Phoenix.TransactionTest do
 
   test "LiveView mount and handle_params create disconnected transactions with child spans", %{
     conn: conn,
-    bypass: bypass
+    ref: ref
   } do
-    ref = setup_bypass_envelope_collector(bypass)
-
     get(conn, ~p"/users")
 
-    transactions = collect_envelopes(ref, 2) |> extract_transactions()
-
-    assert length(transactions) == 2
-    assert [mount_tx, handle_params_tx] = transactions
+    assert [mount_tx, handle_params_tx] = collect_sentry_transactions(ref, 2)
 
     assert mount_tx["transaction"] == "PhoenixAppWeb.UserLive.Index.mount"
     assert length(mount_tx["spans"]) == 1
@@ -145,10 +126,8 @@ defmodule Sentry.Integrations.Phoenix.TransactionTest do
   describe "distributed tracing with sentry-trace header" do
     test "LiveView mount inherits trace context from sentry-trace header", %{
       conn: conn,
-      bypass: bypass
+      ref: ref
     } do
-      ref = setup_bypass_envelope_collector(bypass)
-
       trace_id = "1234567890abcdef1234567890abcdef"
       parent_span_id = "abcdef1234567890"
 
@@ -158,34 +137,23 @@ defmodule Sentry.Integrations.Phoenix.TransactionTest do
 
       get(conn, ~p"/users")
 
-      transactions = collect_envelopes(ref, 2) |> extract_transactions()
-
-      mount_tx =
-        Enum.find(transactions, fn t ->
-          t["transaction"] == "PhoenixAppWeb.UserLive.Index.mount"
-        end)
+      transactions = collect_sentry_transactions(ref, 2)
 
-      handle_params_tx =
-        Enum.find(transactions, fn t ->
-          t["transaction"] == "PhoenixAppWeb.UserLive.Index.handle_params"
-        end)
+      find_sentry_report!(transactions,
+        transaction: "PhoenixAppWeb.UserLive.Index.mount",
+        contexts: %{trace: %{trace_id: trace_id, parent_span_id: parent_span_id}}
+      )
 
-      assert mount_tx != nil
-      assert handle_params_tx != nil
-
-      assert mount_tx["contexts"]["trace"]["trace_id"] == trace_id
-      assert handle_params_tx["contexts"]["trace"]["trace_id"] == trace_id
-
-      assert mount_tx["contexts"]["trace"]["parent_span_id"] == parent_span_id
-      assert handle_params_tx["contexts"]["trace"]["parent_span_id"] == parent_span_id
+      find_sentry_report!(transactions,
+        transaction: "PhoenixAppWeb.UserLive.Index.handle_params",
+        contexts: %{trace: %{trace_id: trace_id, parent_span_id: parent_span_id}}
+      )
     end
 
     test "LiveView handle_event in WebSocket shares trace context with initial request", %{
       conn: conn,
-      bypass: bypass
+      ref: ref
     } do
-      ref = setup_bypass_envelope_collector(bypass)
-
       trace_id = "fedcba0987654321fedcba0987654321"
       parent_span_id = "1234567890fedcba"
 
@@ -197,23 +165,15 @@ defmodule Sentry.Integrations.Phoenix.TransactionTest do
 
       view |> element("#increment-btn") |> render_click()
 
-      transactions = collect_envelopes(ref, 5, timeout: 2000) |> extract_transactions()
-
-      handle_event_tx =
-        Enum.find(transactions, fn t ->
-          String.contains?(t["transaction"], "handle_event#increment")
-        end)
-
-      assert handle_event_tx != nil,
-             "Expected handle_event transaction, got: #{inspect(Enum.map(transactions, & &1["transaction"]))}"
-
-      assert handle_event_tx["contexts"]["trace"]["trace_id"] == trace_id,
-             "Expected trace_id #{trace_id}, got #{handle_event_tx["contexts"]["trace"]["trace_id"]}"
+      find_sentry_transaction!(ref,
+        count: 5,
+        timeout: 2000,
+        transaction: ~r/handle_event#increment/,
+        contexts: %{trace: %{trace_id: trace_id}}
+      )
     end
 
-    test "baggage header is preserved through LiveView lifecycle", %{conn: conn, bypass: bypass} do
-      ref = setup_bypass_envelope_collector(bypass)
-
+    test "baggage header is preserved through LiveView lifecycle", %{conn: conn, ref: ref} do
       trace_id = "abababababababababababababababab"
       parent_span_id = "cdcdcdcdcdcdcdcd"
       baggage = "sentry-environment=production,sentry-release=1.0.0"
@@ -225,15 +185,11 @@ defmodule Sentry.Integrations.Phoenix.TransactionTest do
 
       get(conn, ~p"/users")
 
-      transactions = collect_envelopes(ref, 2) |> extract_transactions()
-
-      mount_tx =
-        Enum.find(transactions, fn t ->
-          t["transaction"] == "PhoenixAppWeb.UserLive.Index.mount"
-        end)
-
-      assert mount_tx != nil
-      assert mount_tx["contexts"]["trace"]["trace_id"] == trace_id
+      find_sentry_transaction!(ref,
+        count: 2,
+        transaction: "PhoenixAppWeb.UserLive.Index.mount",
+        contexts: %{trace: %{trace_id: trace_id}}
+      )
     end
   end
 end
diff --git a/test_integrations/phoenix_app/test/phoenix_app_web/live/user_live_test.exs b/test_integrations/phoenix_app/test/phoenix_app_web/live/user_live_test.exs
index df608453..5dc5cbe8 100644
--- a/test_integrations/phoenix_app/test/phoenix_app_web/live/user_live_test.exs
+++ b/test_integrations/phoenix_app/test/phoenix_app_web/live/user_live_test.exs
@@ -2,6 +2,7 @@ defmodule PhoenixAppWeb.UserLiveTest do
   use PhoenixAppWeb.ConnCase, async: false
 
   import Sentry.TestHelpers
+  import Sentry.Test.Assertions
   import Phoenix.LiveViewTest
   import PhoenixApp.AccountsFixtures
 
@@ -10,7 +11,7 @@ defmodule PhoenixAppWeb.UserLiveTest do
   @invalid_attrs %{name: nil, age: nil}
 
   setup do
-    setup_bypass(traces_sample_rate: 1.0)
+    Sentry.Test.setup_sentry(collect_envelopes: true, traces_sample_rate: 1.0)
   end
 
   defp create_user(_) do
@@ -28,9 +29,7 @@ defmodule PhoenixAppWeb.UserLiveTest do
       assert html =~ user.name
     end
 
-    test "saves new user", %{conn: conn, bypass: bypass} do
-      ref = setup_bypass_envelope_collector(bypass)
-
+    test "saves new user", %{conn: conn, ref: ref} do
       {:ok, index_live, _html} = live(conn, ~p"/users")
 
       assert index_live |> element("a", "New User") |> render_click() =~
@@ -52,21 +51,19 @@ defmodule PhoenixAppWeb.UserLiveTest do
       assert html =~ "User created successfully"
       assert html =~ "some name"
 
-      transactions = collect_envelopes(ref, 10, timeout: 2000) |> extract_transactions()
-
       transaction_save =
-        Enum.find(transactions, fn tx ->
-          tx["transaction"] == "PhoenixAppWeb.UserLive.Index.handle_event#save"
-        end)
-
-      assert transaction_save != nil
-      assert transaction_save["transaction"] == "PhoenixAppWeb.UserLive.Index.handle_event#save"
-      assert transaction_save["transaction_info"]["source"] == "custom"
-
-      assert transaction_save["contexts"]["trace"]["op"] ==
-               "PhoenixAppWeb.UserLive.Index.handle_event#save"
-
-      assert transaction_save["contexts"]["trace"]["origin"] == "opentelemetry_phoenix"
+        find_sentry_transaction!(ref,
+          count: 10,
+          timeout: 2000,
+          transaction: "PhoenixAppWeb.UserLive.Index.handle_event#save",
+          transaction_info: %{"source" => "custom"},
+          contexts: %{
+            trace: %{
+              op: "PhoenixAppWeb.UserLive.Index.handle_event#save",
+              origin: "opentelemetry_phoenix"
+            }
+          }
+        )
 
       assert length(transaction_save["spans"]) == 1
       assert [span] = transaction_save["spans"]