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

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2019 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   import Pleroma.Factory
   9
  10   alias Pleroma.Conversation.Participation
  11   alias Pleroma.List
  12   alias Pleroma.User
  13   alias Pleroma.Web.CommonAPI
  14   alias Pleroma.Web.Streamer
  15   alias Pleroma.Web.Streamer.StreamerSocket
  16   alias Pleroma.Web.Streamer.Worker
  17
  18   @moduletag needs_streamer: true, capture_log: true
  19
  20   @streamer_timeout 150
  21   @streamer_start_wait 10
  22
  23   clear_config_all([:instance, :skip_thread_containment])
  24
  25   describe "user streams" do
  26     setup do
  27       user = insert(:user)
  28       notify = insert(:notification, user: user, activity: build(:note_activity))
  29       {:ok, %{user: user, notify: notify}}
  30     end
  31
  32     test "it sends notify to in the 'user' stream", %{user: user, notify: notify} do
  33       task =
  34         Task.async(fn ->
  35           assert_receive {:text, _}, @streamer_timeout
  36         end)
  37
  38       Streamer.add_socket(
  39         "user",
  40         %{transport_pid: task.pid, assigns: %{user: user}}
  41       )
  42
  43       Streamer.stream("user", notify)
  44       Task.await(task)
  45     end
  46
  47     test "it sends notify to in the 'user:notification' stream", %{user: user, notify: notify} do
  48       task =
  49         Task.async(fn ->
  50           assert_receive {:text, _}, @streamer_timeout
  51         end)
  52
  53       Streamer.add_socket(
  54         "user:notification",
  55         %{transport_pid: task.pid, assigns: %{user: user}}
  56       )
  57
  58       Streamer.stream("user:notification", notify)
  59       Task.await(task)
  60     end
  61
  62     test "it doesn't send notify to the 'user:notification' stream when a user is blocked", %{
  63       user: user
  64     } do
  65       blocked = insert(:user)
  66       {:ok, _user_relationship} = User.block(user, blocked)
  67
  68       task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
  69
  70       Streamer.add_socket(
  71         "user:notification",
  72         %{transport_pid: task.pid, assigns: %{user: user}}
  73       )
  74
  75       {:ok, activity} = CommonAPI.post(user, %{"status" => ":("})
  76       {:ok, notif, _} = CommonAPI.favorite(activity.id, blocked)
  77
  78       Streamer.stream("user:notification", notif)
  79       Task.await(task)
  80     end
  81
  82     test "it doesn't send notify to the 'user:notification' stream when a thread is muted", %{
  83       user: user
  84     } do
  85       user2 = insert(:user)
  86       task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
  87
  88       Streamer.add_socket(
  89         "user:notification",
  90         %{transport_pid: task.pid, assigns: %{user: user}}
  91       )
  92
  93       {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
  94       {:ok, activity} = CommonAPI.add_mute(user, activity)
  95       {:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
  96       Streamer.stream("user:notification", notif)
  97       Task.await(task)
  98     end
  99
 100     test "it doesn't send notify to the 'user:notification' stream' when a domain is blocked", %{
 101       user: user
 102     } do
 103       user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
 104       task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
 105
 106       Streamer.add_socket(
 107         "user:notification",
 108         %{transport_pid: task.pid, assigns: %{user: user}}
 109       )
 110
 111       {:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
 112       {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
 113       {:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
 114
 115       Streamer.stream("user:notification", notif)
 116       Task.await(task)
 117     end
 118
 119     test "it sends follow activities to the 'user:notification' stream", %{
 120       user: user
 121     } do
 122       user2 = insert(:user)
 123       task = Task.async(fn -> assert_receive {:text, _}, @streamer_timeout end)
 124
 125       Process.sleep(@streamer_start_wait)
 126
 127       Streamer.add_socket(
 128         "user:notification",
 129         %{transport_pid: task.pid, assigns: %{user: user}}
 130       )
 131
 132       {:ok, _follower, _followed, _activity} = CommonAPI.follow(user2, user)
 133
 134       # We don't directly pipe the notification to the streamer as it's already
 135       # generated as a side effect of CommonAPI.follow().
 136       Task.await(task)
 137     end
 138   end
 139
 140   test "it sends to public" do
 141     user = insert(:user)
 142     other_user = insert(:user)
 143
 144     task =
 145       Task.async(fn ->
 146         assert_receive {:text, _}, @streamer_timeout
 147       end)
 148
 149     fake_socket = %StreamerSocket{
 150       transport_pid: task.pid,
 151       user: user
 152     }
 153
 154     {:ok, activity} = CommonAPI.post(other_user, %{"status" => "Test"})
 155
 156     topics = %{
 157       "public" => [fake_socket]
 158     }
 159
 160     Worker.push_to_socket(topics, "public", activity)
 161
 162     Task.await(task)
 163
 164     task =
 165       Task.async(fn ->
 166         expected_event =
 167           %{
 168             "event" => "delete",
 169             "payload" => activity.id
 170           }
 171           |> Jason.encode!()
 172
 173         assert_receive {:text, received_event}, @streamer_timeout
 174         assert received_event == expected_event
 175       end)
 176
 177     fake_socket = %StreamerSocket{
 178       transport_pid: task.pid,
 179       user: user
 180     }
 181
 182     {:ok, activity} = CommonAPI.delete(activity.id, other_user)
 183
 184     topics = %{
 185       "public" => [fake_socket]
 186     }
 187
 188     Worker.push_to_socket(topics, "public", activity)
 189
 190     Task.await(task)
 191   end
 192
 193   describe "thread_containment" do
 194     test "it doesn't send to user if recipients invalid and thread containment is enabled" do
 195       Pleroma.Config.put([:instance, :skip_thread_containment], false)
 196       author = insert(:user)
 197       user = insert(:user)
 198       User.follow(user, author, "accept")
 199
 200       activity =
 201         insert(:note_activity,
 202           note:
 203             insert(:note,
 204               user: author,
 205               data: %{"to" => ["TEST-FFF"]}
 206             )
 207         )
 208
 209       task = Task.async(fn -> refute_receive {:text, _}, 1_000 end)
 210       fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
 211       topics = %{"public" => [fake_socket]}
 212       Worker.push_to_socket(topics, "public", activity)
 213
 214       Task.await(task)
 215     end
 216
 217     test "it sends message if recipients invalid and thread containment is disabled" do
 218       Pleroma.Config.put([:instance, :skip_thread_containment], true)
 219       author = insert(:user)
 220       user = insert(:user)
 221       User.follow(user, author, "accept")
 222
 223       activity =
 224         insert(:note_activity,
 225           note:
 226             insert(:note,
 227               user: author,
 228               data: %{"to" => ["TEST-FFF"]}
 229             )
 230         )
 231
 232       task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
 233       fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
 234       topics = %{"public" => [fake_socket]}
 235       Worker.push_to_socket(topics, "public", activity)
 236
 237       Task.await(task)
 238     end
 239
 240     test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
 241       Pleroma.Config.put([:instance, :skip_thread_containment], false)
 242       author = insert(:user)
 243       user = insert(:user, skip_thread_containment: true)
 244       User.follow(user, author, "accept")
 245
 246       activity =
 247         insert(:note_activity,
 248           note:
 249             insert(:note,
 250               user: author,
 251               data: %{"to" => ["TEST-FFF"]}
 252             )
 253         )
 254
 255       task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
 256       fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
 257       topics = %{"public" => [fake_socket]}
 258       Worker.push_to_socket(topics, "public", activity)
 259
 260       Task.await(task)
 261     end
 262   end
 263
 264   describe "blocks" do
 265     test "it doesn't send messages involving blocked users" do
 266       user = insert(:user)
 267       blocked_user = insert(:user)
 268       {:ok, _user_relationship} = User.block(user, blocked_user)
 269
 270       task =
 271         Task.async(fn ->
 272           refute_receive {:text, _}, 1_000
 273         end)
 274
 275       fake_socket = %StreamerSocket{
 276         transport_pid: task.pid,
 277         user: user
 278       }
 279
 280       {:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})
 281
 282       topics = %{
 283         "public" => [fake_socket]
 284       }
 285
 286       Worker.push_to_socket(topics, "public", activity)
 287
 288       Task.await(task)
 289     end
 290
 291     test "it doesn't send messages transitively involving blocked users" do
 292       blocker = insert(:user)
 293       blockee = insert(:user)
 294       friend = insert(:user)
 295
 296       task =
 297         Task.async(fn ->
 298           refute_receive {:text, _}, 1_000
 299         end)
 300
 301       fake_socket = %StreamerSocket{
 302         transport_pid: task.pid,
 303         user: blocker
 304       }
 305
 306       topics = %{
 307         "public" => [fake_socket]
 308       }
 309
 310       {:ok, _user_relationship} = User.block(blocker, blockee)
 311
 312       {:ok, activity_one} = CommonAPI.post(friend, %{"status" => "hey! @#{blockee.nickname}"})
 313
 314       Worker.push_to_socket(topics, "public", activity_one)
 315
 316       {:ok, activity_two} = CommonAPI.post(blockee, %{"status" => "hey! @#{friend.nickname}"})
 317
 318       Worker.push_to_socket(topics, "public", activity_two)
 319
 320       {:ok, activity_three} = CommonAPI.post(blockee, %{"status" => "hey! @#{blocker.nickname}"})
 321
 322       Worker.push_to_socket(topics, "public", activity_three)
 323
 324       Task.await(task)
 325     end
 326   end
 327
 328   test "it doesn't send unwanted DMs to list" do
 329     user_a = insert(:user)
 330     user_b = insert(:user)
 331     user_c = insert(:user)
 332
 333     {:ok, user_a} = User.follow(user_a, user_b)
 334
 335     {:ok, list} = List.create("Test", user_a)
 336     {:ok, list} = List.follow(list, user_b)
 337
 338     task =
 339       Task.async(fn ->
 340         refute_receive {:text, _}, 1_000
 341       end)
 342
 343     fake_socket = %StreamerSocket{
 344       transport_pid: task.pid,
 345       user: user_a
 346     }
 347
 348     {:ok, activity} =
 349       CommonAPI.post(user_b, %{
 350         "status" => "@#{user_c.nickname} Test",
 351         "visibility" => "direct"
 352       })
 353
 354     topics = %{
 355       "list:#{list.id}" => [fake_socket]
 356     }
 357
 358     Worker.handle_call({:stream, "list", activity}, self(), topics)
 359
 360     Task.await(task)
 361   end
 362
 363   test "it doesn't send unwanted private posts to list" do
 364     user_a = insert(:user)
 365     user_b = insert(:user)
 366
 367     {:ok, list} = List.create("Test", user_a)
 368     {:ok, list} = List.follow(list, user_b)
 369
 370     task =
 371       Task.async(fn ->
 372         refute_receive {:text, _}, 1_000
 373       end)
 374
 375     fake_socket = %StreamerSocket{
 376       transport_pid: task.pid,
 377       user: user_a
 378     }
 379
 380     {:ok, activity} =
 381       CommonAPI.post(user_b, %{
 382         "status" => "Test",
 383         "visibility" => "private"
 384       })
 385
 386     topics = %{
 387       "list:#{list.id}" => [fake_socket]
 388     }
 389
 390     Worker.handle_call({:stream, "list", activity}, self(), topics)
 391
 392     Task.await(task)
 393   end
 394
 395   test "it sends wanted private posts to list" do
 396     user_a = insert(:user)
 397     user_b = insert(:user)
 398
 399     {:ok, user_a} = User.follow(user_a, user_b)
 400
 401     {:ok, list} = List.create("Test", user_a)
 402     {:ok, list} = List.follow(list, user_b)
 403
 404     task =
 405       Task.async(fn ->
 406         assert_receive {:text, _}, 1_000
 407       end)
 408
 409     fake_socket = %StreamerSocket{
 410       transport_pid: task.pid,
 411       user: user_a
 412     }
 413
 414     {:ok, activity} =
 415       CommonAPI.post(user_b, %{
 416         "status" => "Test",
 417         "visibility" => "private"
 418       })
 419
 420     Streamer.add_socket(
 421       "list:#{list.id}",
 422       fake_socket
 423     )
 424
 425     Worker.handle_call({:stream, "list", activity}, self(), %{})
 426
 427     Task.await(task)
 428   end
 429
 430   test "it doesn't send muted reblogs" do
 431     user1 = insert(:user)
 432     user2 = insert(:user)
 433     user3 = insert(:user)
 434     CommonAPI.hide_reblogs(user1, user2)
 435
 436     task =
 437       Task.async(fn ->
 438         refute_receive {:text, _}, 1_000
 439       end)
 440
 441     fake_socket = %StreamerSocket{
 442       transport_pid: task.pid,
 443       user: user1
 444     }
 445
 446     {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
 447     {:ok, announce_activity, _} = CommonAPI.repeat(create_activity.id, user2)
 448
 449     topics = %{
 450       "public" => [fake_socket]
 451     }
 452
 453     Worker.push_to_socket(topics, "public", announce_activity)
 454
 455     Task.await(task)
 456   end
 457
 458   test "it doesn't send posts from muted threads" do
 459     user = insert(:user)
 460     user2 = insert(:user)
 461     {:ok, user2, user, _activity} = CommonAPI.follow(user2, user)
 462
 463     {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
 464
 465     {:ok, activity} = CommonAPI.add_mute(user2, activity)
 466
 467     task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
 468
 469     Streamer.add_socket(
 470       "user",
 471       %{transport_pid: task.pid, assigns: %{user: user2}}
 472     )
 473
 474     Streamer.stream("user", activity)
 475     Task.await(task)
 476   end
 477
 478   describe "direct streams" do
 479     setup do
 480       :ok
 481     end
 482
 483     test "it sends conversation update to the 'direct' stream", %{} do
 484       user = insert(:user)
 485       another_user = insert(:user)
 486
 487       task =
 488         Task.async(fn ->
 489           assert_receive {:text, received_event}, @streamer_timeout
 490
 491           assert %{"event" => "conversation", "payload" => received_payload} =
 492                    Jason.decode!(received_event)
 493
 494           assert %{"last_status" => last_status} = Jason.decode!(received_payload)
 495           [participation] = Participation.for_user(user)
 496           assert last_status["pleroma"]["direct_conversation_id"] == participation.id
 497         end)
 498
 499       Streamer.add_socket(
 500         "direct",
 501         %{transport_pid: task.pid, assigns: %{user: user}}
 502       )
 503
 504       {:ok, _create_activity} =
 505         CommonAPI.post(another_user, %{
 506           "status" => "hey @#{user.nickname}",
 507           "visibility" => "direct"
 508         })
 509
 510       Task.await(task)
 511     end
 512
 513     test "it doesn't send conversation update to the 'direct' stream when the last message in the conversation is deleted" do
 514       user = insert(:user)
 515       another_user = insert(:user)
 516
 517       {:ok, create_activity} =
 518         CommonAPI.post(another_user, %{
 519           "status" => "hi @#{user.nickname}",
 520           "visibility" => "direct"
 521         })
 522
 523       task =
 524         Task.async(fn ->
 525           assert_receive {:text, received_event}, @streamer_timeout
 526           assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
 527
 528           refute_receive {:text, _}, @streamer_timeout
 529         end)
 530
 531       Process.sleep(@streamer_start_wait)
 532
 533       Streamer.add_socket(
 534         "direct",
 535         %{transport_pid: task.pid, assigns: %{user: user}}
 536       )
 537
 538       {:ok, _} = CommonAPI.delete(create_activity.id, another_user)
 539
 540       Task.await(task)
 541     end
 542
 543     test "it sends conversation update to the 'direct' stream when a message is deleted" do
 544       user = insert(:user)
 545       another_user = insert(:user)
 546
 547       {:ok, create_activity} =
 548         CommonAPI.post(another_user, %{
 549           "status" => "hi @#{user.nickname}",
 550           "visibility" => "direct"
 551         })
 552
 553       {:ok, create_activity2} =
 554         CommonAPI.post(another_user, %{
 555           "status" => "hi @#{user.nickname}",
 556           "in_reply_to_status_id" => create_activity.id,
 557           "visibility" => "direct"
 558         })
 559
 560       task =
 561         Task.async(fn ->
 562           assert_receive {:text, received_event}, @streamer_timeout
 563           assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
 564
 565           assert_receive {:text, received_event}, @streamer_timeout
 566
 567           assert %{"event" => "conversation", "payload" => received_payload} =
 568                    Jason.decode!(received_event)
 569
 570           assert %{"last_status" => last_status} = Jason.decode!(received_payload)
 571           assert last_status["id"] == to_string(create_activity.id)
 572         end)
 573
 574       Process.sleep(@streamer_start_wait)
 575
 576       Streamer.add_socket(
 577         "direct",
 578         %{transport_pid: task.pid, assigns: %{user: user}}
 579       )
 580
 581       {:ok, _} = CommonAPI.delete(create_activity2.id, another_user)
 582
 583       Task.await(task)
 584     end
 585   end
 586 end