Merge branch 'develop' into feature/moderation-log-filters

21 files changed, 656 insertions, 256 deletions

diff --git a/test/web/activity_pub/activity_pub_controller_test.exs b/test/web/activity_pub/activity_pub_controller_test.exs
index 9698c7099..f83b14452 100644
--- a/test/web/activity_pub/activity_pub_controller_test.exs
+++ b/test/web/activity_pub/activity_pub_controller_test.exs
@@ -4,16 +4,21 @@
 
 defmodule Pleroma.Web.ActivityPub.ActivityPubControllerTest do
   use Pleroma.Web.ConnCase
+  use Oban.Testing, repo: Pleroma.Repo
+
   import Pleroma.Factory
   alias Pleroma.Activity
+  alias Pleroma.Delivery
   alias Pleroma.Instances
   alias Pleroma.Object
+  alias Pleroma.Tests.ObanHelpers
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ObjectView
   alias Pleroma.Web.ActivityPub.Relay
   alias Pleroma.Web.ActivityPub.UserView
   alias Pleroma.Web.ActivityPub.Utils
   alias Pleroma.Web.CommonAPI
+  alias Pleroma.Workers.ReceiverWorker
 
   setup_all do
     Tesla.Mock.mock_global(fn env -> apply(HttpRequestMock, :request, [env]) end)
@@ -365,7 +370,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
 
@@ -407,7 +413,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
 
@@ -436,7 +442,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
 
@@ -526,6 +532,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
@@ -601,6 +609,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
 
@@ -885,4 +894,86 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubControllerTest do
       assert result["totalItems"] == 15
     end
   end
+
+  describe "delivery tracking" do
+    test "it tracks a signed object fetch", %{conn: conn} do
+      user = insert(:user, local: false)
+      activity = insert(:note_activity)
+      object = Object.normalize(activity)
+
+      object_path = String.trim_leading(object.data["id"], Pleroma.Web.Endpoint.url())
+
+      conn
+      |> put_req_header("accept", "application/activity+json")
+      |> assign(:user, user)
+      |> get(object_path)
+      |> json_response(200)
+
+      assert Delivery.get(object.id, user.id)
+    end
+
+    test "it tracks a signed activity fetch", %{conn: conn} do
+      user = insert(:user, local: false)
+      activity = insert(:note_activity)
+      object = Object.normalize(activity)
+
+      activity_path = String.trim_leading(activity.data["id"], Pleroma.Web.Endpoint.url())
+
+      conn
+      |> put_req_header("accept", "application/activity+json")
+      |> assign(:user, user)
+      |> get(activity_path)
+      |> json_response(200)
+
+      assert Delivery.get(object.id, user.id)
+    end
+
+    test "it tracks a signed object fetch when the json is cached", %{conn: conn} do
+      user = insert(:user, local: false)
+      other_user = insert(:user, local: false)
+      activity = insert(:note_activity)
+      object = Object.normalize(activity)
+
+      object_path = String.trim_leading(object.data["id"], Pleroma.Web.Endpoint.url())
+
+      conn
+      |> put_req_header("accept", "application/activity+json")
+      |> assign(:user, user)
+      |> get(object_path)
+      |> json_response(200)
+
+      build_conn()
+      |> put_req_header("accept", "application/activity+json")
+      |> assign(:user, other_user)
+      |> get(object_path)
+      |> json_response(200)
+
+      assert Delivery.get(object.id, user.id)
+      assert Delivery.get(object.id, other_user.id)
+    end
+
+    test "it tracks a signed activity fetch when the json is cached", %{conn: conn} do
+      user = insert(:user, local: false)
+      other_user = insert(:user, local: false)
+      activity = insert(:note_activity)
+      object = Object.normalize(activity)
+
+      activity_path = String.trim_leading(activity.data["id"], Pleroma.Web.Endpoint.url())
+
+      conn
+      |> put_req_header("accept", "application/activity+json")
+      |> assign(:user, user)
+      |> get(activity_path)
+      |> json_response(200)
+
+      build_conn()
+      |> put_req_header("accept", "application/activity+json")
+      |> assign(:user, other_user)
+      |> get(activity_path)
+      |> json_response(200)
+
+      assert Delivery.get(object.id, user.id)
+      assert Delivery.get(object.id, other_user.id)
+    end
+  end
 end
diff --git a/test/web/activity_pub/activity_pub_test.exs b/test/web/activity_pub/activity_pub_test.exs
index f72b44aed..4100108a5 100644
--- a/test/web/activity_pub/activity_pub_test.exs
+++ b/test/web/activity_pub/activity_pub_test.exs
@@ -38,9 +38,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
         stream: fn _, _ -> nil end do
         ActivityPub.stream_out_participations(conversation.participations)
 
-        Enum.each(participations, fn participation ->
-          assert called(Pleroma.Web.Streamer.stream("participation", participation))
-        end)
+        assert called(Pleroma.Web.Streamer.stream("participation", participations))
       end
     end
   end
@@ -686,7 +684,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
       user = insert(:user)
 
       {:ok, like_activity, _object} = ActivityPub.like(user, object_activity)
-      assert called(Pleroma.Web.Federator.publish(like_activity, 5))
+      assert called(Pleroma.Web.Federator.publish(like_activity))
     end
 
     test "returns exist activity if object already liked" do
@@ -747,7 +745,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
       {:ok, unlike_activity, _, object} = ActivityPub.unlike(user, object)
       assert object.data["like_count"] == 0
 
-      assert called(Pleroma.Web.Federator.publish(unlike_activity, 5))
+      assert called(Pleroma.Web.Federator.publish(unlike_activity))
     end
 
     test "unliking a previously liked object" do
diff --git a/test/web/activity_pub/mrf/mediaproxy_warming_policy_test.exs b/test/web/activity_pub/mrf/mediaproxy_warming_policy_test.exs
index 372e789be..95a809d25 100644
--- a/test/web/activity_pub/mrf/mediaproxy_warming_policy_test.exs
+++ b/test/web/activity_pub/mrf/mediaproxy_warming_policy_test.exs
@@ -6,6 +6,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicyTest do
   use Pleroma.DataCase
 
   alias Pleroma.HTTP
+  alias Pleroma.Tests.ObanHelpers
   alias Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicy
 
   import Mock
@@ -24,6 +25,11 @@ defmodule Pleroma.Web.ActivityPub.MRF.MediaProxyWarmingPolicyTest do
   test "it prefetches media proxy URIs" do
     with_mock HTTP, get: fn _, _, _ -> {:ok, []} end do
       MediaProxyWarmingPolicy.filter(@message)
+
+      ObanHelpers.perform_all()
+      # Performing jobs which has been just enqueued
+      ObanHelpers.perform_all()
+
       assert called(HTTP.get(:_, :_, :_))
     end
   end
diff --git a/test/web/activity_pub/publisher_test.exs b/test/web/activity_pub/publisher_test.exs
index 36a39c84c..df03b4008 100644
--- a/test/web/activity_pub/publisher_test.exs
+++ b/test/web/activity_pub/publisher_test.exs
@@ -3,15 +3,18 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 
 defmodule Pleroma.Web.ActivityPub.PublisherTest do
-  use Pleroma.DataCase
+  use Pleroma.Web.ConnCase
 
+  import ExUnit.CaptureLog
   import Pleroma.Factory
   import Tesla.Mock
   import Mock
 
   alias Pleroma.Activity
   alias Pleroma.Instances
+  alias Pleroma.Object
   alias Pleroma.Web.ActivityPub.Publisher
+  alias Pleroma.Web.CommonAPI
 
   @as_public "https://www.w3.org/ns/activitystreams#Public"
 
@@ -188,7 +191,10 @@ defmodule Pleroma.Web.ActivityPub.PublisherTest do
       actor = insert(:user)
       inbox = "http://connrefused.site/users/nick1/inbox"
 
-      assert {:error, _} = Publisher.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1})
+      assert capture_log(fn ->
+               assert {:error, _} =
+                        Publisher.publish_one(%{inbox: inbox, json: "{}", actor: actor, id: 1})
+             end) =~ "connrefused"
 
       assert called(Instances.set_unreachable(inbox))
     end
@@ -212,14 +218,16 @@ defmodule Pleroma.Web.ActivityPub.PublisherTest do
       actor = insert(:user)
       inbox = "http://connrefused.site/users/nick1/inbox"
 
-      assert {:error, _} =
-               Publisher.publish_one(%{
-                 inbox: inbox,
-                 json: "{}",
-                 actor: actor,
-                 id: 1,
-                 unreachable_since: NaiveDateTime.utc_now()
-               })
+      assert capture_log(fn ->
+               assert {:error, _} =
+                        Publisher.publish_one(%{
+                          inbox: inbox,
+                          json: "{}",
+                          actor: actor,
+                          id: 1,
+                          unreachable_since: NaiveDateTime.utc_now()
+                        })
+             end) =~ "connrefused"
 
       refute called(Instances.set_unreachable(inbox))
     end
@@ -257,10 +265,74 @@ defmodule Pleroma.Web.ActivityPub.PublisherTest do
       assert called(
                Pleroma.Web.Federator.Publisher.enqueue_one(Publisher, %{
                  inbox: "https://domain.com/users/nick1/inbox",
-                 actor: actor,
+                 actor_id: actor.id,
                  id: note_activity.data["id"]
                })
              )
     end
+
+    test_with_mock "publishes a delete activity to peers who signed fetch requests to the create acitvity/object.",
+                   Pleroma.Web.Federator.Publisher,
+                   [:passthrough],
+                   [] do
+      fetcher =
+        insert(:user,
+          local: false,
+          info: %{
+            ap_enabled: true,
+            source_data: %{"inbox" => "https://domain.com/users/nick1/inbox"}
+          }
+        )
+
+      another_fetcher =
+        insert(:user,
+          local: false,
+          info: %{
+            ap_enabled: true,
+            source_data: %{"inbox" => "https://domain2.com/users/nick1/inbox"}
+          }
+        )
+
+      actor = insert(:user)
+
+      note_activity = insert(:note_activity, user: actor)
+      object = Object.normalize(note_activity)
+
+      activity_path = String.trim_leading(note_activity.data["id"], Pleroma.Web.Endpoint.url())
+      object_path = String.trim_leading(object.data["id"], Pleroma.Web.Endpoint.url())
+
+      build_conn()
+      |> put_req_header("accept", "application/activity+json")
+      |> assign(:user, fetcher)
+      |> get(object_path)
+      |> json_response(200)
+
+      build_conn()
+      |> put_req_header("accept", "application/activity+json")
+      |> assign(:user, another_fetcher)
+      |> get(activity_path)
+      |> json_response(200)
+
+      {:ok, delete} = CommonAPI.delete(note_activity.id, actor)
+
+      res = Publisher.publish(actor, delete)
+      assert res == :ok
+
+      assert called(
+               Pleroma.Web.Federator.Publisher.enqueue_one(Publisher, %{
+                 inbox: "https://domain.com/users/nick1/inbox",
+                 actor_id: actor.id,
+                 id: delete.data["id"]
+               })
+             )
+
+      assert called(
+               Pleroma.Web.Federator.Publisher.enqueue_one(Publisher, %{
+                 inbox: "https://domain2.com/users/nick1/inbox",
+                 actor_id: actor.id,
+                 id: delete.data["id"]
+               })
+             )
+    end
   end
 end
diff --git a/test/web/activity_pub/relay_test.exs b/test/web/activity_pub/relay_test.exs
index 4f7d592a6..7315dce26 100644
--- a/test/web/activity_pub/relay_test.exs
+++ b/test/web/activity_pub/relay_test.exs
@@ -10,6 +10,7 @@ defmodule Pleroma.Web.ActivityPub.RelayTest do
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.ActivityPub.Relay
 
+  import ExUnit.CaptureLog
   import Pleroma.Factory
   import Mock
 
@@ -20,7 +21,9 @@ defmodule Pleroma.Web.ActivityPub.RelayTest do
 
   describe "follow/1" do
     test "returns errors when user not found" do
-      assert Relay.follow("test-ap-id") == {:error, "Could not fetch by AP id"}
+      assert capture_log(fn ->
+               assert Relay.follow("test-ap-id") == {:error, "Could not fetch by AP id"}
+             end) =~ "Could not fetch by AP id"
     end
 
     test "returns activity" do
@@ -37,7 +40,9 @@ defmodule Pleroma.Web.ActivityPub.RelayTest do
 
   describe "unfollow/1" do
     test "returns errors when user not found" do
-      assert Relay.unfollow("test-ap-id") == {:error, "Could not fetch by AP id"}
+      assert capture_log(fn ->
+               assert Relay.unfollow("test-ap-id") == {:error, "Could not fetch by AP id"}
+             end) =~ "Could not fetch by AP id"
     end
 
     test "returns activity" do
@@ -78,7 +83,9 @@ defmodule Pleroma.Web.ActivityPub.RelayTest do
           }
         )
 
-      assert Relay.publish(activity) == {:error, nil}
+      assert capture_log(fn ->
+               assert Relay.publish(activity) == {:error, nil}
+             end) =~ "[error] error: nil"
     end
 
     test_with_mock "returns announce activity and publish to federate",
@@ -92,7 +99,7 @@ defmodule Pleroma.Web.ActivityPub.RelayTest do
       assert activity.data["type"] == "Announce"
       assert activity.data["actor"] == service_actor.ap_id
       assert activity.data["object"] == obj.data["id"]
-      assert called(Pleroma.Web.Federator.publish(activity, 5))
+      assert called(Pleroma.Web.Federator.publish(activity))
     end
 
     test_with_mock "returns announce activity and not publish to federate",
@@ -106,7 +113,7 @@ defmodule Pleroma.Web.ActivityPub.RelayTest do
       assert activity.data["type"] == "Announce"
       assert activity.data["actor"] == service_actor.ap_id
       assert activity.data["object"] == obj.data["id"]
-      refute called(Pleroma.Web.Federator.publish(activity, 5))
+      refute called(Pleroma.Web.Federator.publish(activity))
     end
   end
 end
diff --git a/test/web/activity_pub/transmogrifier_test.exs b/test/web/activity_pub/transmogrifier_test.exs
index 0661d5d7c..6c296eb0d 100644
--- a/test/web/activity_pub/transmogrifier_test.exs
+++ b/test/web/activity_pub/transmogrifier_test.exs
@@ -8,6 +8,7 @@ defmodule Pleroma.Web.ActivityPub.TransmogrifierTest do
   alias Pleroma.Object
   alias Pleroma.Object.Fetcher
   alias Pleroma.Repo
+  alias Pleroma.Tests.ObanHelpers
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.ActivityPub.Transmogrifier
@@ -102,7 +103,7 @@ defmodule Pleroma.Web.ActivityPub.TransmogrifierTest do
 
       assert capture_log(fn ->
                {:ok, _returned_activity} = Transmogrifier.handle_incoming(data)
-             end) =~ "[error] Couldn't fetch \"\"https://404.site/whatever\"\", error: nil"
+             end) =~ "[error] Couldn't fetch \"https://404.site/whatever\", error: nil"
     end
 
     test "it works for incoming notices" do
@@ -648,6 +649,7 @@ defmodule Pleroma.Web.ActivityPub.TransmogrifierTest do
         |> Poison.decode!()
 
       {:ok, _} = Transmogrifier.handle_incoming(data)
+      ObanHelpers.perform_all()
 
       refute User.get_cached_by_ap_id(ap_id)
     end
@@ -1210,6 +1212,8 @@ defmodule Pleroma.Web.ActivityPub.TransmogrifierTest do
       assert user.info.note_count == 1
 
       {:ok, user} = Transmogrifier.upgrade_user_from_ap_id("https://niu.moe/users/rye")
+      ObanHelpers.perform_all()
+
       assert user.info.ap_enabled
       assert user.info.note_count == 1
       assert user.follower_address == "https://niu.moe/users/rye/followers"
diff --git a/test/web/admin_api/admin_api_controller_test.exs b/test/web/admin_api/admin_api_controller_test.exs
index b87fffc34..66804faac 100644
--- a/test/web/admin_api/admin_api_controller_test.exs
+++ b/test/web/admin_api/admin_api_controller_test.exs
@@ -1779,7 +1779,11 @@ defmodule Pleroma.Web.AdminAPI.AdminAPIControllerTest do
                 %{"tuple" => [":seconds_valid", 60]},
                 %{"tuple" => [":path", ""]},
                 %{"tuple" => [":key1", nil]},
-                %{"tuple" => [":partial_chain", "&:hackney_connect.partial_chain/1"]}
+                %{"tuple" => [":partial_chain", "&:hackney_connect.partial_chain/1"]},
+                %{"tuple" => [":regex1", "~r/https:\/\/example.com/"]},
+                %{"tuple" => [":regex2", "~r/https:\/\/example.com/u"]},
+                %{"tuple" => [":regex3", "~r/https:\/\/example.com/i"]},
+                %{"tuple" => [":regex4", "~r/https:\/\/example.com/s"]}
               ]
             }
           ]
@@ -1796,7 +1800,11 @@ defmodule Pleroma.Web.AdminAPI.AdminAPIControllerTest do
                      %{"tuple" => [":seconds_valid", 60]},
                      %{"tuple" => [":path", ""]},
                      %{"tuple" => [":key1", nil]},
-                     %{"tuple" => [":partial_chain", "&:hackney_connect.partial_chain/1"]}
+                     %{"tuple" => [":partial_chain", "&:hackney_connect.partial_chain/1"]},
+                     %{"tuple" => [":regex1", "~r/https:\\/\\/example.com/"]},
+                     %{"tuple" => [":regex2", "~r/https:\\/\\/example.com/u"]},
+                     %{"tuple" => [":regex3", "~r/https:\\/\\/example.com/i"]},
+                     %{"tuple" => [":regex4", "~r/https:\\/\\/example.com/s"]}
                    ]
                  }
                ]
@@ -2088,7 +2096,7 @@ defmodule Pleroma.Web.AdminAPI.AdminAPIControllerTest do
         post(conn, "/api/pleroma/admin/config", %{
           configs: [
             %{
-              "group" => "pleroma_job_queue",
+              "group" => "oban",
               "key" => ":queues",
               "value" => [
                 %{"tuple" => [":federator_incoming", 50]},
@@ -2106,7 +2114,7 @@ defmodule Pleroma.Web.AdminAPI.AdminAPIControllerTest do
       assert json_response(conn, 200) == %{
                "configs" => [
                  %{
-                   "group" => "pleroma_job_queue",
+                   "group" => "oban",
                    "key" => ":queues",
                    "value" => [
                      %{"tuple" => [":federator_incoming", 50]},
diff --git a/test/web/admin_api/config_test.exs b/test/web/admin_api/config_test.exs
index 3190dc1c8..204446b79 100644
--- a/test/web/admin_api/config_test.exs
+++ b/test/web/admin_api/config_test.exs
@@ -103,6 +103,30 @@ defmodule Pleroma.Web.AdminAPI.ConfigTest do
       assert Config.from_binary(binary) == ~r/comp[lL][aA][iI][nN]er/
     end
 
+    test "link sigil" do
+      binary = Config.transform("~r/https:\/\/example.com/")
+      assert binary == :erlang.term_to_binary(~r/https:\/\/example.com/)
+      assert Config.from_binary(binary) == ~r/https:\/\/example.com/
+    end
+
+    test "link sigil with u modifier" do
+      binary = Config.transform("~r/https:\/\/example.com/u")
+      assert binary == :erlang.term_to_binary(~r/https:\/\/example.com/u)
+      assert Config.from_binary(binary) == ~r/https:\/\/example.com/u
+    end
+
+    test "link sigil with i modifier" do
+      binary = Config.transform("~r/https:\/\/example.com/i")
+      assert binary == :erlang.term_to_binary(~r/https:\/\/example.com/i)
+      assert Config.from_binary(binary) == ~r/https:\/\/example.com/i
+    end
+
+    test "link sigil with s modifier" do
+      binary = Config.transform("~r/https:\/\/example.com/s")
+      assert binary == :erlang.term_to_binary(~r/https:\/\/example.com/s)
+      assert Config.from_binary(binary) == ~r/https:\/\/example.com/s
+    end
+
     test "2 child tuple" do
       binary = Config.transform(%{"tuple" => ["v1", ":v2"]})
       assert binary == :erlang.term_to_binary({"v1", :v2})
diff --git a/test/web/digest_email_worker_test.exs b/test/web/digest_email_worker_test.exs
deleted file mode 100644
index 15002330f..000000000
--- a/test/web/digest_email_worker_test.exs
+++ /dev/null
@@ -1,31 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.DigestEmailWorkerTest do
-  use Pleroma.DataCase
-  import Pleroma.Factory
-
-  alias Pleroma.DigestEmailWorker
-  alias Pleroma.User
-  alias Pleroma.Web.CommonAPI
-
-  test "it sends digest emails" do
-    user = insert(:user)
-
-    date =
-      Timex.now()
-      |> Timex.shift(days: -10)
-      |> Timex.to_naive_datetime()
-
-    user2 = insert(:user, last_digest_emailed_at: date)
-    User.switch_email_notifications(user2, "digest", true)
-    CommonAPI.post(user, %{"status" => "hey @#{user2.nickname}!"})
-
-    DigestEmailWorker.perform()
-
-    assert_received {:email, email}
-    assert email.to == [{user2.name, user2.email}]
-    assert email.subject == "Your digest from #{Pleroma.Config.get(:instance)[:name]}"
-  end
-end
diff --git a/test/web/federator_test.exs b/test/web/federator_test.exs
index 09e54533f..4096d4690 100644
--- a/test/web/federator_test.exs
+++ b/test/web/federator_test.exs
@@ -4,9 +4,14 @@
 
 defmodule Pleroma.Web.FederatorTest do
   alias Pleroma.Instances
+  alias Pleroma.Tests.ObanHelpers
   alias Pleroma.Web.CommonAPI
   alias Pleroma.Web.Federator
+  alias Pleroma.Workers.PublisherWorker
+
   use Pleroma.DataCase
+  use Oban.Testing, repo: Pleroma.Repo
+
   import Pleroma.Factory
   import Mock
 
@@ -24,15 +29,6 @@ defmodule Pleroma.Web.FederatorTest do
   clear_config([:instance, :rewrite_policy])
   clear_config([:mrf_keyword])
 
-  describe "Publisher.perform" do
-    test "call `perform` with unknown task" do
-      assert {
-               :error,
-               "Don't know what to do with this"
-             } = Pleroma.Web.Federator.Publisher.perform("test", :ok, :ok)
-    end
-  end
-
   describe "Publish an activity" do
     setup do
       user = insert(:user)
@@ -53,6 +49,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
@@ -66,6 +63,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
@@ -73,10 +71,7 @@ defmodule Pleroma.Web.FederatorTest do
   end
 
   describe "Targets reachability filtering in `publish`" do
-    test_with_mock "it federates only to reachable instances via AP",
-                   Pleroma.Web.ActivityPub.Publisher,
-                   [:passthrough],
-                   [] do
+    test "it federates only to reachable instances via AP" do
       user = insert(:user)
 
       {inbox1, inbox2} =
@@ -104,20 +99,20 @@ defmodule Pleroma.Web.FederatorTest do
       {:ok, _activity} =
         CommonAPI.post(user, %{"status" => "HI @nick1@domain.com, @nick2@domain2.com!"})
 
-      assert called(
-               Pleroma.Web.ActivityPub.Publisher.publish_one(%{
-                 inbox: inbox1,
-                 unreachable_since: dt
-               })
-             )
+      expected_dt = NaiveDateTime.to_iso8601(dt)
 
-      refute called(Pleroma.Web.ActivityPub.Publisher.publish_one(%{inbox: inbox2}))
+      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_with_mock "it federates only to reachable instances via Websub",
-                   Pleroma.Web.Websub,
-                   [:passthrough],
-                   [] do
+    test "it federates only to reachable instances via Websub" do
       user = insert(:user)
       websub_topic = Pleroma.Web.OStatus.feed_path(user)
 
@@ -142,23 +137,27 @@ defmodule Pleroma.Web.FederatorTest do
 
       {:ok, _activity} = CommonAPI.post(user, %{"status" => "HI"})
 
-      assert called(
-               Pleroma.Web.Websub.publish_one(%{
-                 callback: sub2.callback,
-                 unreachable_since: dt
-               })
-             )
+      expected_callback = sub2.callback
+      expected_dt = NaiveDateTime.to_iso8601(dt)
+
+      ObanHelpers.perform(all_enqueued(worker: PublisherWorker))
 
-      refute called(Pleroma.Web.Websub.publish_one(%{callback: sub1.callback}))
+      assert ObanHelpers.member?(
+               %{
+                 "op" => "publish_one",
+                 "params" => %{
+                   "callback" => expected_callback,
+                   "unreachable_since" => expected_dt
+                 }
+               },
+               all_enqueued(worker: PublisherWorker)
+             )
     end
 
-    test_with_mock "it federates only to reachable instances via Salmon",
-                   Pleroma.Web.Salmon,
-                   [:passthrough],
-                   [] do
+    test "it federates only to reachable instances via Salmon" do
       user = insert(:user)
 
-      remote_user1 =
+      _remote_user1 =
         insert(:user, %{
           local: false,
           nickname: "nick1@domain.com",
@@ -174,6 +173,8 @@ defmodule Pleroma.Web.FederatorTest do
           info: %{salmon: "https://domain2.com/salmon"}
         })
 
+      remote_user2_id = remote_user2.id
+
       dt = NaiveDateTime.utc_now()
       Instances.set_unreachable(remote_user2.ap_id, dt)
 
@@ -182,14 +183,20 @@ defmodule Pleroma.Web.FederatorTest do
       {:ok, _activity} =
         CommonAPI.post(user, %{"status" => "HI @nick1@domain.com, @nick2@domain2.com!"})
 
-      assert called(
-               Pleroma.Web.Salmon.publish_one(%{
-                 recipient: remote_user2,
-                 unreachable_since: dt
-               })
-             )
+      expected_dt = NaiveDateTime.to_iso8601(dt)
+
+      ObanHelpers.perform(all_enqueued(worker: PublisherWorker))
 
-      refute called(Pleroma.Web.Salmon.publish_one(%{recipient: remote_user1}))
+      assert ObanHelpers.member?(
+               %{
+                 "op" => "publish_one",
+                 "params" => %{
+                   "recipient_id" => remote_user2_id,
+                   "unreachable_since" => expected_dt
+                 }
+               },
+               all_enqueued(worker: PublisherWorker)
+             )
     end
   end
 
@@ -209,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
@@ -227,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
 
     test "it does not crash if MRF rejects the post" do
@@ -242,7 +251,8 @@ defmodule Pleroma.Web.FederatorTest do
         File.read!("test/fixtures/mastodon-post-activity.json")
         |> Poison.decode!()
 
-      assert Federator.incoming_ap_doc(params) == :error
+      assert {:ok, job} = Federator.incoming_ap_doc(params)
+      assert :error = ObanHelpers.perform(job)
     end
   end
 end
diff --git a/test/web/instances/instance_test.exs b/test/web/instances/instance_test.exs
index 3fd011fd3..0b53bc6cd 100644
--- a/test/web/instances/instance_test.exs
+++ b/test/web/instances/instance_test.exs
@@ -16,7 +16,8 @@ defmodule Pleroma.Instances.InstanceTest do
 
   describe "set_reachable/1" do
     test "clears `unreachable_since` of existing matching Instance record having non-nil `unreachable_since`" do
-      instance = insert(:instance, unreachable_since: NaiveDateTime.utc_now())
+      unreachable_since = NaiveDateTime.to_iso8601(NaiveDateTime.utc_now())
+      instance = insert(:instance, unreachable_since: unreachable_since)
 
       assert {:ok, instance} = Instance.set_reachable(instance.host)
       refute instance.unreachable_since
diff --git a/test/web/mastodon_api/mastodon_api_controller_test.exs b/test/web/mastodon_api/mastodon_api_controller_test.exs
index e18f8f0d1..fb04748bb 100644
--- a/test/web/mastodon_api/mastodon_api_controller_test.exs
+++ b/test/web/mastodon_api/mastodon_api_controller_test.exs
@@ -13,6 +13,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIControllerTest do
   alias Pleroma.Object
   alias Pleroma.Repo
   alias Pleroma.ScheduledActivity
+  alias Pleroma.Tests.ObanHelpers
   alias Pleroma.User
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.CommonAPI
@@ -744,6 +745,16 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIControllerTest do
     assert id == to_string(activity.id)
   end
 
+  test "get statuses by IDs", %{conn: conn} do
+    %{id: id1} = insert(:note_activity)
+    %{id: id2} = insert(:note_activity)
+
+    query_string = "ids[]=#{id1}&ids[]=#{id2}"
+    conn = get(conn, "/api/v1/statuses/?#{query_string}")
+
+    assert [%{"id" => ^id1}, %{"id" => ^id2}] = Enum.sort_by(json_response(conn, :ok), & &1["id"])
+  end
+
   describe "deleting a status" do
     test "when you created it", %{conn: conn} do
       activity = insert(:note_activity)
@@ -3688,7 +3699,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIControllerTest do
         build_conn()
         |> assign(:user, user)
 
-      [conn: conn, activity: activity]
+      [conn: conn, activity: activity, user: user]
     end
 
     test "returns users who have favorited the status", %{conn: conn, activity: activity} do
@@ -3748,6 +3759,32 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIControllerTest do
       [%{"id" => id}] = response
       assert id == other_user.id
     end
+
+    test "requires authentification for private posts", %{conn: conn, user: user} do
+      other_user = insert(:user)
+
+      {:ok, activity} =
+        CommonAPI.post(user, %{
+          "status" => "@#{other_user.nickname} wanna get some #cofe together?",
+          "visibility" => "direct"
+        })
+
+      {:ok, _, _} = CommonAPI.favorite(activity.id, other_user)
+
+      conn
+      |> assign(:user, nil)
+      |> get("/api/v1/statuses/#{activity.id}/favourited_by")
+      |> json_response(404)
+
+      response =
+        build_conn()
+        |> assign(:user, other_user)
+        |> get("/api/v1/statuses/#{activity.id}/favourited_by")
+        |> json_response(200)
+
+      [%{"id" => id}] = response
+      assert id == other_user.id
+    end
   end
 
   describe "GET /api/v1/statuses/:id/reblogged_by" do
@@ -3759,7 +3796,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIControllerTest do
         build_conn()
         |> assign(:user, user)
 
-      [conn: conn, activity: activity]
+      [conn: conn, activity: activity, user: user]
     end
 
     test "returns users who have reblogged the status", %{conn: conn, activity: activity} do
@@ -3819,6 +3856,29 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIControllerTest do
       [%{"id" => id}] = response
       assert id == other_user.id
     end
+
+    test "requires authentification for private posts", %{conn: conn, user: user} do
+      other_user = insert(:user)
+
+      {:ok, activity} =
+        CommonAPI.post(user, %{
+          "status" => "@#{other_user.nickname} wanna get some #cofe together?",
+          "visibility" => "direct"
+        })
+
+      conn
+      |> assign(:user, nil)
+      |> get("/api/v1/statuses/#{activity.id}/reblogged_by")
+      |> json_response(404)
+
+      response =
+        build_conn()
+        |> assign(:user, other_user)
+        |> get("/api/v1/statuses/#{activity.id}/reblogged_by")
+        |> json_response(200)
+
+      assert [] == response
+    end
   end
 
   describe "POST /auth/password, with valid parameters" do
@@ -3838,6 +3898,7 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIControllerTest do
     end
 
     test "it sends an email to user", %{user: user} do
+      ObanHelpers.perform_all()
       token_record = Repo.get_by(Pleroma.PasswordResetToken, user_id: user.id)
 
       email = Pleroma.Emails.UserEmail.password_reset_email(user, token_record.token)
@@ -3898,6 +3959,8 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIControllerTest do
       |> post("/api/v1/pleroma/accounts/confirmation_resend?email=#{user.email}")
       |> json_response(:no_content)
 
+      ObanHelpers.perform_all()
+
       email = Pleroma.Emails.UserEmail.account_confirmation_email(user)
       notify_email = Config.get([:instance, :notify_email])
       instance_name = Config.get([:instance, :name])
@@ -3953,13 +4016,15 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIControllerTest do
       Config.put([:suggestions, :enabled], true)
       Config.put([:suggestions, :third_party_engine], "http://test500?{{host}}&{{user}}")
 
-      res =
-        conn
-        |> assign(:user, user)
-        |> get("/api/v1/suggestions")
-        |> json_response(500)
+      assert capture_log(fn ->
+               res =
+                 conn
+                 |> assign(:user, user)
+                 |> get("/api/v1/suggestions")
+                 |> json_response(500)
 
-      assert res == "Something went wrong"
+               assert res == "Something went wrong"
+             end) =~ "Could not retrieve suggestions"
     end
 
     test "returns suggestions", %{conn: conn, user: user, other_user: other_user} do
diff --git a/test/web/retry_queue_test.exs b/test/web/retry_queue_test.exs
deleted file mode 100644
index ecb3ce5d0..000000000
--- a/test/web/retry_queue_test.exs
+++ /dev/null
@@ -1,48 +0,0 @@
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2018 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule MockActivityPub do
-  def publish_one({ret, waiter}) do
-    send(waiter, :complete)
-    {ret, "success"}
-  end
-end
-
-defmodule Pleroma.Web.Federator.RetryQueueTest do
-  use Pleroma.DataCase
-  alias Pleroma.Web.Federator.RetryQueue
-
-  @small_retry_count 0
-  @hopeless_retry_count 10
-
-  setup do
-    RetryQueue.reset_stats()
-  end
-
-  test "RetryQueue responds to stats request" do
-    assert %{delivered: 0, dropped: 0} == RetryQueue.get_stats()
-  end
-
-  test "failed posts are retried" do
-    {:retry, _timeout} = RetryQueue.get_retry_params(@small_retry_count)
-
-    wait_task =
-      Task.async(fn ->
-        receive do
-          :complete -> :ok
-        end
-      end)
-
-    RetryQueue.enqueue({:ok, wait_task.pid}, MockActivityPub, @small_retry_count)
-    Task.await(wait_task)
-    assert %{delivered: 1, dropped: 0} == RetryQueue.get_stats()
-  end
-
-  test "posts that have been tried too many times are dropped" do
-    {:drop, _timeout} = RetryQueue.get_retry_params(@hopeless_retry_count)
-
-    RetryQueue.enqueue({:ok, nil}, MockActivityPub, @hopeless_retry_count)
-    assert %{delivered: 0, dropped: 1} == RetryQueue.get_stats()
-  end
-end
diff --git a/test/web/salmon/salmon_test.exs b/test/web/salmon/salmon_test.exs
index e86e76fe9..0186f3fef 100644
--- a/test/web/salmon/salmon_test.exs
+++ b/test/web/salmon/salmon_test.exs
@@ -96,6 +96,6 @@ defmodule Pleroma.Web.Salmon.SalmonTest do
 
     Salmon.publish(user, activity)
 
-    assert called(Publisher.enqueue_one(Salmon, %{recipient: mentioned_user}))
+    assert called(Publisher.enqueue_one(Salmon, %{recipient_id: mentioned_user.id}))
   end
 end
diff --git a/test/web/streamer/ping_test.exs b/test/web/streamer/ping_test.exs
new file mode 100644
index 000000000..3d52c00e4
--- /dev/null
+++ b/test/web/streamer/ping_test.exs
@@ -0,0 +1,36 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.PingTest do
+  use Pleroma.DataCase
+
+  import Pleroma.Factory
+  alias Pleroma.Web.Streamer
+
+  setup do
+    start_supervised({Streamer.supervisor(), [ping_interval: 30]})
+
+    :ok
+  end
+
+  describe "sockets" do
+    setup do
+      user = insert(:user)
+      {:ok, %{user: user}}
+    end
+
+    test "it sends pings", %{user: user} do
+      task =
+        Task.async(fn ->
+          assert_receive {:text, received_event}, 40
+          assert_receive {:text, received_event}, 40
+          assert_receive {:text, received_event}, 40
+        end)
+
+      Streamer.add_socket("public", %{transport_pid: task.pid, assigns: %{user: user}})
+
+      Task.await(task)
+    end
+  end
+end
diff --git a/test/web/streamer/state_test.exs b/test/web/streamer/state_test.exs
new file mode 100644
index 000000000..d1aeac541
--- /dev/null
+++ b/test/web/streamer/state_test.exs
@@ -0,0 +1,54 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.StateTest do
+  use Pleroma.DataCase
+
+  import Pleroma.Factory
+  alias Pleroma.Web.Streamer
+  alias Pleroma.Web.Streamer.StreamerSocket
+
+  @moduletag needs_streamer: true
+
+  describe "sockets" do
+    setup do
+      user = insert(:user)
+      user2 = insert(:user)
+      {:ok, %{user: user, user2: user2}}
+    end
+
+    test "it can add a socket", %{user: user} do
+      Streamer.add_socket("public", %{transport_pid: 1, assigns: %{user: user}})
+
+      assert(%{"public" => [%StreamerSocket{transport_pid: 1}]} = Streamer.get_sockets())
+    end
+
+    test "it can add multiple sockets per user", %{user: user} do
+      Streamer.add_socket("public", %{transport_pid: 1, assigns: %{user: user}})
+      Streamer.add_socket("public", %{transport_pid: 2, assigns: %{user: user}})
+
+      assert(
+        %{
+          "public" => [
+            %StreamerSocket{transport_pid: 2},
+            %StreamerSocket{transport_pid: 1}
+          ]
+        } = Streamer.get_sockets()
+      )
+    end
+
+    test "it will not add a duplicate socket", %{user: user} do
+      Streamer.add_socket("activity", %{transport_pid: 1, assigns: %{user: user}})
+      Streamer.add_socket("activity", %{transport_pid: 1, assigns: %{user: user}})
+
+      assert(
+        %{
+          "activity" => [
+            %StreamerSocket{transport_pid: 1}
+          ]
+        } = Streamer.get_sockets()
+      )
+    end
+  end
+end
diff --git a/test/web/streamer_test.exs b/test/web/streamer/streamer_test.exs
index 96fa7645f..88847e20f 100644
--- a/test/web/streamer_test.exs
+++ b/test/web/streamer/streamer_test.exs
@@ -5,24 +5,20 @@
 defmodule Pleroma.Web.StreamerTest do
   use Pleroma.DataCase
 
+  import Pleroma.Factory
+
   alias Pleroma.List
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI
   alias Pleroma.Web.Streamer
-  import Pleroma.Factory
+  alias Pleroma.Web.Streamer.StreamerSocket
+  alias Pleroma.Web.Streamer.Worker
 
+  @moduletag needs_streamer: true
   clear_config_all([:instance, :skip_thread_containment])
 
   describe "user streams" do
     setup do
-      GenServer.start(Streamer, %{}, name: Streamer)
-
-      on_exit(fn ->
-        if pid = Process.whereis(Streamer) do
-          Process.exit(pid, :kill)
-        end
-      end)
-
       user = insert(:user)
       notify = insert(:notification, user: user, activity: build(:note_activity))
       {:ok, %{user: user, notify: notify}}
@@ -125,11 +121,9 @@ defmodule Pleroma.Web.StreamerTest do
         assert_receive {:text, _}, 4_000
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user
-      }
+      user: user
     }
 
     {:ok, activity} = CommonAPI.post(other_user, %{"status" => "Test"})
@@ -138,7 +132,7 @@ defmodule Pleroma.Web.StreamerTest do
       "public" => [fake_socket]
     }
 
-    Streamer.push_to_socket(topics, "public", activity)
+    Worker.push_to_socket(topics, "public", activity)
 
     Task.await(task)
 
@@ -155,11 +149,9 @@ defmodule Pleroma.Web.StreamerTest do
         assert received_event == expected_event
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user
-      }
+      user: user
     }
 
     {:ok, activity} = CommonAPI.delete(activity.id, other_user)
@@ -168,7 +160,7 @@ defmodule Pleroma.Web.StreamerTest do
       "public" => [fake_socket]
     }
 
-    Streamer.push_to_socket(topics, "public", activity)
+    Worker.push_to_socket(topics, "public", activity)
 
     Task.await(task)
   end
@@ -189,9 +181,9 @@ defmodule Pleroma.Web.StreamerTest do
         )
 
       task = Task.async(fn -> refute_receive {:text, _}, 1_000 end)
-      fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
+      fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
       topics = %{"public" => [fake_socket]}
-      Streamer.push_to_socket(topics, "public", activity)
+      Worker.push_to_socket(topics, "public", activity)
 
       Task.await(task)
     end
@@ -211,9 +203,9 @@ defmodule Pleroma.Web.StreamerTest do
         )
 
       task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
-      fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
+      fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
       topics = %{"public" => [fake_socket]}
-      Streamer.push_to_socket(topics, "public", activity)
+      Worker.push_to_socket(topics, "public", activity)
 
       Task.await(task)
     end
@@ -233,9 +225,9 @@ defmodule Pleroma.Web.StreamerTest do
         )
 
       task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
-      fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
+      fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
       topics = %{"public" => [fake_socket]}
-      Streamer.push_to_socket(topics, "public", activity)
+      Worker.push_to_socket(topics, "public", activity)
 
       Task.await(task)
     end
@@ -251,11 +243,9 @@ defmodule Pleroma.Web.StreamerTest do
         refute_receive {:text, _}, 1_000
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user
-      }
+      user: user
     }
 
     {:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})
@@ -264,7 +254,7 @@ defmodule Pleroma.Web.StreamerTest do
       "public" => [fake_socket]
     }
 
-    Streamer.push_to_socket(topics, "public", activity)
+    Worker.push_to_socket(topics, "public", activity)
 
     Task.await(task)
   end
@@ -284,11 +274,9 @@ defmodule Pleroma.Web.StreamerTest do
         refute_receive {:text, _}, 1_000
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user_a
-      }
+      user: user_a
     }
 
     {:ok, activity} =
@@ -301,7 +289,7 @@ defmodule Pleroma.Web.StreamerTest do
       "list:#{list.id}" => [fake_socket]
     }
 
-    Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
+    Worker.handle_call({:stream, "list", activity}, self(), topics)
 
     Task.await(task)
   end
@@ -318,11 +306,9 @@ defmodule Pleroma.Web.StreamerTest do
         refute_receive {:text, _}, 1_000
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user_a
-      }
+      user: user_a
     }
 
     {:ok, activity} =
@@ -335,12 +321,12 @@ defmodule Pleroma.Web.StreamerTest do
       "list:#{list.id}" => [fake_socket]
     }
 
-    Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
+    Worker.handle_call({:stream, "list", activity}, self(), topics)
 
     Task.await(task)
   end
 
-  test "it send wanted private posts to list" do
+  test "it sends wanted private posts to list" do
     user_a = insert(:user)
     user_b = insert(:user)
 
@@ -354,11 +340,9 @@ defmodule Pleroma.Web.StreamerTest do
         assert_receive {:text, _}, 1_000
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user_a
-      }
+      user: user_a
     }
 
     {:ok, activity} =
@@ -367,11 +351,12 @@ defmodule Pleroma.Web.StreamerTest do
         "visibility" => "private"
       })
 
-    topics = %{
-      "list:#{list.id}" => [fake_socket]
-    }
+    Streamer.add_socket(
+      "list:#{list.id}",
+      fake_socket
+    )
 
-    Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
+    Worker.handle_call({:stream, "list", activity}, self(), %{})
 
     Task.await(task)
   end
@@ -387,11 +372,9 @@ defmodule Pleroma.Web.StreamerTest do
         refute_receive {:text, _}, 1_000
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user1
-      }
+      user: user1
     }
 
     {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
@@ -401,7 +384,7 @@ defmodule Pleroma.Web.StreamerTest do
       "public" => [fake_socket]
     }
 
-    Streamer.push_to_socket(topics, "public", announce_activity)
+    Worker.push_to_socket(topics, "public", announce_activity)
 
     Task.await(task)
   end
@@ -417,6 +400,8 @@ defmodule Pleroma.Web.StreamerTest do
 
     task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)
 
+    Process.sleep(4000)
+
     Streamer.add_socket(
       "user",
       %{transport_pid: task.pid, assigns: %{user: user2}}
@@ -428,14 +413,6 @@ defmodule Pleroma.Web.StreamerTest do
 
   describe "direct streams" do
     setup do
-      GenServer.start(Streamer, %{}, name: Streamer)
-
-      on_exit(fn ->
-        if pid = Process.whereis(Streamer) do
-          Process.exit(pid, :kill)
-        end
-      end)
-
       :ok
     end
 
@@ -480,6 +457,8 @@ defmodule Pleroma.Web.StreamerTest do
           refute_receive {:text, _}, 4_000
         end)
 
+      Process.sleep(1000)
+
       Streamer.add_socket(
         "direct",
         %{transport_pid: task.pid, assigns: %{user: user}}
@@ -521,6 +500,8 @@ defmodule Pleroma.Web.StreamerTest do
           assert last_status["id"] == to_string(create_activity.id)
         end)
 
+      Process.sleep(1000)
+
       Streamer.add_socket(
         "direct",
         %{transport_pid: task.pid, assigns: %{user: user}}
diff --git a/test/web/twitter_api/twitter_api_test.exs b/test/web/twitter_api/twitter_api_test.exs
index c5b18234e..3c0528776 100644
--- a/test/web/twitter_api/twitter_api_test.exs
+++ b/test/web/twitter_api/twitter_api_test.exs
@@ -5,6 +5,7 @@
 defmodule Pleroma.Web.TwitterAPI.TwitterAPITest do
   use Pleroma.DataCase
   alias Pleroma.Repo
+  alias Pleroma.Tests.ObanHelpers
   alias Pleroma.User
   alias Pleroma.UserInviteToken
   alias Pleroma.Web.MastodonAPI.AccountView
@@ -68,6 +69,7 @@ defmodule Pleroma.Web.TwitterAPI.TwitterAPITest do
     }
 
     {:ok, user} = TwitterAPI.register_user(data)
+    ObanHelpers.perform_all()
 
     assert user.info.confirmation_pending
 
diff --git a/test/web/twitter_api/util_controller_test.exs b/test/web/twitter_api/util_controller_test.exs
index cf8e69d2b..0a2a48fb7 100644
--- a/test/web/twitter_api/util_controller_test.exs
+++ b/test/web/twitter_api/util_controller_test.exs
@@ -4,10 +4,13 @@
 
 defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
   use Pleroma.Web.ConnCase
+  use Oban.Testing, repo: Pleroma.Repo
 
   alias Pleroma.Repo
+  alias Pleroma.Tests.ObanHelpers
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI
+  import ExUnit.CaptureLog
   import Pleroma.Factory
   import Mock
 
@@ -42,8 +45,7 @@ defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
         {File, [],
          read!: fn "follow_list.txt" ->
            "Account address,Show boosts\n#{user2.ap_id},true"
-         end},
-        {PleromaJobQueue, [:passthrough], []}
+         end}
       ]) do
         response =
           conn
@@ -51,15 +53,16 @@ defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
           |> post("/api/pleroma/follow_import", %{"list" => %Plug.Upload{path: "follow_list.txt"}})
           |> json_response(:ok)
 
-        assert called(
-                 PleromaJobQueue.enqueue(
-                   :background,
-                   User,
-                   [:follow_import, user1, [user2.ap_id]]
-                 )
-               )
-
         assert response == "job started"
+
+        assert ObanHelpers.member?(
+                 %{
+                   "op" => "follow_import",
+                   "follower_id" => user1.id,
+                   "followed_identifiers" => [user2.ap_id]
+                 },
+                 all_enqueued(worker: Pleroma.Workers.BackgroundWorker)
+               )
       end
     end
 
@@ -118,8 +121,7 @@ defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
       user3 = insert(:user)
 
       with_mocks([
-        {File, [], read!: fn "blocks_list.txt" -> "#{user2.ap_id} #{user3.ap_id}" end},
-        {PleromaJobQueue, [:passthrough], []}
+        {File, [], read!: fn "blocks_list.txt" -> "#{user2.ap_id} #{user3.ap_id}" end}
       ]) do
         response =
           conn
@@ -127,15 +129,16 @@ defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
           |> post("/api/pleroma/blocks_import", %{"list" => %Plug.Upload{path: "blocks_list.txt"}})
           |> json_response(:ok)
 
-        assert called(
-                 PleromaJobQueue.enqueue(
-                   :background,
-                   User,
-                   [:blocks_import, user1, [user2.ap_id, user3.ap_id]]
-                 )
-               )
-
         assert response == "job started"
+
+        assert ObanHelpers.member?(
+                 %{
+                   "op" => "blocks_import",
+                   "blocker_id" => user1.id,
+                   "blocked_identifiers" => [user2.ap_id, user3.ap_id]
+                 },
+                 all_enqueued(worker: Pleroma.Workers.BackgroundWorker)
+               )
       end
     end
   end
@@ -338,12 +341,14 @@ defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
     test "show follow page with error when user cannot fecth by `acct` link", %{conn: conn} do
       user = insert(:user)
 
-      response =
-        conn
-        |> assign(:user, user)
-        |> get("/ostatus_subscribe?acct=https://mastodon.social/users/not_found")
+      assert capture_log(fn ->
+               response =
+                 conn
+                 |> assign(:user, user)
+                 |> get("/ostatus_subscribe?acct=https://mastodon.social/users/not_found")
 
-      assert html_response(response, 200) =~ "Error fetching user"
+               assert html_response(response, 200) =~ "Error fetching user"
+             end) =~ "Object has been deleted"
     end
   end
 
@@ -557,6 +562,7 @@ defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
         |> json_response(:ok)
 
       assert response == %{"status" => "success"}
+      ObanHelpers.perform_all()
 
       user = User.get_cached_by_id(user.id)
 
@@ -662,4 +668,111 @@ defmodule Pleroma.Web.TwitterAPI.UtilControllerTest do
       assert called(Pleroma.Captcha.new())
     end
   end
+
+  defp with_credentials(conn, username, password) do
+    header_content = "Basic " <> Base.encode64("#{username}:#{password}")
+    put_req_header(conn, "authorization", header_content)
+  end
+
+  defp valid_user(_context) do
+    user = insert(:user)
+    [user: user]
+  end
+
+  describe "POST /api/pleroma/change_email" do
+    setup [:valid_user]
+
+    test "without credentials", %{conn: conn} do
+      conn = post(conn, "/api/pleroma/change_email")
+      assert json_response(conn, 403) == %{"error" => "Invalid credentials."}
+    end
+
+    test "with credentials and invalid password", %{conn: conn, user: current_user} do
+      conn =
+        conn
+        |> with_credentials(current_user.nickname, "test")
+        |> post("/api/pleroma/change_email", %{
+          "password" => "hi",
+          "email" => "test@test.com"
+        })
+
+      assert json_response(conn, 200) == %{"error" => "Invalid password."}
+    end
+
+    test "with credentials, valid password and invalid email", %{
+      conn: conn,
+      user: current_user
+    } do
+      conn =
+        conn
+        |> with_credentials(current_user.nickname, "test")
+        |> post("/api/pleroma/change_email", %{
+          "password" => "test",
+          "email" => "foobar"
+        })
+
+      assert json_response(conn, 200) == %{"error" => "Email has invalid format."}
+    end
+
+    test "with credentials, valid password and no email", %{
+      conn: conn,
+      user: current_user
+    } do
+      conn =
+        conn
+        |> with_credentials(current_user.nickname, "test")
+        |> post("/api/pleroma/change_email", %{
+          "password" => "test"
+        })
+
+      assert json_response(conn, 200) == %{"error" => "Email can't be blank."}
+    end
+
+    test "with credentials, valid password and blank email", %{
+      conn: conn,
+      user: current_user
+    } do
+      conn =
+        conn
+        |> with_credentials(current_user.nickname, "test")
+        |> post("/api/pleroma/change_email", %{
+          "password" => "test",
+          "email" => ""
+        })
+
+      assert json_response(conn, 200) == %{"error" => "Email can't be blank."}
+    end
+
+    test "with credentials, valid password and non unique email", %{
+      conn: conn,
+      user: current_user
+    } do
+      user = insert(:user)
+
+      conn =
+        conn
+        |> with_credentials(current_user.nickname, "test")
+        |> post("/api/pleroma/change_email", %{
+          "password" => "test",
+          "email" => user.email
+        })
+
+      assert json_response(conn, 200) == %{"error" => "Email has already been taken."}
+    end
+
+    test "with credentials, valid password and valid email", %{
+      conn: conn,
+      user: current_user
+    } do
+      conn =
+        conn
+        |> with_credentials(current_user.nickname, "test")
+        |> post("/api/pleroma/change_email", %{
+          "password" => "test",
+          "email" => "cofe@foobar.com"
+        })
+
+      assert json_response(conn, 200) == %{"status" => "success"}
+    end
+  end
 end
diff --git a/test/web/web_finger/web_finger_controller_test.exs b/test/web/web_finger/web_finger_controller_test.exs
index e23086b2a..bd3ccaaf7 100644
--- a/test/web/web_finger/web_finger_controller_test.exs
+++ b/test/web/web_finger/web_finger_controller_test.exs
@@ -5,6 +5,7 @@
 defmodule Pleroma.Web.WebFinger.WebFingerControllerTest do
   use Pleroma.Web.ConnCase
 
+  import ExUnit.CaptureLog
   import Pleroma.Factory
   import Tesla.Mock
 
@@ -75,11 +76,13 @@ defmodule Pleroma.Web.WebFinger.WebFingerControllerTest do
   test "Sends a 404 when invalid format" do
     user = insert(:user)
 
-    assert_raise Phoenix.NotAcceptableError, fn ->
-      build_conn()
-      |> put_req_header("accept", "text/html")
-      |> get("/.well-known/webfinger?resource=acct:#{user.nickname}@localhost")
-    end
+    assert capture_log(fn ->
+             assert_raise Phoenix.NotAcceptableError, fn ->
+               build_conn()
+               |> put_req_header("accept", "text/html")
+               |> get("/.well-known/webfinger?resource=acct:#{user.nickname}@localhost")
+             end
+           end) =~ "no supported media type in accept header"
   end
 
   test "Sends a 400 when resource param is missing" do
diff --git a/test/web/websub/websub_test.exs b/test/web/websub/websub_test.exs
index 74386d7db..929acf5a2 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.Tests.ObanHelpers
   alias Pleroma.Web.Router.Helpers
   alias Pleroma.Web.Websub
   alias Pleroma.Web.Websub.WebsubClientSubscription
   alias Pleroma.Web.Websub.WebsubServerSubscription
+  alias Pleroma.Workers.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)