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