如何配置Phoenix Channel测试超时？同步推送场景测试适配

Hey there! I see exactly why you're hitting a timeout in your tests—your SyncChannel has three Process.sleep(2000) calls, adding up to 6 seconds of wait time total, and Phoenix's default Channel test timeout is usually 5 seconds. Let's walk through a few ways to fix this:

1. Set a Custom Timeout for Individual Tests #

The simplest quick fix is to increase the timeout just for this test case. You can do this with a test tag or by specifying the timeout directly in assert_receive:

Using a Test Tag #

defmodule AppWeb.SyncChannelTest do
  use AppWeb.ChannelCase

  # Give the test 7 seconds to complete (plenty for 6 seconds of sleep)
  @tag timeout: 7000
  test "receives all sync push messages", %{socket: socket} do
    {:ok, _, socket} = subscribe_and_join(socket, "sync:database", %{})

    # Wait for each push event to arrive
    assert_receive %Phoenix.Socket.Message{event: "one"}
    assert_receive %Phoenix.Socket.Message{event: "two"}
    assert_receive %Phoenix.Socket.Message{event: "three"}
  end
end

Specifying Timeout in assert_receive #

If you prefer, you can set the timeout per assertion instead:

assert_receive %Phoenix.Socket.Message{event: "one"}, 7000
assert_receive %Phoenix.Socket.Message{event: "two"}, 7000
assert_receive %Phoenix.Socket.Message{event: "three"}, 7000

2. Adjust Global Test Timeout (Less Recommended) #

If you have multiple tests that need longer timeouts, you can change the global ExUnit timeout in test/test_helper.exs:

ExUnit.start(timeout: 7000) # Default is usually 5000 (5 seconds)

Keep in mind this slows down your entire test suite if a test hangs, so it's better to use per-test timeouts when possible.

3. Optimize Your Test (Best Practice) #

Instead of waiting for real sleep time (which makes tests slow), you can bypass the Process.sleep calls entirely. Here's how:

Refactor the Channel for Testability #

Extract the sleep logic into a separate function that you can override in tests:

# In your SyncChannel module
defmodule AppWeb.SyncChannel do
  use AppWeb, :channel

  def join("sync:database", _payload, socket) do
    send(self(), {:sync, :database})
    {:ok, socket}
  end

  def handle_info({:sync, :database}, socket) do
    wait()
    push(socket, "one", %{})
    wait()
    push(socket, "two", %{})
    wait()
    push(socket, "three", %{})
    {:noreply, socket}
  end

  # Make the wait function overridable
  defp wait, do: Process.sleep(2000)
end

Then in your test, override the wait/0 function to skip the sleep:

defmodule AppWeb.SyncChannelTest do
  use AppWeb.ChannelCase
  import AppWeb.SyncChannel, only: [wait: 0]

  # Override to no-op during tests
  def wait, do: :ok

  test "receives all sync messages instantly", %{socket: socket} do
    {:ok, _, socket} = subscribe_and_join(socket, "sync:database", %{})

    # Send the sync event directly to the channel process
    send(socket.pid, {:sync, :database})

    # Assert messages arrive without waiting
    assert_receive %Phoenix.Socket.Message{event: "one"}, 100
    assert_receive %Phoenix.Socket.Message{event: "two"}, 100
    assert_receive %Phoenix.Socket.Message{event: "three"}, 100
  end
end

This way your tests run in milliseconds instead of seconds, which keeps your test suite fast and focused on logic rather than wait times.

内容的提问来源于stack exchange，提问作者Sam Houston