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