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