git.squeep.com Git - akkoma/blob - test/web/streamer/streamer_test.exs

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