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