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 sends chat message notifications to the 'user:notification' stream", %{user: user} do
 130       other_user = insert(:user)
 131
 132       {:ok, create_activity} = CommonAPI.post_chat_message(other_user, user, "hey")
 133
 134       notify =
 135         Repo.get_by(Pleroma.Notification, user_id: user.id, activity_id: create_activity.id)
 136         |> Repo.preload(:activity)
 137
 138       Streamer.get_topic_and_add_socket("user:notification", user)
 139       Streamer.stream("user:notification", notify)
 140       assert_receive {:render_with_user, _, _, ^notify}
 141       refute Streamer.filtered_by_user?(user, notify)
 142     end
 143
 144     test "it doesn't send notify to the 'user:notification' stream when a user is blocked", %{
 145       user: user
 146     } do
 147       blocked = insert(:user)
 148       {:ok, _user_relationship} = User.block(user, blocked)
 149
 150       Streamer.get_topic_and_add_socket("user:notification", user)
 151
 152       {:ok, activity} = CommonAPI.post(user, %{status: ":("})
 153       {:ok, _} = CommonAPI.favorite(blocked, activity.id)
 154
 155       refute_receive _
 156     end
 157
 158     test "it doesn't send notify to the 'user:notification' stream when a thread is muted", %{
 159       user: user
 160     } do
 161       user2 = insert(:user)
 162
 163       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
 164       {:ok, _} = CommonAPI.add_mute(user, activity)
 165
 166       Streamer.get_topic_and_add_socket("user:notification", user)
 167
 168       {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
 169
 170       refute_receive _
 171       assert Streamer.filtered_by_user?(user, favorite_activity)
 172     end
 173
 174     test "it sends favorite to 'user:notification' stream'", %{
 175       user: user
 176     } do
 177       user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
 178
 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       assert_receive {:render_with_user, _, "notification.json", notif}
 184       assert notif.activity.id == favorite_activity.id
 185       refute Streamer.filtered_by_user?(user, notif)
 186     end
 187
 188     test "it doesn't send the 'user:notification' stream' when a domain is blocked", %{
 189       user: user
 190     } do
 191       user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
 192
 193       {:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
 194       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
 195       Streamer.get_topic_and_add_socket("user:notification", user)
 196       {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
 197
 198       refute_receive _
 199       assert Streamer.filtered_by_user?(user, favorite_activity)
 200     end
 201
 202     test "it sends follow activities to the 'user:notification' stream", %{
 203       user: user
 204     } do
 205       user_url = user.ap_id
 206       user2 = insert(:user)
 207
 208       body =
 209         File.read!("test/fixtures/users_mock/localhost.json")
 210         |> String.replace("{{nickname}}", user.nickname)
 211         |> Jason.encode!()
 212
 213       Tesla.Mock.mock_global(fn
 214         %{method: :get, url: ^user_url} ->
 215           %Tesla.Env{status: 200, body: body}
 216       end)
 217
 218       Streamer.get_topic_and_add_socket("user:notification", user)
 219       {:ok, _follower, _followed, follow_activity} = CommonAPI.follow(user2, user)
 220
 221       assert_receive {:render_with_user, _, "notification.json", notif}
 222       assert notif.activity.id == follow_activity.id
 223       refute Streamer.filtered_by_user?(user, notif)
 224     end
 225   end
 226
 227   test "it sends to public authenticated" do
 228     user = insert(:user)
 229     other_user = insert(:user)
 230
 231     Streamer.get_topic_and_add_socket("public", other_user)
 232
 233     {:ok, activity} = CommonAPI.post(user, %{status: "Test"})
 234     assert_receive {:render_with_user, _, _, ^activity}
 235     refute Streamer.filtered_by_user?(user, activity)
 236   end
 237
 238   test "works for deletions" do
 239     user = insert(:user)
 240     other_user = insert(:user)
 241     {:ok, activity} = CommonAPI.post(other_user, %{status: "Test"})
 242
 243     Streamer.get_topic_and_add_socket("public", user)
 244
 245     {:ok, _} = CommonAPI.delete(activity.id, other_user)
 246     activity_id = activity.id
 247     assert_receive {:text, event}
 248     assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
 249   end
 250
 251   test "it sends to public unauthenticated" do
 252     user = insert(:user)
 253
 254     Streamer.get_topic_and_add_socket("public", nil)
 255
 256     {:ok, activity} = CommonAPI.post(user, %{status: "Test"})
 257     activity_id = activity.id
 258     assert_receive {:text, event}
 259     assert %{"event" => "update", "payload" => payload} = Jason.decode!(event)
 260     assert %{"id" => ^activity_id} = Jason.decode!(payload)
 261
 262     {:ok, _} = CommonAPI.delete(activity.id, user)
 263     assert_receive {:text, event}
 264     assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
 265   end
 266
 267   describe "thread_containment" do
 268     test "it filters to user if recipients invalid and thread containment is enabled" do
 269       Pleroma.Config.put([:instance, :skip_thread_containment], false)
 270       author = insert(:user)
 271       user = insert(:user)
 272       User.follow(user, author, :follow_accept)
 273
 274       activity =
 275         insert(:note_activity,
 276           note:
 277             insert(:note,
 278               user: author,
 279               data: %{"to" => ["TEST-FFF"]}
 280             )
 281         )
 282
 283       Streamer.get_topic_and_add_socket("public", user)
 284       Streamer.stream("public", activity)
 285       assert_receive {:render_with_user, _, _, ^activity}
 286       assert Streamer.filtered_by_user?(user, activity)
 287     end
 288
 289     test "it sends message if recipients invalid and thread containment is disabled" do
 290       Pleroma.Config.put([:instance, :skip_thread_containment], true)
 291       author = insert(:user)
 292       user = insert(:user)
 293       User.follow(user, author, :follow_accept)
 294
 295       activity =
 296         insert(:note_activity,
 297           note:
 298             insert(:note,
 299               user: author,
 300               data: %{"to" => ["TEST-FFF"]}
 301             )
 302         )
 303
 304       Streamer.get_topic_and_add_socket("public", user)
 305       Streamer.stream("public", activity)
 306
 307       assert_receive {:render_with_user, _, _, ^activity}
 308       refute Streamer.filtered_by_user?(user, activity)
 309     end
 310
 311     test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
 312       Pleroma.Config.put([:instance, :skip_thread_containment], false)
 313       author = insert(:user)
 314       user = insert(:user, skip_thread_containment: true)
 315       User.follow(user, author, :follow_accept)
 316
 317       activity =
 318         insert(:note_activity,
 319           note:
 320             insert(:note,
 321               user: author,
 322               data: %{"to" => ["TEST-FFF"]}
 323             )
 324         )
 325
 326       Streamer.get_topic_and_add_socket("public", user)
 327       Streamer.stream("public", activity)
 328
 329       assert_receive {:render_with_user, _, _, ^activity}
 330       refute Streamer.filtered_by_user?(user, activity)
 331     end
 332   end
 333
 334   describe "blocks" do
 335     test "it filters messages involving blocked users" do
 336       user = insert(:user)
 337       blocked_user = insert(:user)
 338       {:ok, _user_relationship} = User.block(user, blocked_user)
 339
 340       Streamer.get_topic_and_add_socket("public", user)
 341       {:ok, activity} = CommonAPI.post(blocked_user, %{status: "Test"})
 342       assert_receive {:render_with_user, _, _, ^activity}
 343       assert Streamer.filtered_by_user?(user, activity)
 344     end
 345
 346     test "it filters messages transitively involving blocked users" do
 347       blocker = insert(:user)
 348       blockee = insert(:user)
 349       friend = insert(:user)
 350
 351       Streamer.get_topic_and_add_socket("public", blocker)
 352
 353       {:ok, _user_relationship} = User.block(blocker, blockee)
 354
 355       {:ok, activity_one} = CommonAPI.post(friend, %{status: "hey! @#{blockee.nickname}"})
 356
 357       assert_receive {:render_with_user, _, _, ^activity_one}
 358       assert Streamer.filtered_by_user?(blocker, activity_one)
 359
 360       {:ok, activity_two} = CommonAPI.post(blockee, %{status: "hey! @#{friend.nickname}"})
 361
 362       assert_receive {:render_with_user, _, _, ^activity_two}
 363       assert Streamer.filtered_by_user?(blocker, activity_two)
 364
 365       {:ok, activity_three} = CommonAPI.post(blockee, %{status: "hey! @#{blocker.nickname}"})
 366
 367       assert_receive {:render_with_user, _, _, ^activity_three}
 368       assert Streamer.filtered_by_user?(blocker, activity_three)
 369     end
 370   end
 371
 372   describe "lists" do
 373     test "it doesn't send unwanted DMs to list" do
 374       user_a = insert(:user)
 375       user_b = insert(:user)
 376       user_c = insert(:user)
 377
 378       {:ok, user_a} = User.follow(user_a, user_b)
 379
 380       {:ok, list} = List.create("Test", user_a)
 381       {:ok, list} = List.follow(list, user_b)
 382
 383       Streamer.get_topic_and_add_socket("list", user_a, %{"list" => list.id})
 384
 385       {:ok, _activity} =
 386         CommonAPI.post(user_b, %{
 387           status: "@#{user_c.nickname} Test",
 388           visibility: "direct"
 389         })
 390
 391       refute_receive _
 392     end
 393
 394     test "it doesn't send unwanted private posts to list" do
 395       user_a = insert(:user)
 396       user_b = insert(:user)
 397
 398       {:ok, list} = List.create("Test", user_a)
 399       {:ok, list} = List.follow(list, user_b)
 400
 401       Streamer.get_topic_and_add_socket("list", user_a, %{"list" => list.id})
 402
 403       {:ok, _activity} =
 404         CommonAPI.post(user_b, %{
 405           status: "Test",
 406           visibility: "private"
 407         })
 408
 409       refute_receive _
 410     end
 411
 412     test "it sends wanted private posts to list" do
 413       user_a = insert(:user)
 414       user_b = insert(:user)
 415
 416       {:ok, user_a} = User.follow(user_a, user_b)
 417
 418       {:ok, list} = List.create("Test", user_a)
 419       {:ok, list} = List.follow(list, user_b)
 420
 421       Streamer.get_topic_and_add_socket("list", user_a, %{"list" => list.id})
 422
 423       {:ok, activity} =
 424         CommonAPI.post(user_b, %{
 425           status: "Test",
 426           visibility: "private"
 427         })
 428
 429       assert_receive {:render_with_user, _, _, ^activity}
 430       refute Streamer.filtered_by_user?(user_a, activity)
 431     end
 432   end
 433
 434   describe "muted reblogs" do
 435     test "it filters muted reblogs" do
 436       user1 = insert(:user)
 437       user2 = insert(:user)
 438       user3 = insert(:user)
 439       CommonAPI.follow(user1, user2)
 440       CommonAPI.hide_reblogs(user1, user2)
 441
 442       {:ok, create_activity} = CommonAPI.post(user3, %{status: "I'm kawen"})
 443
 444       Streamer.get_topic_and_add_socket("user", user1)
 445       {:ok, announce_activity} = CommonAPI.repeat(create_activity.id, user2)
 446       assert_receive {:render_with_user, _, _, ^announce_activity}
 447       assert Streamer.filtered_by_user?(user1, announce_activity)
 448     end
 449
 450     test "it filters reblog notification for reblog-muted actors" do
 451       user1 = insert(:user)
 452       user2 = insert(:user)
 453       CommonAPI.follow(user1, user2)
 454       CommonAPI.hide_reblogs(user1, user2)
 455
 456       {:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
 457       Streamer.get_topic_and_add_socket("user", user1)
 458       {:ok, _announce_activity} = CommonAPI.repeat(create_activity.id, user2)
 459
 460       assert_receive {:render_with_user, _, "notification.json", notif}
 461       assert Streamer.filtered_by_user?(user1, notif)
 462     end
 463
 464     test "it send non-reblog notification for reblog-muted actors" do
 465       user1 = insert(:user)
 466       user2 = insert(:user)
 467       CommonAPI.follow(user1, user2)
 468       CommonAPI.hide_reblogs(user1, user2)
 469
 470       {:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
 471       Streamer.get_topic_and_add_socket("user", user1)
 472       {:ok, _favorite_activity} = CommonAPI.favorite(user2, create_activity.id)
 473
 474       assert_receive {:render_with_user, _, "notification.json", notif}
 475       refute Streamer.filtered_by_user?(user1, notif)
 476     end
 477   end
 478
 479   test "it filters posts from muted threads" do
 480     user = insert(:user)
 481     user2 = insert(:user)
 482     Streamer.get_topic_and_add_socket("user", user2)
 483     {:ok, user2, user, _activity} = CommonAPI.follow(user2, user)
 484     {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
 485     {:ok, _} = CommonAPI.add_mute(user2, activity)
 486     assert_receive {:render_with_user, _, _, ^activity}
 487     assert Streamer.filtered_by_user?(user2, activity)
 488   end
 489
 490   describe "direct streams" do
 491     setup do
 492       :ok
 493     end
 494
 495     test "it sends conversation update to the 'direct' stream", %{} do
 496       user = insert(:user)
 497       another_user = insert(:user)
 498
 499       Streamer.get_topic_and_add_socket("direct", user)
 500
 501       {:ok, _create_activity} =
 502         CommonAPI.post(another_user, %{
 503           status: "hey @#{user.nickname}",
 504           visibility: "direct"
 505         })
 506
 507       assert_receive {:text, received_event}
 508
 509       assert %{"event" => "conversation", "payload" => received_payload} =
 510                Jason.decode!(received_event)
 511
 512       assert %{"last_status" => last_status} = Jason.decode!(received_payload)
 513       [participation] = Participation.for_user(user)
 514       assert last_status["pleroma"]["direct_conversation_id"] == participation.id
 515     end
 516
 517     test "it doesn't send conversation update to the 'direct' stream when the last message in the conversation is deleted" do
 518       user = insert(:user)
 519       another_user = insert(:user)
 520
 521       Streamer.get_topic_and_add_socket("direct", user)
 522
 523       {:ok, create_activity} =
 524         CommonAPI.post(another_user, %{
 525           status: "hi @#{user.nickname}",
 526           visibility: "direct"
 527         })
 528
 529       create_activity_id = create_activity.id
 530       assert_receive {:render_with_user, _, _, ^create_activity}
 531       assert_receive {:text, received_conversation1}
 532       assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
 533
 534       {:ok, _} = CommonAPI.delete(create_activity_id, another_user)
 535
 536       assert_receive {:text, received_event}
 537
 538       assert %{"event" => "delete", "payload" => ^create_activity_id} =
 539                Jason.decode!(received_event)
 540
 541       refute_receive _
 542     end
 543
 544     test "it sends conversation update to the 'direct' stream when a message is deleted" do
 545       user = insert(:user)
 546       another_user = insert(:user)
 547       Streamer.get_topic_and_add_socket("direct", user)
 548
 549       {:ok, create_activity} =
 550         CommonAPI.post(another_user, %{
 551           status: "hi @#{user.nickname}",
 552           visibility: "direct"
 553         })
 554
 555       {:ok, create_activity2} =
 556         CommonAPI.post(another_user, %{
 557           status: "hi @#{user.nickname} 2",
 558           in_reply_to_status_id: create_activity.id,
 559           visibility: "direct"
 560         })
 561
 562       assert_receive {:render_with_user, _, _, ^create_activity}
 563       assert_receive {:render_with_user, _, _, ^create_activity2}
 564       assert_receive {:text, received_conversation1}
 565       assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
 566       assert_receive {:text, received_conversation1}
 567       assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
 568
 569       {:ok, _} = CommonAPI.delete(create_activity2.id, another_user)
 570
 571       assert_receive {:text, received_event}
 572       assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
 573
 574       assert_receive {:text, received_event}
 575
 576       assert %{"event" => "conversation", "payload" => received_payload} =
 577                Jason.decode!(received_event)
 578
 579       assert %{"last_status" => last_status} = Jason.decode!(received_payload)
 580       assert last_status["id"] == to_string(create_activity.id)
 581     end
 582   end
 583 end