Unit Testing a GenStage Producer

craigambrose

I’m working on my first GenStage producer, and I want to unit test it as I write the handle demand method. I’m having trouble simplifying the use of GenStage enough to write a simple test. Here’s a very simple producer that when asked for demand, outputs a list of three atoms.

defmodule OutputNonsense do
  use GenStage

  def init(arg) do
    {:producer, :some_kind_of_state}
  end

  def handle_demand(demand, state) when demand > 0 do
    {:noreply, things = [:whatever, :you, :want], state}
  end
end

When it comes to writing a test for this, this is the best I’ve been able to manage. It outputs the three atoms to the console, but never finishes execution.

defmodule OutputNonsenseTest do
  use ExUnit.Case

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

    GenStage.stream([stage])
    |> Stream.each(&IO.inspect/1)
    |> Stream.run()
  end
end

I realise that I’m streaming the results into IO.inspect, rather than collecting them to check in an assert. I’m really not sure how to do that. Would love any pointers.

5 comments

#testing #exunit #genstage

3 3682 5

2017-09-26 14:58:47 UTC

Showing Posts 1 to 5

Show Best Posts
Show All (oldest first)
Show All (newest first)

wfgilman

I have a producer which queries tasks from a database table. I test each callback directly.

For example, my handle_demand/2 callback has the following code:

  def handle_demand(demand, state) when demand > 0 do
    limit = demand + state
    get_events(limit)
  end

  ## Helper functions.
  defp get_events(limit) do
    {:ok, {count, events}} = Task.get_waiting_tasks(limit, :requesting)
    {:noreply, events, limit - count}
  end

This is the test I wrote:

test "handle_demand updates state" do
  insert!(:task)
  {:noreply, _tasks, state} = Feeder.handle_demand(50, 100)

  assert state == 149
end

insert!/1 is a factory which inserts a task into the table, so there’s only one event. The callback asks for 50 in addition to the existing 100, but only gets one. So the resulting state is 149.

I dunno whether this is helpful.

Post #1

vidalraphael

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

Post #2