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