diff --git a/lib/channel_spec/testing.ex b/lib/channel_spec/testing.ex
index 018fb05..d0ff956 100644
--- a/lib/channel_spec/testing.ex
+++ b/lib/channel_spec/testing.ex
@@ -8,7 +8,16 @@ defmodule ChannelSpec.Testing do
       import Phoenix.ChannelTest, except: [push: 3, subscribe_and_join: 3]
 
       import unquote(__MODULE__),
-        only: [push: 3, assert_reply_spec: 2, assert_reply_spec: 3, subscribe_and_join: 3]
+        only: [
+          push: 3,
+          assert_reply_spec: 2,
+          assert_reply_spec: 3,
+          subscribe_and_join: 3,
+          assert_broadcast_spec: 3,
+          assert_broadcast_spec: 4,
+          assert_push_spec: 3,
+          assert_push_spec: 4
+        ]
     end
   end
 
@@ -141,4 +150,108 @@ defmodule ChannelSpec.Testing do
       end
     end
   end
+
+  @doc """
+  Same as `Phoenix.ChannelTest.assert_push/3` but verifies the message
+  against the subscription schema defined for the socket that handled the message.
+  """
+  defmacro assert_push_spec(
+             socket,
+             event,
+             payload,
+             timeout \\ Application.fetch_env!(:ex_unit, :assert_receive_timeout)
+           ) do
+    quote do
+      assert_receive %Phoenix.Socket.Message{
+                       event: unquote(event),
+                       payload: unquote(payload) = payload
+                     },
+                     unquote(timeout)
+
+      socket = unquote(socket)
+      socket_schema = unquote(socket).handler.__socket_schemas__()
+      topic = unquote(socket).assigns.__channel_topic__
+      event = unquote(event)
+
+      with true <- function_exported?(socket.handler, :__socket_schemas__, 0),
+           %{} = schema <-
+             socket_schema["channels"][topic]["subscriptions"][event] do
+        case Xema.validate(schema, payload) do
+          :ok ->
+            :ok
+
+          {:error, %m{} = error} ->
+            raise SpecError,
+              message: """
+              Channel push doesn't match spec for subscription #{event}:
+
+              Payload:
+                #{inspect(payload)}
+
+              Error:
+                #{m.format_error(error)}
+
+              Schema:
+                #{inspect(schema)}
+              """
+        end
+
+        payload
+      else
+        _ -> payload
+      end
+    end
+  end
+
+  @doc """
+  Same as `Phoenix.ChannelTest.assert_broadcast/3` but verifies the message
+  against the subscription schema defined for the socket that handled the message.
+  """
+  defmacro assert_broadcast_spec(
+             socket,
+             event,
+             payload,
+             timeout \\ Application.fetch_env!(:ex_unit, :assert_receive_timeout)
+           ) do
+    quote do
+      assert_receive %Phoenix.Socket.Broadcast{
+                       event: unquote(event),
+                       payload: unquote(payload) = payload
+                     },
+                     unquote(timeout)
+
+      socket = unquote(socket)
+      socket_schema = unquote(socket).handler.__socket_schemas__()
+      topic = unquote(socket).assigns.__channel_topic__
+      event = unquote(event)
+
+      with true <- function_exported?(socket.handler, :__socket_schemas__, 0),
+           %{} = schema <-
+             socket_schema["channels"][topic]["subscriptions"][event] do
+        case Xema.validate(schema, payload) do
+          :ok ->
+            :ok
+
+          {:error, %m{} = error} ->
+            raise SpecError,
+              message: """
+              Channel broadcast doesn't match spec for subscription #{event}:
+
+              Payload:
+                #{inspect(payload)}
+
+              Error:
+                #{m.format_error(error)}
+
+              Schema:
+                #{inspect(schema)}
+              """
+        end
+
+        payload
+      else
+        _ -> payload
+      end
+    end
+  end
 end
diff --git a/test/channel_spec/operations_test.exs b/test/channel_spec/operations_test.exs
index 0da9200..a4b7574 100644
--- a/test/channel_spec/operations_test.exs
+++ b/test/channel_spec/operations_test.exs
@@ -194,7 +194,8 @@ defmodule ChannelSpec.OperationsTest do
                  end
                )
 
-      assert err.message == ~s(The schema for subscription "foo" is not a valid schema map or module.\n)
+      assert err.message ==
+               ~s(The schema for subscription "foo" is not a valid schema map or module.\n)
     end
   end
 end
diff --git a/test/channel_spec/testing_test.exs b/test/channel_spec/testing_test.exs
index 1ca9aac..18ae30e 100644
--- a/test/channel_spec/testing_test.exs
+++ b/test/channel_spec/testing_test.exs
@@ -347,4 +347,114 @@ defmodule ChannelSpec.TestingTest do
       assert_reply_spec ref, :ok, :works
     end
   end
+
+  describe "assert_push_spec/3" do
+    @tag :capture_log
+    test "validates the subscription spec", %{mod: mod} do
+      defmodule :"#{mod}.RoomChannel" do
+        use Phoenix.Channel
+        use ChannelHandler.Router
+        use ChannelSpec.Operations
+
+        def join("room:" <> _, _params, socket) do
+          {:ok, socket}
+        end
+
+        subscription "server_msg", %{type: :object, properties: %{body: %{type: :integer}}}
+
+        def handle_in(_, _, socket) do
+          Phoenix.Channel.push(socket, "server_msg", %{"body" => true})
+          {:reply, {:ok, :works}, socket}
+        end
+      end
+
+      defmodule :"#{mod}.UserSocket" do
+        use ChannelSpec.Socket
+
+        channel "room:*", :"#{mod}.RoomChannel"
+      end
+
+      defmodule :"#{mod}.Endpoint" do
+        use Phoenix.Endpoint, otp_app: :channel_spec
+
+        Phoenix.Endpoint.socket("/socket", :"#{mod}.UserSocket")
+
+        defoverridable config: 1, config: 2
+        def config(:pubsub_server), do: __MODULE__.PubSub
+        def config(which), do: super(which)
+        def config(which, default), do: super(which, default)
+      end
+
+      start_supervised({Phoenix.PubSub, name: :"#{mod}.Endpoint.PubSub"})
+
+      {:ok, _endpoint_pid} = start_supervised(:"#{mod}.Endpoint")
+
+      {:ok, _, socket} =
+        :"#{mod}.UserSocket"
+        |> build_socket("room:123", %{}, :"#{mod}.Endpoint")
+        |> subscribe_and_join(:"#{mod}.RoomChannel", "room:123")
+
+      _ref = push(socket, "new_msg", %{"body" => 123})
+
+      error = catch_error(assert_push_spec(socket, "server_msg", _))
+
+      assert %ChannelSpec.Testing.SpecError{} = error
+      assert error.message =~ "Channel push doesn't match spec for subscription server_msg"
+    end
+  end
+
+  describe "assert_broadcast_spec/4" do
+    @tag :capture_log
+    test "validates the subscription spec", %{mod: mod} do
+      defmodule :"#{mod}.RoomChannel" do
+        use Phoenix.Channel
+        use ChannelHandler.Router
+        use ChannelSpec.Operations
+
+        def join("room:" <> _, _params, socket) do
+          {:ok, socket}
+        end
+
+        subscription "server_msg", %{type: :object, properties: %{body: %{type: :integer}}}
+
+        def handle_in(_, _, socket) do
+          broadcast(socket, "server_msg", %{"body" => true})
+          {:reply, {:ok, :works}, socket}
+        end
+      end
+
+      defmodule :"#{mod}.UserSocket" do
+        use ChannelSpec.Socket
+
+        channel "room:*", :"#{mod}.RoomChannel"
+      end
+
+      defmodule :"#{mod}.Endpoint" do
+        use Phoenix.Endpoint, otp_app: :channel_spec
+
+        Phoenix.Endpoint.socket("/socket", :"#{mod}.UserSocket")
+
+        defoverridable config: 1, config: 2
+        def config(:pubsub_server), do: __MODULE__.PubSub
+        def config(which), do: super(which)
+        def config(which, default), do: super(which, default)
+      end
+
+      start_supervised({Phoenix.PubSub, name: :"#{mod}.Endpoint.PubSub"})
+
+      {:ok, _endpoint_pid} = start_supervised(:"#{mod}.Endpoint")
+
+      {:ok, _, socket} =
+        :"#{mod}.UserSocket"
+        |> build_socket("room:123", %{}, :"#{mod}.Endpoint")
+        |> subscribe_and_join(:"#{mod}.RoomChannel", "room:123")
+
+      _ref = push(socket, "new_msg", %{"body" => 123})
+
+      error = catch_error(assert_broadcast_spec(socket, "server_msg", _))
+
+      assert %ChannelSpec.Testing.SpecError{} = error
+      assert error.message =~ "Channel broadcast doesn't match spec for subscription server_msg"
+    end
+  end
 end