diff --git a/README.md b/README.md
index fda26a9e..42027fcb 100644
--- a/README.md
+++ b/README.md
@@ -139,6 +139,7 @@ The [`Turbo::TestAssertions`](./lib/turbo/test_assertions.rb) concern provides T
 
 The [`Turbo::TestAssertions::IntegrationTestAssertions`](./lib/turbo/test_assertions/integration_test_assertions.rb) are built on top of `Turbo::TestAssertions`, and add support for passing a `status:` keyword. They are automatically included in [`ActionDispatch::IntegrationTest`](https://edgeguides.rubyonrails.org/testing.html#integration-testing).
 
+The [`Turbo::Broadcastable::TestHelper`](./lib/turbo/broadcastable/test_helper.rb) concern provides Action Cable-aware test helpers that assert that `<turbo-stream>` elements were or were not broadcast over Action Cable. They are not automatically included. To use them in your tests, make sure to `include Turbo::Broadcastable::TestHelper`.
 
 ## Development
 
diff --git a/lib/turbo/broadcastable/test_helper.rb b/lib/turbo/broadcastable/test_helper.rb
new file mode 100644
index 00000000..c26d487d
--- /dev/null
+++ b/lib/turbo/broadcastable/test_helper.rb
@@ -0,0 +1,172 @@
+module Turbo
+  module Broadcastable
+    module TestHelper
+      extend ActiveSupport::Concern
+
+      included do
+        include ActionCable::TestHelper
+
+        include Turbo::Streams::StreamName
+      end
+
+      # Asserts that `<turbo-stream>` elements were broadcast over Action Cable
+      #
+      # === Arguments
+      #
+      # * <tt>stream_name_or_object</tt> the objects used to generate the
+      #   channel Action Cable name, or the name itself
+      # * <tt>&block</tt> optional block executed before the
+      #   assertion
+      #
+      # === Options
+      #
+      # * <tt>count:</tt> the number of `<turbo-stream>` elements that are
+      # expected to be broadcast
+      #
+      # Asserts `<turbo-stream>` elements were broadcast:
+      #
+      #     message = Message.find(1)
+      #     message.broadcast_replace_to "messages"
+      #
+      #     assert_turbo_stream_broadcasts "messages"
+      #
+      # Asserts that two `<turbo-stream>` elements were broadcast:
+      #
+      #     message = Message.find(1)
+      #     message.broadcast_replace_to "messages"
+      #     message.broadcast_remove_to "messages"
+      #
+      #     assert_turbo_stream_broadcasts "messages", count: 2
+      #
+      # You can pass a block to run before the assertion:
+      #
+      #     message = Message.find(1)
+      #
+      #     assert_turbo_stream_broadcasts "messages" do
+      #       message.broadcast_append_to "messages"
+      #     end
+      #
+      # In addition to a String, the helper also accepts an Object or Array to
+      # determine the name of the channel the elements are broadcast to:
+      #
+      #     message = Message.find(1)
+      #
+      #     assert_turbo_stream_broadcasts message do
+      #       message.broadcast_replace
+      #     end
+      #
+      def assert_turbo_stream_broadcasts(stream_name_or_object, count: nil, &block)
+        payloads = capture_turbo_stream_broadcasts(stream_name_or_object, &block)
+        stream_name = stream_name_from(stream_name_or_object)
+
+        if count.nil?
+          assert_not_empty payloads, "Expected at least one broadcast on #{stream_name.inspect}, but there were none"
+        else
+          broadcasts = "Turbo Stream broadcast".pluralize(count)
+
+          assert count == payloads.count, "Expected #{count} #{broadcasts} on #{stream_name.inspect}, but there were none"
+        end
+      end
+
+      # Asserts that no `<turbo-stream>` elements were broadcast over Action Cable
+      #
+      # === Arguments
+      #
+      # * <tt>stream_name_or_object</tt> the objects used to generate the
+      #   channel Action Cable name, or the name itself
+      # * <tt>&block</tt> optional block executed before the
+      #   assertion
+      #
+      # Asserts that no `<turbo-stream>` elements were broadcast:
+      #
+      #     message = Message.find(1)
+      #     message.broadcast_replace_to "messages"
+      #
+      #     assert_no_turbo_stream_broadcasts "messages" # fails with MiniTest::Assertion error
+      #
+      # You can pass a block to run before the assertion:
+      #
+      #     message = Message.find(1)
+      #
+      #     assert_no_turbo_stream_broadcasts "messages" do
+      #       # do something other than broadcast to "messages"
+      #     end
+      #
+      # In addition to a String, the helper also accepts an Object or Array to
+      # determine the name of the channel the elements are broadcast to:
+      #
+      #     message = Message.find(1)
+      #
+      #     assert_no_turbo_stream_broadcasts message do
+      #       # do something other than broadcast to "message_1"
+      #     end
+      #
+      def assert_no_turbo_stream_broadcasts(stream_name_or_object, &block)
+        block&.call
+
+        stream_name = stream_name_from(stream_name_or_object)
+
+        payloads = broadcasts(stream_name)
+
+        assert payloads.empty?, "Expected no broadcasts on #{stream_name.inspect}, but there were #{payloads.count}"
+      end
+
+      # Captures any `<turbo-stream>` elements that were broadcast over Action Cable
+      #
+      # === Arguments
+      #
+      # * <tt>stream_name_or_object</tt> the objects used to generate the
+      #   channel Action Cable name, or the name itself
+      # * <tt>&block</tt> optional block to capture broadcasts during execution
+      #
+      # Returns any `<turbo-stream>` elements that have been broadcast as an
+      # Array of <tt>Nokogiri::XML::Element</tt> instances
+      #
+      #     message = Message.find(1)
+      #     message.broadcast_append_to "messages"
+      #     message.broadcast_prepend_to "messages"
+      #
+      #     turbo_streams = capture_turbo_stream_broadcasts "messages"
+      #
+      #     assert_equal "append", turbo_streams.first["action"]
+      #     assert_equal "prepend", turbo_streams.second["action"]
+      #
+      # You can pass a block to limit the scope of the broadcasts being captured:
+      #
+      #     message = Message.find(1)
+      #
+      #     turbo_streams = capture_turbo_stream_broadcasts "messages" do
+      #       message.broadcast_append_to "messages"
+      #     end
+      #
+      #     assert_equal "append", turbo_streams.first["action"]
+      #
+      # In addition to a String, the helper also accepts an Object or Array to
+      # determine the name of the channel the elements are broadcast to:
+      #
+      #     message = Message.find(1)
+      #
+      #     replace, remove = capture_turbo_stream_broadcasts message do
+      #       message.broadcast_replace
+      #       message.broadcast_remove
+      #     end
+      #
+      #     assert_equal "replace", replace["action"]
+      #     assert_equal "replace", remove["action"]
+      #
+      def capture_turbo_stream_broadcasts(stream_name_or_object, &block)
+        block&.call
+
+        stream_name = stream_name_from(stream_name_or_object)
+        payloads = broadcasts(stream_name)
+
+        payloads.flat_map do |payload|
+          html = ActiveSupport::JSON.decode(payload)
+          document = Nokogiri::HTML5.parse(html)
+
+          document.at("body").element_children
+        end
+      end
+    end
+  end
+end
diff --git a/lib/turbo/engine.rb b/lib/turbo/engine.rb
index 08fd91c4..487bf39b 100644
--- a/lib/turbo/engine.rb
+++ b/lib/turbo/engine.rb
@@ -75,6 +75,7 @@ class Engine < Rails::Engine
     initializer "turbo.test_assertions" do
       ActiveSupport.on_load(:active_support_test_case) do
         require "turbo/test_assertions"
+        require "turbo/broadcastable/test_helper"
 
         include Turbo::TestAssertions
       end
diff --git a/test/broadcastable/test_helper_test.rb b/test/broadcastable/test_helper_test.rb
new file mode 100644
index 00000000..c417b805
--- /dev/null
+++ b/test/broadcastable/test_helper_test.rb
@@ -0,0 +1,264 @@
+require "test_helper"
+
+class Turbo::Broadcastable::TestHelper::CaptureTurboStreamBroadcastsTest < ActiveSupport::TestCase
+  include Turbo::Broadcastable::TestHelper
+
+  test "#capture_turbo_stream_broadcasts returns <turbo-stream> elements broadcast on a stream name" do
+    message = Message.new(id: 1)
+
+    message.broadcast_replace_to "messages"
+    message.broadcast_remove_to "messages"
+    replace, remove, *rest = capture_turbo_stream_broadcasts "messages"
+
+    assert_empty rest
+    assert_equal "replace", replace["action"]
+    assert_equal "remove", remove["action"]
+    assert_not_empty replace.at("template").element_children
+    assert_nil remove.at("template")
+  end
+
+  test "#capture_turbo_stream_broadcasts returns an empty Array when no broadcasts happened on a stream name" do
+    assert_empty capture_turbo_stream_broadcasts("messages")
+  end
+
+  test "#capture_turbo_stream_broadcasts returns <turbo-stream> elements broadcast on a stream object" do
+    message = Message.new(id: 1)
+
+    message.broadcast_replace
+    message.broadcast_remove
+    replace, remove, *rest = capture_turbo_stream_broadcasts message
+
+    assert_empty rest
+    assert_equal "replace", replace["action"]
+    assert_equal "remove", remove["action"]
+    assert_not_empty replace.at("template").element_children
+    assert_nil remove.at("template")
+  end
+
+  test "#capture_turbo_stream_broadcasts returns <turbo-stream> elements broadcast on an Array of stream objects" do
+    message = Message.new(id: 1)
+
+    message.broadcast_replace_to [message, :special]
+    message.broadcast_remove_to [message, :special]
+    replace, remove, *rest = capture_turbo_stream_broadcasts [message, :special]
+
+    assert_empty rest
+    assert_equal "replace", replace["action"]
+    assert_equal "remove", remove["action"]
+    assert_not_empty replace.at("template").element_children
+    assert_nil remove.at("template")
+  end
+
+  test "#capture_turbo_stream_broadcasts returns <turbo-stream> elements broadcast on a stream name from a block" do
+    message = Message.new(id: 1)
+
+    replace, remove, *rest = capture_turbo_stream_broadcasts "messages" do
+      message.broadcast_replace_to "messages"
+      message.broadcast_remove_to "messages"
+    end
+
+    assert_equal "replace", replace["action"]
+    assert_equal "remove", remove["action"]
+    assert_empty rest
+  end
+
+  test "#capture_turbo_stream_broadcasts returns <turbo-stream> elements broadcast on a stream object from a block" do
+    message = Message.new(id: 1)
+
+    replace, remove, *rest = capture_turbo_stream_broadcasts message do
+      message.broadcast_replace
+      message.broadcast_remove
+    end
+
+    assert_empty rest
+    assert_equal "replace", replace["action"]
+    assert_equal "remove", remove["action"]
+    assert_not_empty replace.at("template").element_children
+    assert_nil remove.at("template")
+  end
+
+  test "#capture_turbo_stream_broadcasts returns <turbo-stream> elements broadcast on an Array of stream objects from a block" do
+    message = Message.new(id: 1)
+
+    replace, remove, *rest = capture_turbo_stream_broadcasts [message, :special] do
+      message.broadcast_replace_to [message, :special]
+      message.broadcast_remove_to [message, :special]
+    end
+
+    assert_empty rest
+    assert_equal "replace", replace["action"]
+    assert_equal "remove", remove["action"]
+    assert_not_empty replace.at("template").element_children
+    assert_nil remove.at("template")
+  end
+
+  test "#capture_turbo_stream_broadcasts returns an empty Array when no broadcasts happened on a stream name from a block" do
+    streams = capture_turbo_stream_broadcasts "messages" do
+      # no-op
+    end
+
+    assert_empty streams
+  end
+end
+
+class Turbo::Broadcastable::TestHelper::AssertTurboStreamBroadcastsTest < ActiveSupport::TestCase
+  include Turbo::Broadcastable::TestHelper
+
+  test "#assert_turbo_stream_broadcasts passes when there is a broadcast" do
+    message = Message.new(id: 1)
+
+    message.broadcast_replace_to "messages"
+
+    assert_turbo_stream_broadcasts "messages"
+  end
+
+  test "#assert_turbo_stream_broadcasts passes when there are multiple broadcasts" do
+    message = Message.new(id: 1)
+
+    message.broadcast_replace_to "messages"
+    message.broadcast_remove_to "messages"
+
+    assert_turbo_stream_broadcasts "messages"
+  end
+
+  test "#assert_turbo_stream_broadcasts fails when no broadcasts happened on a stream name" do
+    assert_raises MiniTest::Assertion do
+      assert_turbo_stream_broadcasts "messages"
+    end
+  end
+
+  test "#assert_turbo_stream_broadcasts with a count: optional fails when no broadcasts happened on a stream name" do
+    singular_failure = assert_raises MiniTest::Assertion do
+      assert_turbo_stream_broadcasts "messages", count: 1
+    end
+
+    assert_includes singular_failure.message, %(1 Turbo Stream broadcast on "messages")
+
+    plural_failure = assert_raises MiniTest::Assertion do
+      assert_turbo_stream_broadcasts "messages", count: 2
+    end
+
+    assert_includes plural_failure.message, %(2 Turbo Stream broadcasts on "messages")
+  end
+
+  test "#assert_turbo_stream_broadcasts passes when broadcast on a stream object" do
+    message = Message.new(id: 1)
+
+    message.broadcast_replace
+    message.broadcast_remove
+
+    assert_turbo_stream_broadcasts message, count: 2
+  end
+
+  test "#assert_turbo_stream_broadcasts passes when broadcast on an Array of stream objects" do
+    message = Message.new(id: 1)
+
+    message.broadcast_replace_to [message, :special]
+    message.broadcast_remove_to [message, :special]
+
+    assert_turbo_stream_broadcasts [message, :special], count: 2
+  end
+
+  test "#assert_turbo_stream_broadcasts with a count: option passes when broadcast on a stream name from a block" do
+    message = Message.new(id: 1)
+
+    assert_turbo_stream_broadcasts "messages", count: 2 do
+      message.broadcast_replace_to "messages"
+      message.broadcast_remove_to "messages"
+    end
+  end
+
+  test "#assert_turbo_stream_broadcasts returns <turbo-stream> elements broadcast on a stream object from a block" do
+    message = Message.new(id: 1)
+
+    assert_turbo_stream_broadcasts message, count: 2 do
+      message.broadcast_replace
+      message.broadcast_remove
+    end
+  end
+
+  test "#assert_turbo_stream_broadcasts returns <turbo-stream> elements broadcast on an Array of stream objects from a block" do
+    message = Message.new(id: 1)
+
+    assert_turbo_stream_broadcasts [message, :special], count: 2 do
+      message.broadcast_replace_to [message, :special]
+      message.broadcast_remove_to [message, :special]
+    end
+  end
+
+  test "#assert_turbo_stream_broadcasts fails when no broadcasts happened on a stream name from a block" do
+    assert_raises MiniTest::Assertion do
+      assert_turbo_stream_broadcasts "messages" do
+        # no-op
+      end
+    end
+  end
+end
+
+class Turbo::Broadcastable::TestHelper::AssertNoTurboStreamBroadcastsTest < ActiveSupport::TestCase
+  include Turbo::Broadcastable::TestHelper
+
+  test "#assert_no_turbo_stream_broadcasts asserts no broadcasts with a stream name" do
+    assert_no_turbo_stream_broadcasts "messages"
+  end
+
+  test "#assert_no_turbo_stream_broadcasts asserts no broadcasts with a stream name from a block" do
+    assert_no_turbo_stream_broadcasts "messages" do
+      # no-op
+    end
+  end
+
+  test "#assert_no_turbo_stream_broadcasts asserts no broadcasts with a stream object" do
+    message = Message.new(id: 1)
+
+    assert_no_turbo_stream_broadcasts message
+  end
+
+  test "#assert_no_turbo_stream_broadcasts asserts no broadcasts with a stream object from a block" do
+    message = Message.new(id: 1)
+
+    assert_no_turbo_stream_broadcasts message do
+      # no-op
+    end
+  end
+
+  test "#assert_no_turbo_stream_broadcasts fails when when a broadcast happened on a stream name" do
+    message = Message.new(id: 1)
+
+    assert_raises MiniTest::Assertion do
+      message.broadcast_remove_to "messages"
+
+      assert_no_turbo_stream_broadcasts "messages"
+    end
+  end
+
+  test "#assert_no_turbo_stream_broadcasts fails when when a broadcast happened on a stream name from a block" do
+    message = Message.new(id: 1)
+
+    assert_raises MiniTest::Assertion do
+      assert_no_turbo_stream_broadcasts "messages" do
+        message.broadcast_remove_to "messages"
+      end
+    end
+  end
+
+  test "#assert_no_turbo_stream_broadcasts fails when when a broadcast happened on a stream object" do
+    message = Message.new(id: 1)
+
+    assert_raises MiniTest::Assertion do
+      message.broadcast_remove
+
+      assert_no_turbo_stream_broadcasts message
+    end
+  end
+
+  test "#assert_no_turbo_stream_broadcasts fails when when a broadcast happened on a stream object from a block" do
+    message = Message.new(id: 1)
+
+    assert_raises MiniTest::Assertion do
+      assert_no_turbo_stream_broadcasts message do
+        message.broadcast_remove
+      end
+    end
+  end
+end