GenStage Integration testing

Questions & Help Questions
2

In your integration test, you would test that the job is processed correctly through the entire pipeline when the thing C does happens. If that’s inserting into a database, then test that the record was created.

My setup would be to subscribe a fake consumer to each stage and just emit the events it receives to the test process:

defmodule EchoConsumer do
  use GenStage

  def start_link(producer, test) do
    GenStage.start_link(__MODULE__, {producer, test})
  end

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

  def handle_events(events, _, test) do
    for event <- events, do: send(test, {:received, event})
  end
end

test "A produces events" do
  {:ok, _} = StageA.start_link()
  {:ok, _} = EchoConsumer.start_link(StageA, self())
  StageA.produce_event(1)
  assert_receive {:event, 1}
end