git.squeep.com Git - akkoma/blob - test/web/streamer_test.exs

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2018 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Web.StreamerTest do
   6   use Pleroma.DataCase
   7
   8   alias Pleroma.List
   9   alias Pleroma.User
  10   alias Pleroma.Web.CommonAPI
  11   alias Pleroma.Web.Streamer
  12   import Pleroma.Factory
  13
  14   setup do
  15     skip_thread_containment = Pleroma.Config.get([:instance, :skip_thread_containment])
  16
  17     on_exit(fn ->
  18       Pleroma.Config.put([:instance, :skip_thread_containment], skip_thread_containment)
  19     end)
  20
  21     :ok
  22   end
  23
  24   describe "user streams" do
  25     setup do
  26       GenServer.start(Streamer, %{}, name: Streamer)
  27
  28       on_exit(fn ->
  29         if pid = Process.whereis(Streamer) do
  30           Process.exit(pid, :kill)
  31         end
  32       end)
  33
  34       user = insert(:user)
  35       notify = insert(:notification, user: user, activity: build(:note_activity))
  36       {:ok, %{user: user, notify: notify}}
  37     end
  38
  39     test "it sends notify to in the 'user' stream", %{user: user, notify: notify} do
  40       task =
  41         Task.async(fn ->
  42           assert_receive {:text, _}, 4_000
  43         end)
  44
  45       Streamer.add_socket(
  46         "user",
  47         %{transport_pid: task.pid, assigns: %{user: user}}
  48       )
  49
  50       Streamer.stream("user", notify)
  51       Task.await(task)
  52     end
  53
  54     test "it sends notify to in the 'user:notification' stream", %{user: user, notify: notify} do
  55       task =
  56         Task.async(fn ->
  57           assert_receive {:text, _}, 4_000
  58         end)
  59
  60       Streamer.add_socket(
  61         "user:notification",
  62         %{transport_pid: task.pid, assigns: %{user: user}}
  63       )
  64
  65       Streamer.stream("user:notification", notify)
  66       Task.await(task)
  67     end
  68   end
  69
  70   test "it sends to public" do
  71     user = insert(:user)
  72     other_user = insert(:user)
  73
  74     task =
  75       Task.async(fn ->
  76         assert_receive {:text, _}, 4_000
  77       end)
  78
  79     fake_socket = %{
  80       transport_pid: task.pid,
  81       assigns: %{
  82         user: user
  83       }
  84     }
  85
  86     {:ok, activity} = CommonAPI.post(other_user, %{"status" => "Test"})
  87
  88     topics = %{
  89       "public" => [fake_socket]
  90     }
  91
  92     Streamer.push_to_socket(topics, "public", activity)
  93
  94     Task.await(task)
  95
  96     task =
  97       Task.async(fn ->
  98         expected_event =
  99           %{
 100             "event" => "delete",
 101             "payload" => activity.id
 102           }
 103           |> Jason.encode!()
 104
 105         assert_receive {:text, received_event}, 4_000
 106         assert received_event == expected_event
 107       end)
 108
 109     fake_socket = %{
 110       transport_pid: task.pid,
 111       assigns: %{
 112         user: user
 113       }
 114     }
 115
 116     {:ok, activity} = CommonAPI.delete(activity.id, other_user)
 117
 118     topics = %{
 119       "public" => [fake_socket]
 120     }
 121
 122     Streamer.push_to_socket(topics, "public", activity)
 123
 124     Task.await(task)
 125   end
 126
 127   describe "thread_containment" do
 128     test "it doesn't send to user if recipients invalid and thread containment is enabled" do
 129       Pleroma.Config.put([:instance, :skip_thread_containment], false)
 130       author = insert(:user)
 131       user = insert(:user, following: [author.ap_id])
 132
 133       activity =
 134         insert(:note_activity,
 135           note:
 136             insert(:note,
 137               user: author,
 138               data: %{"to" => ["TEST-FFF"]}
 139             )
 140         )
 141
 142       task = Task.async(fn -> refute_receive {:text, _}, 1_000 end)
 143       fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
 144       topics = %{"public" => [fake_socket]}
 145       Streamer.push_to_socket(topics, "public", activity)
 146
 147       Task.await(task)
 148     end
 149
 150     test "it sends message if recipients invalid and thread containment is disabled" do
 151       Pleroma.Config.put([:instance, :skip_thread_containment], true)
 152       author = insert(:user)
 153       user = insert(:user, following: [author.ap_id])
 154
 155       activity =
 156         insert(:note_activity,
 157           note:
 158             insert(:note,
 159               user: author,
 160               data: %{"to" => ["TEST-FFF"]}
 161             )
 162         )
 163
 164       task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
 165       fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
 166       topics = %{"public" => [fake_socket]}
 167       Streamer.push_to_socket(topics, "public", activity)
 168
 169       Task.await(task)
 170     end
 171
 172     test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
 173       Pleroma.Config.put([:instance, :skip_thread_containment], false)
 174       author = insert(:user)
 175       user = insert(:user, following: [author.ap_id], info: %{skip_thread_containment: true})
 176
 177       activity =
 178         insert(:note_activity,
 179           note:
 180             insert(:note,
 181               user: author,
 182               data: %{"to" => ["TEST-FFF"]}
 183             )
 184         )
 185
 186       task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
 187       fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
 188       topics = %{"public" => [fake_socket]}
 189       Streamer.push_to_socket(topics, "public", activity)
 190
 191       Task.await(task)
 192     end
 193   end
 194
 195   test "it doesn't send to blocked users" do
 196     user = insert(:user)
 197     blocked_user = insert(:user)
 198     {:ok, user} = User.block(user, blocked_user)
 199
 200     task =
 201       Task.async(fn ->
 202         refute_receive {:text, _}, 1_000
 203       end)
 204
 205     fake_socket = %{
 206       transport_pid: task.pid,
 207       assigns: %{
 208         user: user
 209       }
 210     }
 211
 212     {:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})
 213
 214     topics = %{
 215       "public" => [fake_socket]
 216     }
 217
 218     Streamer.push_to_socket(topics, "public", activity)
 219
 220     Task.await(task)
 221   end
 222
 223   test "it doesn't send unwanted DMs to list" do
 224     user_a = insert(:user)
 225     user_b = insert(:user)
 226     user_c = insert(:user)
 227
 228     {:ok, user_a} = User.follow(user_a, user_b)
 229
 230     {:ok, list} = List.create("Test", user_a)
 231     {:ok, list} = List.follow(list, user_b)
 232
 233     task =
 234       Task.async(fn ->
 235         refute_receive {:text, _}, 1_000
 236       end)
 237
 238     fake_socket = %{
 239       transport_pid: task.pid,
 240       assigns: %{
 241         user: user_a
 242       }
 243     }
 244
 245     {:ok, activity} =
 246       CommonAPI.post(user_b, %{
 247         "status" => "@#{user_c.nickname} Test",
 248         "visibility" => "direct"
 249       })
 250
 251     topics = %{
 252       "list:#{list.id}" => [fake_socket]
 253     }
 254
 255     Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
 256
 257     Task.await(task)
 258   end
 259
 260   test "it doesn't send unwanted private posts to list" do
 261     user_a = insert(:user)
 262     user_b = insert(:user)
 263
 264     {:ok, list} = List.create("Test", user_a)
 265     {:ok, list} = List.follow(list, user_b)
 266
 267     task =
 268       Task.async(fn ->
 269         refute_receive {:text, _}, 1_000
 270       end)
 271
 272     fake_socket = %{
 273       transport_pid: task.pid,
 274       assigns: %{
 275         user: user_a
 276       }
 277     }
 278
 279     {:ok, activity} =
 280       CommonAPI.post(user_b, %{
 281         "status" => "Test",
 282         "visibility" => "private"
 283       })
 284
 285     topics = %{
 286       "list:#{list.id}" => [fake_socket]
 287     }
 288
 289     Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
 290
 291     Task.await(task)
 292   end
 293
 294   test "it send wanted private posts to list" do
 295     user_a = insert(:user)
 296     user_b = insert(:user)
 297
 298     {:ok, user_a} = User.follow(user_a, user_b)
 299
 300     {:ok, list} = List.create("Test", user_a)
 301     {:ok, list} = List.follow(list, user_b)
 302
 303     task =
 304       Task.async(fn ->
 305         assert_receive {:text, _}, 1_000
 306       end)
 307
 308     fake_socket = %{
 309       transport_pid: task.pid,
 310       assigns: %{
 311         user: user_a
 312       }
 313     }
 314
 315     {:ok, activity} =
 316       CommonAPI.post(user_b, %{
 317         "status" => "Test",
 318         "visibility" => "private"
 319       })
 320
 321     topics = %{
 322       "list:#{list.id}" => [fake_socket]
 323     }
 324
 325     Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
 326
 327     Task.await(task)
 328   end
 329
 330   test "it doesn't send muted reblogs" do
 331     user1 = insert(:user)
 332     user2 = insert(:user)
 333     user3 = insert(:user)
 334     CommonAPI.hide_reblogs(user1, user2)
 335
 336     task =
 337       Task.async(fn ->
 338         refute_receive {:text, _}, 1_000
 339       end)
 340
 341     fake_socket = %{
 342       transport_pid: task.pid,
 343       assigns: %{
 344         user: user1
 345       }
 346     }
 347
 348     {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
 349     {:ok, announce_activity, _} = CommonAPI.repeat(create_activity.id, user2)
 350
 351     topics = %{
 352       "public" => [fake_socket]
 353     }
 354
 355     Streamer.push_to_socket(topics, "public", announce_activity)
 356
 357     Task.await(task)
 358   end
 359
 360   describe "direct streams" do
 361     setup do
 362       GenServer.start(Streamer, %{}, name: Streamer)
 363
 364       on_exit(fn ->
 365         if pid = Process.whereis(Streamer) do
 366           Process.exit(pid, :kill)
 367         end
 368       end)
 369
 370       :ok
 371     end
 372
 373     test "it sends conversation update to the 'direct' stream", %{} do
 374       user = insert(:user)
 375       another_user = insert(:user)
 376
 377       task =
 378         Task.async(fn ->
 379           assert_receive {:text, _received_event}, 4_000
 380         end)
 381
 382       Streamer.add_socket(
 383         "direct",
 384         %{transport_pid: task.pid, assigns: %{user: user}}
 385       )
 386
 387       {:ok, _create_activity} =
 388         CommonAPI.post(another_user, %{
 389           "status" => "hey @#{user.nickname}",
 390           "visibility" => "direct"
 391         })
 392
 393       Task.await(task)
 394     end
 395
 396     test "it doesn't send conversation update to the 'direct' streamj when the last message in the conversation is deleted" do
 397       user = insert(:user)
 398       another_user = insert(:user)
 399
 400       {:ok, create_activity} =
 401         CommonAPI.post(another_user, %{
 402           "status" => "hi @#{user.nickname}",
 403           "visibility" => "direct"
 404         })
 405
 406       task =
 407         Task.async(fn ->
 408           assert_receive {:text, received_event}, 4_000
 409           assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
 410
 411           refute_receive {:text, _}, 4_000
 412         end)
 413
 414       Streamer.add_socket(
 415         "direct",
 416         %{transport_pid: task.pid, assigns: %{user: user}}
 417       )
 418
 419       {:ok, _} = CommonAPI.delete(create_activity.id, another_user)
 420
 421       Task.await(task)
 422     end
 423
 424     test "it sends conversation update to the 'direct' stream when a message is deleted" do
 425       user = insert(:user)
 426       another_user = insert(:user)
 427
 428       {:ok, create_activity} =
 429         CommonAPI.post(another_user, %{
 430           "status" => "hi @#{user.nickname}",
 431           "visibility" => "direct"
 432         })
 433
 434       {:ok, create_activity2} =
 435         CommonAPI.post(another_user, %{
 436           "status" => "hi @#{user.nickname}",
 437           "in_reply_to_status_id" => create_activity.id,
 438           "visibility" => "direct"
 439         })
 440
 441       task =
 442         Task.async(fn ->
 443           assert_receive {:text, received_event}, 4_000
 444           assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
 445
 446           assert_receive {:text, received_event}, 4_000
 447
 448           assert %{"event" => "conversation", "payload" => received_payload} =
 449                    Jason.decode!(received_event)
 450
 451           assert %{"last_status" => last_status} = Jason.decode!(received_payload)
 452           assert last_status["id"] == to_string(create_activity.id)
 453         end)
 454
 455       Streamer.add_socket(
 456         "direct",
 457         %{transport_pid: task.pid, assigns: %{user: user}}
 458       )
 459
 460       {:ok, _} = CommonAPI.delete(create_activity2.id, another_user)
 461
 462       Task.await(task)
 463     end
 464   end
 465 end