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

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2019 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)
 191       User.follow(user, author, "accept")
 192
 193       activity =
 194         insert(:note_activity,
 195           note:
 196             insert(:note,
 197               user: author,
 198               data: %{"to" => ["TEST-FFF"]}
 199             )
 200         )
 201
 202       task = Task.async(fn -> refute_receive {:text, _}, 1_000 end)
 203       fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
 204       topics = %{"public" => [fake_socket]}
 205       Worker.push_to_socket(topics, "public", activity)
 206
 207       Task.await(task)
 208     end
 209
 210     test "it sends message if recipients invalid and thread containment is disabled" do
 211       Pleroma.Config.put([:instance, :skip_thread_containment], true)
 212       author = insert(:user)
 213       user = insert(:user)
 214       User.follow(user, author, "accept")
 215
 216       activity =
 217         insert(:note_activity,
 218           note:
 219             insert(:note,
 220               user: author,
 221               data: %{"to" => ["TEST-FFF"]}
 222             )
 223         )
 224
 225       task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
 226       fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
 227       topics = %{"public" => [fake_socket]}
 228       Worker.push_to_socket(topics, "public", activity)
 229
 230       Task.await(task)
 231     end
 232
 233     test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
 234       Pleroma.Config.put([:instance, :skip_thread_containment], false)
 235       author = insert(:user)
 236       user = insert(:user, skip_thread_containment: true)
 237       User.follow(user, author, "accept")
 238
 239       activity =
 240         insert(:note_activity,
 241           note:
 242             insert(:note,
 243               user: author,
 244               data: %{"to" => ["TEST-FFF"]}
 245             )
 246         )
 247
 248       task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
 249       fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
 250       topics = %{"public" => [fake_socket]}
 251       Worker.push_to_socket(topics, "public", activity)
 252
 253       Task.await(task)
 254     end
 255   end
 256
 257   describe "blocks" do
 258     test "it doesn't send messages involving blocked users" do
 259       user = insert(:user)
 260       blocked_user = insert(:user)
 261       {:ok, user} = User.block(user, blocked_user)
 262
 263       task =
 264         Task.async(fn ->
 265           refute_receive {:text, _}, 1_000
 266         end)
 267
 268       fake_socket = %StreamerSocket{
 269         transport_pid: task.pid,
 270         user: user
 271       }
 272
 273       {:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})
 274
 275       topics = %{
 276         "public" => [fake_socket]
 277       }
 278
 279       Worker.push_to_socket(topics, "public", activity)
 280
 281       Task.await(task)
 282     end
 283
 284     test "it doesn't send messages transitively involving blocked users" do
 285       blocker = insert(:user)
 286       blockee = insert(:user)
 287       friend = insert(:user)
 288
 289       task =
 290         Task.async(fn ->
 291           refute_receive {:text, _}, 1_000
 292         end)
 293
 294       fake_socket = %StreamerSocket{
 295         transport_pid: task.pid,
 296         user: blocker
 297       }
 298
 299       topics = %{
 300         "public" => [fake_socket]
 301       }
 302
 303       {:ok, blocker} = User.block(blocker, blockee)
 304
 305       {:ok, activity_one} = CommonAPI.post(friend, %{"status" => "hey! @#{blockee.nickname}"})
 306
 307       Worker.push_to_socket(topics, "public", activity_one)
 308
 309       {:ok, activity_two} = CommonAPI.post(blockee, %{"status" => "hey! @#{friend.nickname}"})
 310
 311       Worker.push_to_socket(topics, "public", activity_two)
 312
 313       {:ok, activity_three} = CommonAPI.post(blockee, %{"status" => "hey! @#{blocker.nickname}"})
 314
 315       Worker.push_to_socket(topics, "public", activity_three)
 316
 317       Task.await(task)
 318     end
 319   end
 320
 321   test "it doesn't send unwanted DMs to list" do
 322     user_a = insert(:user)
 323     user_b = insert(:user)
 324     user_c = insert(:user)
 325
 326     {:ok, user_a} = User.follow(user_a, user_b)
 327
 328     {:ok, list} = List.create("Test", user_a)
 329     {:ok, list} = List.follow(list, user_b)
 330
 331     task =
 332       Task.async(fn ->
 333         refute_receive {:text, _}, 1_000
 334       end)
 335
 336     fake_socket = %StreamerSocket{
 337       transport_pid: task.pid,
 338       user: user_a
 339     }
 340
 341     {:ok, activity} =
 342       CommonAPI.post(user_b, %{
 343         "status" => "@#{user_c.nickname} Test",
 344         "visibility" => "direct"
 345       })
 346
 347     topics = %{
 348       "list:#{list.id}" => [fake_socket]
 349     }
 350
 351     Worker.handle_call({:stream, "list", activity}, self(), topics)
 352
 353     Task.await(task)
 354   end
 355
 356   test "it doesn't send unwanted private posts to list" do
 357     user_a = insert(:user)
 358     user_b = insert(:user)
 359
 360     {:ok, list} = List.create("Test", user_a)
 361     {:ok, list} = List.follow(list, user_b)
 362
 363     task =
 364       Task.async(fn ->
 365         refute_receive {:text, _}, 1_000
 366       end)
 367
 368     fake_socket = %StreamerSocket{
 369       transport_pid: task.pid,
 370       user: user_a
 371     }
 372
 373     {:ok, activity} =
 374       CommonAPI.post(user_b, %{
 375         "status" => "Test",
 376         "visibility" => "private"
 377       })
 378
 379     topics = %{
 380       "list:#{list.id}" => [fake_socket]
 381     }
 382
 383     Worker.handle_call({:stream, "list", activity}, self(), topics)
 384
 385     Task.await(task)
 386   end
 387
 388   test "it sends wanted private posts to list" do
 389     user_a = insert(:user)
 390     user_b = insert(:user)
 391
 392     {:ok, user_a} = User.follow(user_a, user_b)
 393
 394     {:ok, list} = List.create("Test", user_a)
 395     {:ok, list} = List.follow(list, user_b)
 396
 397     task =
 398       Task.async(fn ->
 399         assert_receive {:text, _}, 1_000
 400       end)
 401
 402     fake_socket = %StreamerSocket{
 403       transport_pid: task.pid,
 404       user: user_a
 405     }
 406
 407     {:ok, activity} =
 408       CommonAPI.post(user_b, %{
 409         "status" => "Test",
 410         "visibility" => "private"
 411       })
 412
 413     Streamer.add_socket(
 414       "list:#{list.id}",
 415       fake_socket
 416     )
 417
 418     Worker.handle_call({:stream, "list", activity}, self(), %{})
 419
 420     Task.await(task)
 421   end
 422
 423   test "it doesn't send muted reblogs" do
 424     user1 = insert(:user)
 425     user2 = insert(:user)
 426     user3 = insert(:user)
 427     CommonAPI.hide_reblogs(user1, user2)
 428
 429     task =
 430       Task.async(fn ->
 431         refute_receive {:text, _}, 1_000
 432       end)
 433
 434     fake_socket = %StreamerSocket{
 435       transport_pid: task.pid,
 436       user: user1
 437     }
 438
 439     {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
 440     {:ok, announce_activity, _} = CommonAPI.repeat(create_activity.id, user2)
 441
 442     topics = %{
 443       "public" => [fake_socket]
 444     }
 445
 446     Worker.push_to_socket(topics, "public", announce_activity)
 447
 448     Task.await(task)
 449   end
 450
 451   test "it doesn't send posts from muted threads" do
 452     user = insert(:user)
 453     user2 = insert(:user)
 454     {:ok, user2, user, _activity} = CommonAPI.follow(user2, user)
 455
 456     {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
 457
 458     {:ok, activity} = CommonAPI.add_mute(user2, activity)
 459
 460     task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)
 461
 462     Process.sleep(4000)
 463
 464     Streamer.add_socket(
 465       "user",
 466       %{transport_pid: task.pid, assigns: %{user: user2}}
 467     )
 468
 469     Streamer.stream("user", activity)
 470     Task.await(task)
 471   end
 472
 473   describe "direct streams" do
 474     setup do
 475       :ok
 476     end
 477
 478     test "it sends conversation update to the 'direct' stream", %{} do
 479       user = insert(:user)
 480       another_user = insert(:user)
 481
 482       task =
 483         Task.async(fn ->
 484           assert_receive {:text, _received_event}, 4_000
 485         end)
 486
 487       Streamer.add_socket(
 488         "direct",
 489         %{transport_pid: task.pid, assigns: %{user: user}}
 490       )
 491
 492       {:ok, _create_activity} =
 493         CommonAPI.post(another_user, %{
 494           "status" => "hey @#{user.nickname}",
 495           "visibility" => "direct"
 496         })
 497
 498       Task.await(task)
 499     end
 500
 501     test "it doesn't send conversation update to the 'direct' streamj when the last message in the conversation is deleted" do
 502       user = insert(:user)
 503       another_user = insert(:user)
 504
 505       {:ok, create_activity} =
 506         CommonAPI.post(another_user, %{
 507           "status" => "hi @#{user.nickname}",
 508           "visibility" => "direct"
 509         })
 510
 511       task =
 512         Task.async(fn ->
 513           assert_receive {:text, received_event}, 4_000
 514           assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
 515
 516           refute_receive {:text, _}, 4_000
 517         end)
 518
 519       Process.sleep(1000)
 520
 521       Streamer.add_socket(
 522         "direct",
 523         %{transport_pid: task.pid, assigns: %{user: user}}
 524       )
 525
 526       {:ok, _} = CommonAPI.delete(create_activity.id, another_user)
 527
 528       Task.await(task)
 529     end
 530
 531     test "it sends conversation update to the 'direct' stream when a message is deleted" do
 532       user = insert(:user)
 533       another_user = insert(:user)
 534
 535       {:ok, create_activity} =
 536         CommonAPI.post(another_user, %{
 537           "status" => "hi @#{user.nickname}",
 538           "visibility" => "direct"
 539         })
 540
 541       {:ok, create_activity2} =
 542         CommonAPI.post(another_user, %{
 543           "status" => "hi @#{user.nickname}",
 544           "in_reply_to_status_id" => create_activity.id,
 545           "visibility" => "direct"
 546         })
 547
 548       task =
 549         Task.async(fn ->
 550           assert_receive {:text, received_event}, 4_000
 551           assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
 552
 553           assert_receive {:text, received_event}, 4_000
 554
 555           assert %{"event" => "conversation", "payload" => received_payload} =
 556                    Jason.decode!(received_event)
 557
 558           assert %{"last_status" => last_status} = Jason.decode!(received_payload)
 559           assert last_status["id"] == to_string(create_activity.id)
 560         end)
 561
 562       Process.sleep(1000)
 563
 564       Streamer.add_socket(
 565         "direct",
 566         %{transport_pid: task.pid, assigns: %{user: user}}
 567       )
 568
 569       {:ok, _} = CommonAPI.delete(create_activity2.id, another_user)
 570
 571       Task.await(task)
 572     end
 573   end
 574 end