[#1149] Oban jobs implementation for :federator_incoming and :federator_outgoing queues.

6 files changed, 101 insertions, 25 deletions

diff --git a/test/activity_test.exs b/test/activity_test.exs
index b27f6fd36..b9c12adb2 100644
--- a/test/activity_test.exs
+++ b/test/activity_test.exs
@@ -6,6 +6,7 @@ defmodule Pleroma.ActivityTest do
   use Pleroma.DataCase
   alias Pleroma.Activity
   alias Pleroma.Bookmark
+  alias Pleroma.ObanHelpers
   alias Pleroma.Object
   alias Pleroma.ThreadMute
   import Pleroma.Factory
@@ -125,7 +126,8 @@ defmodule Pleroma.ActivityTest do
       }
 
       {:ok, local_activity} = Pleroma.Web.CommonAPI.post(user, %{"status" => "find me!"})
-      {:ok, remote_activity} = Pleroma.Web.Federator.incoming_ap_doc(params)
+      {:ok, job} = Pleroma.Web.Federator.incoming_ap_doc(params)
+      {:ok, remote_activity} = ObanHelpers.perform(job)
       %{local_activity: local_activity, remote_activity: remote_activity, user: user}
     end
 
diff --git a/test/support/oban_helpers.ex b/test/support/oban_helpers.ex
new file mode 100644
index 000000000..54b5a9566
--- /dev/null
+++ b/test/support/oban_helpers.ex
@@ -0,0 +1,36 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2018 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.ObanHelpers do
+  @moduledoc """
+  Oban test helpers.
+  """
+
+  alias Pleroma.Repo
+
+  def perform(%Oban.Job{} = job) do
+    res = apply(String.to_existing_atom("Elixir." <> job.worker), :perform, [job])
+    Repo.delete(job)
+    res
+  end
+
+  def perform(jobs) when is_list(jobs) do
+    for job <- jobs, do: perform(job)
+  end
+
+  def member?(%{} = job_args, jobs) when is_list(jobs) do
+    Enum.any?(jobs, fn job ->
+      member?(job_args, job.args)
+    end)
+  end
+
+  def member?(%{} = test_attrs, %{} = attrs) do
+    Enum.all?(
+      test_attrs,
+      fn {k, _v} -> member?(test_attrs[k], attrs[k]) end
+    )
+  end
+
+  def member?(x, y), do: x == y
+end
diff --git a/test/user_test.exs b/test/user_test.exs
index 70c376384..ee6d8e8f3 100644
--- a/test/user_test.exs
+++ b/test/user_test.exs
@@ -5,6 +5,7 @@
 defmodule Pleroma.UserTest do
   alias Pleroma.Activity
   alias Pleroma.Builders.UserBuilder
+  alias Pleroma.ObanHelpers
   alias Pleroma.Object
   alias Pleroma.Repo
   alias Pleroma.User
@@ -1044,8 +1045,16 @@ defmodule Pleroma.UserTest do
 
       {:ok, _user} = User.delete(user)
 
-      assert [%{args: %{"params" => %{"inbox" => "http://mastodon.example.org/inbox"}}}] =
+      assert ObanHelpers.member?(
+               %{
+                 "op" => "publish_one",
+                 "params" => %{
+                   "inbox" => "http://mastodon.example.org/inbox",
+                   "id" => "pleroma:fakeid"
+                 }
+               },
                all_enqueued(worker: Pleroma.Workers.Publisher)
+             )
 
       Pleroma.Config.put(config_path, initial_setting)
     end
diff --git a/test/web/activity_pub/activity_pub_controller_test.exs b/test/web/activity_pub/activity_pub_controller_test.exs
index 40344f17e..1d809164f 100644
--- a/test/web/activity_pub/activity_pub_controller_test.exs
+++ b/test/web/activity_pub/activity_pub_controller_test.exs
@@ -4,15 +4,19 @@
 
 defmodule Pleroma.Web.ActivityPub.ActivityPubControllerTest do
   use Pleroma.Web.ConnCase
+  use Oban.Testing, repo: Pleroma.Repo
+
   import Pleroma.Factory
   alias Pleroma.Activity
   alias Pleroma.Instances
+  alias Pleroma.ObanHelpers
   alias Pleroma.Object
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ObjectView
   alias Pleroma.Web.ActivityPub.UserView
   alias Pleroma.Web.ActivityPub.Utils
   alias Pleroma.Web.CommonAPI
+  alias Pleroma.Workers.Receiver, as: ReceiverWorker
 
   setup_all do
     Tesla.Mock.mock_global(fn env -> apply(HttpRequestMock, :request, [env]) end)
@@ -232,7 +236,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubControllerTest do
         |> post("/inbox", data)
 
       assert "ok" == json_response(conn, 200)
-      :timer.sleep(500)
+
+      ObanHelpers.perform(all_enqueued(worker: ReceiverWorker))
       assert Activity.get_by_ap_id(data["id"])
     end
 
@@ -274,7 +279,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubControllerTest do
         |> post("/users/#{user.nickname}/inbox", data)
 
       assert "ok" == json_response(conn, 200)
-      :timer.sleep(500)
+      ObanHelpers.perform(all_enqueued(worker: ReceiverWorker))
       assert Activity.get_by_ap_id(data["id"])
     end
 
@@ -303,7 +308,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubControllerTest do
         |> post("/users/#{recipient.nickname}/inbox", data)
 
       assert "ok" == json_response(conn, 200)
-      :timer.sleep(500)
+      ObanHelpers.perform(all_enqueued(worker: ReceiverWorker))
       assert Activity.get_by_ap_id(data["id"])
     end
 
@@ -382,6 +387,8 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubControllerTest do
       |> post("/users/#{recipient.nickname}/inbox", data)
       |> json_response(200)
 
+      ObanHelpers.perform(all_enqueued(worker: ReceiverWorker))
+
       activity = Activity.get_by_ap_id(data["id"])
 
       assert activity.id
@@ -457,6 +464,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubControllerTest do
         |> post("/users/#{user.nickname}/outbox", data)
 
       result = json_response(conn, 201)
+
       assert Activity.get_by_ap_id(result["id"])
     end
 
diff --git a/test/web/federator_test.exs b/test/web/federator_test.exs
index 5c1704548..ebe962da2 100644
--- a/test/web/federator_test.exs
+++ b/test/web/federator_test.exs
@@ -4,8 +4,10 @@
 
 defmodule Pleroma.Web.FederatorTest do
   alias Pleroma.Instances
+  alias Pleroma.ObanHelpers
   alias Pleroma.Web.CommonAPI
   alias Pleroma.Web.Federator
+  alias Pleroma.Workers.Publisher, as: PublisherWorker
 
   use Pleroma.DataCase
   use Oban.Testing, repo: Pleroma.Repo
@@ -45,6 +47,7 @@ defmodule Pleroma.Web.FederatorTest do
     } do
       with_mocks([relay_mock]) do
         Federator.publish(activity)
+        ObanHelpers.perform(all_enqueued(worker: PublisherWorker))
       end
 
       assert_received :relay_publish
@@ -58,6 +61,7 @@ defmodule Pleroma.Web.FederatorTest do
 
       with_mocks([relay_mock]) do
         Federator.publish(activity)
+        ObanHelpers.perform(all_enqueued(worker: PublisherWorker))
       end
 
       refute_received :relay_publish
@@ -97,8 +101,15 @@ defmodule Pleroma.Web.FederatorTest do
 
       expected_dt = NaiveDateTime.to_iso8601(dt)
 
-      assert [%{args: %{"params" => %{"inbox" => ^inbox1, "unreachable_since" => ^expected_dt}}}] =
-               all_enqueued(worker: Pleroma.Workers.Publisher)
+      ObanHelpers.perform(all_enqueued(worker: PublisherWorker))
+
+      assert ObanHelpers.member?(
+               %{
+                 "op" => "publish_one",
+                 "params" => %{"inbox" => inbox1, "unreachable_since" => expected_dt}
+               },
+               all_enqueued(worker: PublisherWorker)
+             )
     end
 
     test "it federates only to reachable instances via Websub" do
@@ -129,16 +140,18 @@ defmodule Pleroma.Web.FederatorTest do
       expected_callback = sub2.callback
       expected_dt = NaiveDateTime.to_iso8601(dt)
 
-      assert [
+      ObanHelpers.perform(all_enqueued(worker: PublisherWorker))
+
+      assert ObanHelpers.member?(
                %{
-                 args: %{
-                   "params" => %{
-                     "callback" => ^expected_callback,
-                     "unreachable_since" => ^expected_dt
-                   }
+                 "op" => "publish_one",
+                 "params" => %{
+                   "callback" => expected_callback,
+                   "unreachable_since" => expected_dt
                  }
-               }
-             ] = all_enqueued(worker: Pleroma.Workers.Publisher)
+               },
+               all_enqueued(worker: PublisherWorker)
+             )
     end
 
     test "it federates only to reachable instances via Salmon" do
@@ -172,16 +185,18 @@ defmodule Pleroma.Web.FederatorTest do
 
       expected_dt = NaiveDateTime.to_iso8601(dt)
 
-      assert [
+      ObanHelpers.perform(all_enqueued(worker: PublisherWorker))
+
+      assert ObanHelpers.member?(
                %{
-                 args: %{
-                   "params" => %{
-                     "recipient_id" => ^remote_user2_id,
-                     "unreachable_since" => ^expected_dt
-                   }
+                 "op" => "publish_one",
+                 "params" => %{
+                   "recipient_id" => remote_user2_id,
+                   "unreachable_since" => expected_dt
                  }
-               }
-             ] = all_enqueued(worker: Pleroma.Workers.Publisher)
+               },
+               all_enqueued(worker: PublisherWorker)
+             )
     end
   end
 
@@ -201,7 +216,8 @@ defmodule Pleroma.Web.FederatorTest do
         "to" => ["https://www.w3.org/ns/activitystreams#Public"]
       }
 
-      {:ok, _activity} = Federator.incoming_ap_doc(params)
+      assert {:ok, job} = Federator.incoming_ap_doc(params)
+      assert {:ok, _activity} = ObanHelpers.perform(job)
     end
 
     test "rejects incoming AP docs with incorrect origin" do
@@ -219,7 +235,8 @@ defmodule Pleroma.Web.FederatorTest do
         "to" => ["https://www.w3.org/ns/activitystreams#Public"]
       }
 
-      :error = Federator.incoming_ap_doc(params)
+      assert {:ok, job} = Federator.incoming_ap_doc(params)
+      assert :error = ObanHelpers.perform(job)
     end
   end
 end
diff --git a/test/web/websub/websub_test.exs b/test/web/websub/websub_test.exs
index 74386d7db..b704a558a 100644
--- a/test/web/websub/websub_test.exs
+++ b/test/web/websub/websub_test.exs
@@ -4,11 +4,14 @@
 
 defmodule Pleroma.Web.WebsubTest do
   use Pleroma.DataCase
+  use Oban.Testing, repo: Pleroma.Repo
 
+  alias Pleroma.ObanHelpers
   alias Pleroma.Web.Router.Helpers
   alias Pleroma.Web.Websub
   alias Pleroma.Web.Websub.WebsubClientSubscription
   alias Pleroma.Web.Websub.WebsubServerSubscription
+  alias Pleroma.Workers.Subscriber, as: SubscriberWorker
 
   import Pleroma.Factory
   import Tesla.Mock
@@ -224,6 +227,7 @@ defmodule Pleroma.Web.WebsubTest do
         })
 
       _refresh = Websub.refresh_subscriptions()
+      ObanHelpers.perform(all_enqueued(worker: SubscriberWorker))
 
       assert still_good == Repo.get(WebsubClientSubscription, still_good.id)
       refute needs_refresh == Repo.get(WebsubClientSubscription, needs_refresh.id)