Add wanda event handling policy

config/wanda.exs

@@ -7,6 +7,8 @@ config :trento, :messaging, adapter: Trento.Messaging.Adapters.AMQP
 
 config :trento, Trento.Integration.Checks, adapter: Trento.Integration.Checks.Wanda
 
+config :trento, Trento.Integration.Checks.Wanda, policy: Trento.Integration.Checks.Wanda.Policy
+
 config :trento, Trento.Messaging.Publisher, adapter: Trento.Messaging.Adapters.AMQP
 
 config :trento, Trento.Messaging.Adapters.AMQP,

lib/trento/application/integration/checks/adapter/wanda/messaging/amqp/processor.ex

@@ -5,39 +5,22 @@ defmodule Trento.Integration.Checks.Wanda.Messaging.AMQP.Processor do
 
   @behaviour GenRMQ.Processor
 
-  alias Trento.Checks.V1.ExecutionCompleted
   alias Trento.Contracts
-  alias Trento.Domain.Commands.CompleteChecksExecutionWanda
 
   require Logger
 
   def process(%GenRMQ.Message{payload: payload} = message) do
     Logger.debug("Received message: #{inspect(message)}")
 
-    case Contracts.from_event(payload) do
-      {:ok, event} ->
-        handle(event)
-
+    with {:ok, event} <- Contracts.from_event(payload),
+         {:ok, command, opts} <- adapter().handle(event) do
+      Trento.Commanded.dispatch(command, opts)
+    else
       {:error, reason} ->
         {:error, reason}
     end
   end
 
-  defp handle(%ExecutionCompleted{
-         execution_id: execution_id,
-         group_id: group_id,
-         result: result
-       }) do
-    with {:ok, command} <-
-           CompleteChecksExecutionWanda.new(%{
-             cluster_id: group_id,
-             health: map_health(result)
-           }) do
-      Trento.Commanded.dispatch(command, correlation_id: execution_id)
-    end
-  end
-
-  defp map_health(:CRITICAL), do: :critical
-  defp map_health(:WARNING), do: :warning
-  defp map_health(:PASSING), do: :passing
+  defp adapter,
+    do: Application.fetch_env!(:trento, Trento.Integration.Checks.Wanda)[:policy]
 end

lib/trento/application/integration/checks/adapter/wanda/policy/behaviour.ex

@@ -0,0 +1,10 @@
+defmodule Trento.Integration.Checks.Wanda.Behaviour do
+  @moduledoc """
+  Wanda event policy behaviour
+  """
+
+  alias Trento.Checks.V1.ExecutionCompleted
+
+  @callback handle(event :: ExecutionCompleted.t()) ::
+              {:ok, command :: any, opts :: Keyword.t()} | {:error, any}
+end

lib/trento/application/integration/checks/adapter/wanda/policy/policy.ex

@@ -0,0 +1,32 @@
+defmodule Trento.Integration.Checks.Wanda.Policy do
+  @moduledoc """
+  Wanda event policies
+  """
+
+  @behaviour Trento.Integration.Checks.Wanda.Behaviour
+
+  alias Trento.Checks.V1.ExecutionCompleted
+  alias Trento.Domain.Commands.CompleteChecksExecutionWanda
+
+  def handle(%ExecutionCompleted{
+        execution_id: execution_id,
+        group_id: group_id,
+        result: result
+      }) do
+    case CompleteChecksExecutionWanda.new(%{
+           cluster_id: group_id,
+           health: map_health(result)
+         }) do
+      {:ok, command} ->
+        opts = [correlation_id: execution_id]
+        {:ok, command, opts}
+
+      error ->
+        error
+    end
+  end
+
+  defp map_health(:CRITICAL), do: :critical
+  defp map_health(:WARNING), do: :warning
+  defp map_health(:PASSING), do: :passing
+end

test/test_helper.exs

@@ -16,6 +16,14 @@ Application.put_env(:trento, Trento.Integration.Prometheus,
 Mox.defmock(Trento.Messaging.Adapters.Mock, for: Trento.Messaging.Adapters.Behaviour)
 Application.put_env(:trento, :messaging, adapter: Trento.Messaging.Adapters.Mock)
 
+Mox.defmock(Trento.Integration.Checks.Wanda.Policy.Mock,
+  for: Trento.Integration.Checks.Wanda.Behaviour
+)
+
+Application.put_env(:trento, Trento.Integration.Checks.Wanda,
+  policy: Trento.Integration.Checks.Wanda.Policy.Mock
+)
+
 Mox.defmock(GenRMQ.Processor.Mock, for: GenRMQ.Processor)
 
 Application.ensure_all_started(:ex_machina, :faker)

test/trento/application/integration/checks/wanda/messaging/amqp/processor_test.exs

@@ -3,36 +3,62 @@ defmodule Trento.Integration.Checks.Wanda.Messaging.AMQP.ProcessorTest do
   use ExUnit.Case, async: true
   use Trento.DataCase
 
+  import Mox
   import Mock
 
   alias Trento.Integration.Checks.Wanda.Messaging.AMQP.Processor
 
   alias Trento.Checks.V1.ExecutionCompleted
   alias Trento.Contracts
-  alias Trento.Domain.Commands.CompleteChecksExecutionWanda
-
-  require Trento.Domain.Enums.Health, as: Health
 
   describe "process" do
-    test "should process ExecutionCompleted event" do
-      execution_id = UUID.uuid4()
-      cluster_id = UUID.uuid4()
-
-      with_mock Trento.Commanded, dispatch: fn _, _ -> :ok end do
-        execution_completed = Contracts.to_event(%ExecutionCompleted{
-          execution_id: execution_id,
-          group_id: cluster_id,
+    test "should process valid event and dispatch command" do
+      execution_completed =
+        Contracts.to_event(%ExecutionCompleted{
+          execution_id: UUID.uuid4(),
+          group_id: UUID.uuid4(),
           result: :PASSING
         })
-        message = %GenRMQ.Message{payload: execution_completed, attributes: %{}, channel: nil}
 
+      message = %GenRMQ.Message{payload: execution_completed, attributes: %{}, channel: nil}
+      command = "some-command"
+      opts = [correlation_id: UUID.uuid4()]
+
+      expect(Trento.Integration.Checks.Wanda.Policy.Mock, :handle, fn _ ->
+        {:ok, command, opts}
+      end)
+
+      with_mock Trento.Commanded, dispatch: fn _, _ -> :ok end do
         assert :ok = Processor.process(message)
 
-        assert_called Trento.Commanded.dispatch(%CompleteChecksExecutionWanda{
-                        cluster_id: cluster_id,
-                        health: Health.passing()
-                      }, correlation_id: execution_id)
+        assert_called Trento.Commanded.dispatch(
+                        command,
+                        opts
+                      )
       end
     end
+
+    test "should return error if the event handling fails" do
+      execution_completed =
+        Contracts.to_event(%ExecutionCompleted{
+          execution_id: UUID.uuid4(),
+          group_id: UUID.uuid4(),
+          result: :PASSING
+        })
+
+      message = %GenRMQ.Message{payload: execution_completed, attributes: %{}, channel: nil}
+
+      expect(Trento.Integration.Checks.Wanda.Policy.Mock, :handle, fn _ ->
+        {:error, :handling_error}
+      end)
+
+      assert {:error, :handling_error} = Processor.process(message)
+    end
+
+    @tag capture_log: true
+    test "should return error if the event cannot be decoded" do
+      message = %GenRMQ.Message{payload: "bad-payload", attributes: %{}, channel: nil}
+      assert {:error, :decoding_error} = Processor.process(message)
+    end
   end
 end

test/trento/application/integration/checks/wanda/policy/policy_test.exs

@@ -0,0 +1,29 @@
+defmodule Trento.Integration.Checks.Wanda.PolicyTest do
+  @moduledoc false
+  use ExUnit.Case
+
+  alias Trento.Checks.V1.ExecutionCompleted
+  alias Trento.Domain.Commands.CompleteChecksExecutionWanda
+  alias Trento.Integration.Checks.Wanda.Policy
+
+  require Trento.Domain.Enums.Health, as: Health
+
+  describe "handle" do
+    test "should handle ExecutionCompleted event" do
+      execution_id = UUID.uuid4()
+      cluster_id = UUID.uuid4()
+
+      assert {:ok,
+              %CompleteChecksExecutionWanda{
+                cluster_id: ^cluster_id,
+                health: Health.passing()
+              },
+              [correlation_id: ^execution_id]} =
+               Policy.handle(%ExecutionCompleted{
+                 execution_id: execution_id,
+                 group_id: cluster_id,
+                 result: :PASSING
+               })
+    end
+  end
+end