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

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