Repository: plataformatec/broadway
Pull request #136 (2019-11-21 18:16:43)

Allow passing `:metadata` field to `test_messages`
BroadwayRabbitMQ.Producer allows appending certain metadata fields
to the message. This comes useful when we want to retrieve additional
information that is exposed by RabbitMQ, but not part of the message
body.

If our producer's `handle_message` function relies on `:metadata`, currently
we cannot test it with `test_messages`.

josevalim: Looks great @laszloesl! Can you please add some tests too?
laszloesl: Sure, I will.
laszloesl: Thanks for the quick feedback @josevalim . Will these tests do?

Patch:
diff --git a/lib/broadway.ex b/lib/broadway.ex
index af7a68b..f83147a 100644
--- a/lib/broadway.ex
+++ b/lib/broadway.ex
@@ -751,6 +751,10 @@ defmodule Broadway do
     in is immediately delivered. When set to `:bulk`, batch is
     delivered when its size or timeout is reached. Defaults to `:flush`.
 
+  * `:metadata` - optionally a map of additional fields to add to the
+    message. This can be used, for example, when testing
+    `BroadwayRabbitMQ.Producer`.
+
   ## Examples
 
   For example, in your tests, you may do:
@@ -761,13 +765,19 @@ defmodule Broadway do
       assert length(failed) == 0
 
   """
-  @spec test_messages(GenServer.server(), data :: [term]) :: reference
+  @spec test_messages(GenServer.server(), data :: [term], opts :: Keyword.t()) :: reference
   def test_messages(broadway, data, opts \\ []) when is_list(data) and is_list(opts) do
     batch_mode = Keyword.get(opts, :batch_mode, :flush)
+    metadata = Map.new(Keyword.get(opts, :metadata, []))
+
     ref = make_ref()
     ack = {Broadway.CallerAcknowledger, {self(), ref}, :ok}
 
-    messages = Enum.map(data, &%Message{data: &1, acknowledger: ack, batch_mode: batch_mode})
+    messages =
+      Enum.map(
+        data,
+        &%Message{data: &1, acknowledger: ack, batch_mode: batch_mode, metadata: metadata}
+      )
 
     :ok = push_messages(broadway, messages)
     ref


Test patch:
diff --git a/test/broadway_test.exs b/test/broadway_test.exs
index d9cd129..81e98b1 100644
--- a/test/broadway_test.exs
+++ b/test/broadway_test.exs
@@ -404,6 +404,48 @@ defmodule BroadwayTest do
     assert_receive {:ack, :ref, [_, _], []}
   end
 
+  describe "test_messages/3" do
+    setup do
+      handle_message = fn message, _ ->
+        message
+      end
+
+      context = %{
+        handle_message: handle_message
+      }
+
+      broadway_name = new_unique_name()
+
+      {:ok, broadway} =
+        Broadway.start_link(CustomHandlers,
+          name: broadway_name,
+          context: context,
+          producer: [module: {ManualProducer, []}],
+          processors: [default: [stages: 1, min_demand: 1, max_demand: 2]]
+        )
+
+      %{broadway: broadway}
+    end
+
+    test "metadata field is added to message, if specified in test_messages",
+         %{broadway: broadway} do
+      ref =
+        Broadway.test_messages(broadway, [:message],
+          batch_mode: :flush,
+          metadata: %{field: :value}
+        )
+
+      assert_receive {:ack, ^ref, [%Message{data: :message, metadata: %{field: :value}}], []}
+    end
+
+    test "metadata field in message defaults to %{}, if not specified in test_messages",
+         %{broadway: broadway} do
+      ref = Broadway.test_messages(broadway, [:message], batch_mode: :flush)
+
+      assert_receive {:ack, ^ref, [%Message{data: :message, metadata: %{}}], []}
+    end
+  end
+
   describe "processor" do
     setup do
       test_pid = self()