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