Merge branch 'streamer-get-topic' into 'develop'

Expand and authorize streams in Streamer directly

See merge request pleroma/pleroma!2519

3 files changed, 105 insertions, 35 deletions

diff --git a/test/integration/mastodon_websocket_test.exs b/test/integration/mastodon_websocket_test.exs
index 109c7b4cb..f61150cd2 100644
--- a/test/integration/mastodon_websocket_test.exs
+++ b/test/integration/mastodon_websocket_test.exs
@@ -32,7 +32,7 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
 
   test "refuses invalid requests" do
     capture_log(fn ->
-      assert {:error, {400, _}} = start_socket()
+      assert {:error, {404, _}} = start_socket()
       assert {:error, {404, _}} = start_socket("?stream=ncjdk")
       Process.sleep(30)
     end)
@@ -40,8 +40,8 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
 
   test "requires authentication and a valid token for protected streams" do
     capture_log(fn ->
-      assert {:error, {403, _}} = start_socket("?stream=user&access_token=aaaaaaaaaaaa")
-      assert {:error, {403, _}} = start_socket("?stream=user")
+      assert {:error, {401, _}} = start_socket("?stream=user&access_token=aaaaaaaaaaaa")
+      assert {:error, {401, _}} = start_socket("?stream=user")
       Process.sleep(30)
     end)
   end
@@ -100,7 +100,7 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
       assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")
 
       assert capture_log(fn ->
-               assert {:error, {403, "Forbidden"}} = start_socket("?stream=user")
+               assert {:error, {401, _}} = start_socket("?stream=user")
                Process.sleep(30)
              end) =~ ":badarg"
     end
@@ -109,7 +109,7 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
       assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")
 
       assert capture_log(fn ->
-               assert {:error, {403, "Forbidden"}} = start_socket("?stream=user:notification")
+               assert {:error, {401, _}} = start_socket("?stream=user:notification")
                Process.sleep(30)
              end) =~ ":badarg"
     end
@@ -118,7 +118,7 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
       assert {:ok, _} = start_socket("?stream=user", [{"Sec-WebSocket-Protocol", token.token}])
 
       assert capture_log(fn ->
-               assert {:error, {403, "Forbidden"}} =
+               assert {:error, {401, _}} =
                         start_socket("?stream=user", [{"Sec-WebSocket-Protocol", "I am a friend"}])
 
                Process.sleep(30)
diff --git a/test/notification_test.exs b/test/notification_test.exs
index 24e5f0c73..4dfbc1019 100644
--- a/test/notification_test.exs
+++ b/test/notification_test.exs
@@ -170,13 +170,13 @@ defmodule Pleroma.NotificationTest do
 
       task =
         Task.async(fn ->
-          Streamer.add_socket("user", user)
+          Streamer.get_topic_and_add_socket("user", user)
           assert_receive {:render_with_user, _, _, _}, 4_000
         end)
 
       task_user_notification =
         Task.async(fn ->
-          Streamer.add_socket("user:notification", user)
+          Streamer.get_topic_and_add_socket("user:notification", user)
           assert_receive {:render_with_user, _, _, _}, 4_000
         end)
 
diff --git a/test/web/streamer/streamer_test.exs b/test/web/streamer/streamer_test.exs
index ee530f4e9..db07c5df5 100644
--- a/test/web/streamer/streamer_test.exs
+++ b/test/web/streamer/streamer_test.exs
@@ -17,6 +17,76 @@ defmodule Pleroma.Web.StreamerTest do
 
   setup do: clear_config([:instance, :skip_thread_containment])
 
+  describe "get_topic without an user" do
+    test "allows public" do
+      assert {:ok, "public"} = Streamer.get_topic("public", nil)
+      assert {:ok, "public:local"} = Streamer.get_topic("public:local", nil)
+      assert {:ok, "public:media"} = Streamer.get_topic("public:media", nil)
+      assert {:ok, "public:local:media"} = Streamer.get_topic("public:local:media", nil)
+    end
+
+    test "allows hashtag streams" do
+      assert {:ok, "hashtag:cofe"} = Streamer.get_topic("hashtag", nil, %{"tag" => "cofe"})
+    end
+
+    test "disallows user streams" do
+      assert {:error, _} = Streamer.get_topic("user", nil)
+      assert {:error, _} = Streamer.get_topic("user:notification", nil)
+      assert {:error, _} = Streamer.get_topic("direct", nil)
+    end
+
+    test "disallows list streams" do
+      assert {:error, _} = Streamer.get_topic("list", nil, %{"list" => 42})
+    end
+  end
+
+  describe "get_topic with an user" do
+    setup do
+      user = insert(:user)
+      {:ok, %{user: user}}
+    end
+
+    test "allows public streams", %{user: user} do
+      assert {:ok, "public"} = Streamer.get_topic("public", user)
+      assert {:ok, "public:local"} = Streamer.get_topic("public:local", user)
+      assert {:ok, "public:media"} = Streamer.get_topic("public:media", user)
+      assert {:ok, "public:local:media"} = Streamer.get_topic("public:local:media", user)
+    end
+
+    test "allows user streams", %{user: user} do
+      expected_user_topic = "user:#{user.id}"
+      expected_notif_topic = "user:notification:#{user.id}"
+      expected_direct_topic = "direct:#{user.id}"
+      assert {:ok, ^expected_user_topic} = Streamer.get_topic("user", user)
+      assert {:ok, ^expected_notif_topic} = Streamer.get_topic("user:notification", user)
+      assert {:ok, ^expected_direct_topic} = Streamer.get_topic("direct", user)
+    end
+
+    test "allows hashtag streams", %{user: user} do
+      assert {:ok, "hashtag:cofe"} = Streamer.get_topic("hashtag", user, %{"tag" => "cofe"})
+    end
+
+    test "disallows registering to an user stream", %{user: user} do
+      another_user = insert(:user)
+      assert {:error, _} = Streamer.get_topic("user:#{another_user.id}", user)
+      assert {:error, _} = Streamer.get_topic("user:notification:#{another_user.id}", user)
+      assert {:error, _} = Streamer.get_topic("direct:#{another_user.id}", user)
+    end
+
+    test "allows list stream that are owned by the user", %{user: user} do
+      {:ok, list} = List.create("Test", user)
+      assert {:error, _} = Streamer.get_topic("list:#{list.id}", user)
+      assert {:ok, _} = Streamer.get_topic("list", user, %{"list" => list.id})
+    end
+
+    test "disallows list stream that are not owned by the user", %{user: user} do
+      another_user = insert(:user)
+      {:ok, list} = List.create("Test", another_user)
+      assert {:error, _} = Streamer.get_topic("list:#{list.id}", user)
+      assert {:error, _} = Streamer.get_topic("list", user, %{"list" => list.id})
+    end
+  end
+
   describe "user streams" do
     setup do
       user = insert(:user)
@@ -25,14 +95,14 @@ defmodule Pleroma.Web.StreamerTest do
     end
 
     test "it streams the user's post in the 'user' stream", %{user: user} do
-      Streamer.add_socket("user", user)
+      Streamer.get_topic_and_add_socket("user", user)
       {:ok, activity} = CommonAPI.post(user, %{"status" => "hey"})
       assert_receive {:render_with_user, _, _, ^activity}
       refute Streamer.filtered_by_user?(user, activity)
     end
 
     test "it streams boosts of the user in the 'user' stream", %{user: user} do
-      Streamer.add_socket("user", user)
+      Streamer.get_topic_and_add_socket("user", user)
 
       other_user = insert(:user)
       {:ok, activity} = CommonAPI.post(other_user, %{"status" => "hey"})
@@ -43,14 +113,14 @@ defmodule Pleroma.Web.StreamerTest do
     end
 
     test "it sends notify to in the 'user' stream", %{user: user, notify: notify} do
-      Streamer.add_socket("user", user)
+      Streamer.get_topic_and_add_socket("user", user)
       Streamer.stream("user", notify)
       assert_receive {:render_with_user, _, _, ^notify}
       refute Streamer.filtered_by_user?(user, notify)
     end
 
     test "it sends notify to in the 'user:notification' stream", %{user: user, notify: notify} do
-      Streamer.add_socket("user:notification", user)
+      Streamer.get_topic_and_add_socket("user:notification", user)
       Streamer.stream("user:notification", notify)
       assert_receive {:render_with_user, _, _, ^notify}
       refute Streamer.filtered_by_user?(user, notify)
@@ -62,7 +132,7 @@ defmodule Pleroma.Web.StreamerTest do
       blocked = insert(:user)
       {:ok, _user_relationship} = User.block(user, blocked)
 
-      Streamer.add_socket("user:notification", user)
+      Streamer.get_topic_and_add_socket("user:notification", user)
 
       {:ok, activity} = CommonAPI.post(user, %{"status" => ":("})
       {:ok, _} = CommonAPI.favorite(blocked, activity.id)
@@ -78,7 +148,7 @@ defmodule Pleroma.Web.StreamerTest do
       {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
       {:ok, _} = CommonAPI.add_mute(user, activity)
 
-      Streamer.add_socket("user:notification", user)
+      Streamer.get_topic_and_add_socket("user:notification", user)
 
       {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
 
@@ -92,7 +162,7 @@ defmodule Pleroma.Web.StreamerTest do
       user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
 
       {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
-      Streamer.add_socket("user:notification", user)
+      Streamer.get_topic_and_add_socket("user:notification", user)
       {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
 
       assert_receive {:render_with_user, _, "notification.json", notif}
@@ -107,7 +177,7 @@ defmodule Pleroma.Web.StreamerTest do
 
       {:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
       {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
-      Streamer.add_socket("user:notification", user)
+      Streamer.get_topic_and_add_socket("user:notification", user)
       {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
 
       refute_receive _
@@ -130,7 +200,7 @@ defmodule Pleroma.Web.StreamerTest do
           %Tesla.Env{status: 200, body: body}
       end)
 
-      Streamer.add_socket("user:notification", user)
+      Streamer.get_topic_and_add_socket("user:notification", user)
       {:ok, _follower, _followed, follow_activity} = CommonAPI.follow(user2, user)
 
       assert_receive {:render_with_user, _, "notification.json", notif}
@@ -143,7 +213,7 @@ defmodule Pleroma.Web.StreamerTest do
     user = insert(:user)
     other_user = insert(:user)
 
-    Streamer.add_socket("public", other_user)
+    Streamer.get_topic_and_add_socket("public", other_user)
 
     {:ok, activity} = CommonAPI.post(user, %{"status" => "Test"})
     assert_receive {:render_with_user, _, _, ^activity}
@@ -155,7 +225,7 @@ defmodule Pleroma.Web.StreamerTest do
     other_user = insert(:user)
     {:ok, activity} = CommonAPI.post(other_user, %{"status" => "Test"})
 
-    Streamer.add_socket("public", user)
+    Streamer.get_topic_and_add_socket("public", user)
 
     {:ok, _} = CommonAPI.delete(activity.id, other_user)
     activity_id = activity.id
@@ -166,7 +236,7 @@ defmodule Pleroma.Web.StreamerTest do
   test "it sends to public unauthenticated" do
     user = insert(:user)
 
-    Streamer.add_socket("public", nil)
+    Streamer.get_topic_and_add_socket("public", nil)
 
     {:ok, activity} = CommonAPI.post(user, %{"status" => "Test"})
     activity_id = activity.id
@@ -195,7 +265,7 @@ defmodule Pleroma.Web.StreamerTest do
             )
         )
 
-      Streamer.add_socket("public", user)
+      Streamer.get_topic_and_add_socket("public", user)
       Streamer.stream("public", activity)
       assert_receive {:render_with_user, _, _, ^activity}
       assert Streamer.filtered_by_user?(user, activity)
@@ -216,7 +286,7 @@ defmodule Pleroma.Web.StreamerTest do
             )
         )
 
-      Streamer.add_socket("public", user)
+      Streamer.get_topic_and_add_socket("public", user)
       Streamer.stream("public", activity)
 
       assert_receive {:render_with_user, _, _, ^activity}
@@ -238,7 +308,7 @@ defmodule Pleroma.Web.StreamerTest do
             )
         )
 
-      Streamer.add_socket("public", user)
+      Streamer.get_topic_and_add_socket("public", user)
       Streamer.stream("public", activity)
 
       assert_receive {:render_with_user, _, _, ^activity}
@@ -252,7 +322,7 @@ defmodule Pleroma.Web.StreamerTest do
       blocked_user = insert(:user)
       {:ok, _user_relationship} = User.block(user, blocked_user)
 
-      Streamer.add_socket("public", user)
+      Streamer.get_topic_and_add_socket("public", user)
       {:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})
       assert_receive {:render_with_user, _, _, ^activity}
       assert Streamer.filtered_by_user?(user, activity)
@@ -263,7 +333,7 @@ defmodule Pleroma.Web.StreamerTest do
       blockee = insert(:user)
       friend = insert(:user)
 
-      Streamer.add_socket("public", blocker)
+      Streamer.get_topic_and_add_socket("public", blocker)
 
       {:ok, _user_relationship} = User.block(blocker, blockee)
 
@@ -295,7 +365,7 @@ defmodule Pleroma.Web.StreamerTest do
       {:ok, list} = List.create("Test", user_a)
       {:ok, list} = List.follow(list, user_b)
 
-      Streamer.add_socket("list:#{list.id}", user_a)
+      Streamer.get_topic_and_add_socket("list", user_a, %{"list" => list.id})
 
       {:ok, _activity} =
         CommonAPI.post(user_b, %{
@@ -313,7 +383,7 @@ defmodule Pleroma.Web.StreamerTest do
       {:ok, list} = List.create("Test", user_a)
       {:ok, list} = List.follow(list, user_b)
 
-      Streamer.add_socket("list:#{list.id}", user_a)
+      Streamer.get_topic_and_add_socket("list", user_a, %{"list" => list.id})
 
       {:ok, _activity} =
         CommonAPI.post(user_b, %{
@@ -333,7 +403,7 @@ defmodule Pleroma.Web.StreamerTest do
       {:ok, list} = List.create("Test", user_a)
       {:ok, list} = List.follow(list, user_b)
 
-      Streamer.add_socket("list:#{list.id}", user_a)
+      Streamer.get_topic_and_add_socket("list", user_a, %{"list" => list.id})
 
       {:ok, activity} =
         CommonAPI.post(user_b, %{
@@ -356,7 +426,7 @@ defmodule Pleroma.Web.StreamerTest do
 
       {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
 
-      Streamer.add_socket("user", user1)
+      Streamer.get_topic_and_add_socket("user", user1)
       {:ok, announce_activity, _} = CommonAPI.repeat(create_activity.id, user2)
       assert_receive {:render_with_user, _, _, ^announce_activity}
       assert Streamer.filtered_by_user?(user1, announce_activity)
@@ -369,7 +439,7 @@ defmodule Pleroma.Web.StreamerTest do
       CommonAPI.hide_reblogs(user1, user2)
 
       {:ok, create_activity} = CommonAPI.post(user1, %{"status" => "I'm kawen"})
-      Streamer.add_socket("user", user1)
+      Streamer.get_topic_and_add_socket("user", user1)
       {:ok, _favorite_activity, _} = CommonAPI.repeat(create_activity.id, user2)
 
       assert_receive {:render_with_user, _, "notification.json", notif}
@@ -383,7 +453,7 @@ defmodule Pleroma.Web.StreamerTest do
       CommonAPI.hide_reblogs(user1, user2)
 
       {:ok, create_activity} = CommonAPI.post(user1, %{"status" => "I'm kawen"})
-      Streamer.add_socket("user", user1)
+      Streamer.get_topic_and_add_socket("user", user1)
       {:ok, _favorite_activity} = CommonAPI.favorite(user2, create_activity.id)
 
       assert_receive {:render_with_user, _, "notification.json", notif}
@@ -394,7 +464,7 @@ defmodule Pleroma.Web.StreamerTest do
   test "it filters posts from muted threads" do
     user = insert(:user)
     user2 = insert(:user)
-    Streamer.add_socket("user", user2)
+    Streamer.get_topic_and_add_socket("user", user2)
     {:ok, user2, user, _activity} = CommonAPI.follow(user2, user)
     {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
     {:ok, _} = CommonAPI.add_mute(user2, activity)
@@ -411,7 +481,7 @@ defmodule Pleroma.Web.StreamerTest do
       user = insert(:user)
       another_user = insert(:user)
 
-      Streamer.add_socket("direct", user)
+      Streamer.get_topic_and_add_socket("direct", user)
 
       {:ok, _create_activity} =
         CommonAPI.post(another_user, %{
@@ -433,7 +503,7 @@ defmodule Pleroma.Web.StreamerTest do
       user = insert(:user)
       another_user = insert(:user)
 
-      Streamer.add_socket("direct", user)
+      Streamer.get_topic_and_add_socket("direct", user)
 
       {:ok, create_activity} =
         CommonAPI.post(another_user, %{
@@ -459,7 +529,7 @@ defmodule Pleroma.Web.StreamerTest do
     test "it sends conversation update to the 'direct' stream when a message is deleted" do
       user = insert(:user)
       another_user = insert(:user)
-      Streamer.add_socket("direct", user)
+      Streamer.get_topic_and_add_socket("direct", user)
 
       {:ok, create_activity} =
         CommonAPI.post(another_user, %{