Unit Testing a GenStage Producer

Questions & Help Questions
, ,
3

Usually a create a simple consumer that just forwards every event it receives to the process that started it:

defmodule TestConsumer do
  def start_link(producer) do
    GenStage.start_link(__MODULE__, {producer, self()})
  end

  def init({producer, owner}) do
    {:consumer, owner, subscribe_to: [producer]}
  end

  def handle_events(events, _from, owner) do
    send(owner, {:received, events})
    {:noreply, [], owner}
  end
end

Then I can use it in my tests to assert about the produced events:

defmodule OutputNonsenseTest do
  use ExUnit.Case

  test "check the results" do
    {:ok, stage} = GenStage.start_link(OutputNonsense, arg = :unused)
    {:ok, _cons} = TestConsumer.start_link(stage)

    assert_receive {:received, events}
    # assertions about the events

    # The test consumer will also stop, since it is subscribed to the stage
    GenStage.stop(stage)
  end
end

I hope it helps you somehow