Add Pipeline module, test for federation.

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -18,8 +18,6 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
   alias Pleroma.Web.ActivityPub.MRF
   alias Pleroma.Web.ActivityPub.Transmogrifier
   alias Pleroma.Web.ActivityPub.Utils
-  alias Pleroma.Web.ActivityPub.ObjectValidator
-  alias Pleroma.Web.ActivityPub.SideEffects
   alias Pleroma.Web.Streamer
   alias Pleroma.Web.WebFinger
   alias Pleroma.Workers.BackgroundWorker
@@ -125,25 +123,10 @@ def increase_poll_votes_if_vote(%{
 
   def increase_poll_votes_if_vote(_create_data), do: :noop
 
-  @spec common_pipeline(map(), keyword()) :: {:ok, Activity.t(), keyword()} | {:error, any()}
-  def common_pipeline(object, meta) do
-    with {_, {:ok, validated_object, meta}} <-
-           {:validate_object, ObjectValidator.validate(object, meta)},
-         {_, {:ok, mrfd_object}} <- {:mrf_object, MRF.filter(validated_object)},
-         {_, {:ok, %Activity{} = activity, meta}} <-
-           {:persist_object, persist(mrfd_object, meta)},
-         {_, {:ok, %Activity{} = activity, meta}} <-
-           {:execute_side_effects, SideEffects.handle(activity, meta)} do
-      {:ok, activity, meta}
-    else
-      e -> {:error, e}
-    end
-  end
-
   # TODO rewrite in with style
   @spec persist(map(), keyword()) :: {:ok, Activity.t() | Object.t()}
   def persist(object, meta) do
-    local = Keyword.get(meta, :local)
+    local = Keyword.fetch!(meta, :local)
     {recipients, _, _} = get_recipients(object)
 
     {:ok, activity} =

lib/pleroma/web/activity_pub/pipeline.ex

+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.ActivityPub.Pipeline do
+  alias Pleroma.Activity
+  alias Pleroma.Web.ActivityPub.ActivityPub
+  alias Pleroma.Web.ActivityPub.MRF
+  alias Pleroma.Web.ActivityPub.ObjectValidator
+  alias Pleroma.Web.ActivityPub.SideEffects
+  alias Pleroma.Web.Federator
+
+  @spec common_pipeline(map(), keyword()) :: {:ok, Activity.t(), keyword()} | {:error, any()}
+  def common_pipeline(object, meta) do
+    with {_, {:ok, validated_object, meta}} <-
+           {:validate_object, ObjectValidator.validate(object, meta)},
+         {_, {:ok, mrfd_object}} <- {:mrf_object, MRF.filter(validated_object)},
+         {_, {:ok, %Activity{} = activity, meta}} <-
+           {:persist_object, ActivityPub.persist(mrfd_object, meta)},
+         {_, {:ok, %Activity{} = activity, meta}} <-
+           {:execute_side_effects, SideEffects.handle(activity, meta)},
+         {_, {:ok, _}} <- {:federation, maybe_federate(activity, meta)} do
+      {:ok, activity, meta}
+    else
+      e -> {:error, e}
+    end
+  end
+
+  defp maybe_federate(activity, meta) do
+    with {:ok, local} <- Keyword.fetch(meta, :local) do
+      if local do
+        Federator.publish(activity)
+        {:ok, :federated}
+      else
+        {:ok, :not_federated}
+      end
+    else
+      _e -> {:error, "local not set in meta"}
+    end
+  end
+end

lib/pleroma/web/activity_pub/transmogrifier.ex

@@ -12,12 +12,13 @@ defmodule Pleroma.Web.ActivityPub.Transmogrifier do
   alias Pleroma.Repo
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ActivityPub
+  alias Pleroma.Web.ActivityPub.ObjectValidator
+  alias Pleroma.Web.ActivityPub.ObjectValidators.LikeValidator
+  alias Pleroma.Web.ActivityPub.Pipeline
   alias Pleroma.Web.ActivityPub.Utils
   alias Pleroma.Web.ActivityPub.Visibility
   alias Pleroma.Web.Federator
   alias Pleroma.Workers.TransmogrifierWorker
-  alias Pleroma.Web.ActivityPub.ObjectValidator
-  alias Pleroma.Web.ActivityPub.ObjectValidators.LikeValidator
 
   import Ecto.Query
 
@@ -572,7 +573,7 @@ def handle_incoming(%{"type" => "Like"} = data, _options) do
          {_, {:ok, cast_data}} <-
            {:maybe_add_recipients, maybe_add_recipients_from_object(cast_data)},
          {_, {:ok, activity, _meta}} <-
-           {:common_pipeline, ActivityPub.common_pipeline(cast_data, local: false)} do
+           {:common_pipeline, Pipeline.common_pipeline(cast_data, local: false)} do
       {:ok, activity}
     else
       e -> {:error, e}

lib/pleroma/web/common_api/common_api.ex

@@ -11,6 +11,7 @@ defmodule Pleroma.Web.CommonAPI do
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.ActivityPub.Builder
+  alias Pleroma.Web.ActivityPub.Pipeline
   alias Pleroma.Web.ActivityPub.Utils
   alias Pleroma.Web.ActivityPub.Visibility
 
@@ -106,7 +107,7 @@ def favorite(%User{} = user, id) do
          {_, {:ok, like_object, meta}} <- {:build_object, Builder.like(user, object)},
          {_, {:ok, %Activity{} = activity, _meta}} <-
            {:common_pipeline,
-            ActivityPub.common_pipeline(like_object, Keyword.put(meta, :local, true))} do
+            Pipeline.common_pipeline(like_object, Keyword.put(meta, :local, true))} do
       {:ok, activity}
     else
       e ->

test/web/activity_pub/pipeline_test.exs

+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.ActivityPub.PipelineTest do
+  use Pleroma.DataCase
+
+  import Mock
+  import Pleroma.Factory
+
+  describe "common_pipeline/2" do
+    test "it goes through validation, filtering, persisting, side effects and federation for local activities" do
+      activity = insert(:note_activity)
+      meta = [local: true]
+
+      with_mocks([
+        {Pleroma.Web.ActivityPub.ObjectValidator, [], [validate: fn o, m -> {:ok, o, m} end]},
+        {
+          Pleroma.Web.ActivityPub.MRF,
+          [],
+          [filter: fn o -> {:ok, o} end]
+        },
+        {
+          Pleroma.Web.ActivityPub.ActivityPub,
+          [],
+          [persist: fn o, m -> {:ok, o, m} end]
+        },
+        {
+          Pleroma.Web.ActivityPub.SideEffects,
+          [],
+          [handle: fn o, m -> {:ok, o, m} end]
+        },
+        {
+          Pleroma.Web.Federator,
+          [],
+          [publish: fn _o -> :ok end]
+        }
+      ]) do
+        assert {:ok, ^activity, ^meta} =
+                 Pleroma.Web.ActivityPub.Pipeline.common_pipeline(activity, meta)
+
+        assert_called(Pleroma.Web.ActivityPub.ObjectValidator.validate(activity, meta))
+        assert_called(Pleroma.Web.ActivityPub.MRF.filter(activity))
+        assert_called(Pleroma.Web.ActivityPub.ActivityPub.persist(activity, meta))
+        assert_called(Pleroma.Web.ActivityPub.SideEffects.handle(activity, meta))
+        assert_called(Pleroma.Web.Federator.publish(activity))
+      end
+    end
+
+    test "it goes through validation, filtering, persisting, side effects without federation for remote activities" do
+      activity = insert(:note_activity)
+      meta = [local: false]
+
+      with_mocks([
+        {Pleroma.Web.ActivityPub.ObjectValidator, [], [validate: fn o, m -> {:ok, o, m} end]},
+        {
+          Pleroma.Web.ActivityPub.MRF,
+          [],
+          [filter: fn o -> {:ok, o} end]
+        },
+        {
+          Pleroma.Web.ActivityPub.ActivityPub,
+          [],
+          [persist: fn o, m -> {:ok, o, m} end]
+        },
+        {
+          Pleroma.Web.ActivityPub.SideEffects,
+          [],
+          [handle: fn o, m -> {:ok, o, m} end]
+        },
+        {
+          Pleroma.Web.Federator,
+          [],
+          []
+        }
+      ]) do
+        assert {:ok, ^activity, ^meta} =
+                 Pleroma.Web.ActivityPub.Pipeline.common_pipeline(activity, meta)
+
+        assert_called(Pleroma.Web.ActivityPub.ObjectValidator.validate(activity, meta))
+        assert_called(Pleroma.Web.ActivityPub.MRF.filter(activity))
+        assert_called(Pleroma.Web.ActivityPub.ActivityPub.persist(activity, meta))
+        assert_called(Pleroma.Web.ActivityPub.SideEffects.handle(activity, meta))
+      end
+    end
+  end
+end