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