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