git.squeep.com Git - akkoma/blob - test/pleroma/web/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.Chat
  11   alias Pleroma.Chat.MessageReference
  12   alias Pleroma.Conversation.Participation
  13   alias Pleroma.List
  14   alias Pleroma.Object
  15   alias Pleroma.User
  16   alias Pleroma.Web.CommonAPI
  17   alias Pleroma.Web.Streamer
  18   alias Pleroma.Web.StreamerView
  19
  20   @moduletag needs_streamer: true, capture_log: true
  21
  22   setup do: clear_config([:instance, :skip_thread_containment])
  23
  24   describe "get_topic/_ (unauthenticated)" do
  25     test "allows public" do
  26       assert {:ok, "public"} = Streamer.get_topic("public", nil, nil)
  27       assert {:ok, "public:local"} = Streamer.get_topic("public:local", nil, nil)
  28       assert {:ok, "public:media"} = Streamer.get_topic("public:media", nil, nil)
  29       assert {:ok, "public:local:media"} = Streamer.get_topic("public:local:media", nil, nil)
  30     end
  31
  32     test "allows hashtag streams" do
  33       assert {:ok, "hashtag:cofe"} = Streamer.get_topic("hashtag", nil, nil, %{"tag" => "cofe"})
  34     end
  35
  36     test "disallows user streams" do
  37       assert {:error, _} = Streamer.get_topic("user", nil, nil)
  38       assert {:error, _} = Streamer.get_topic("user:notification", nil, nil)
  39       assert {:error, _} = Streamer.get_topic("direct", nil, nil)
  40     end
  41
  42     test "disallows list streams" do
  43       assert {:error, _} = Streamer.get_topic("list", nil, nil, %{"list" => 42})
  44     end
  45   end
  46
  47   describe "get_topic/_ (authenticated)" do
  48     setup do: oauth_access(["read"])
  49
  50     test "allows public streams (regardless of OAuth token scopes)", %{
  51       user: user,
  52       token: read_oauth_token
  53     } do
  54       with oauth_token <- [nil, read_oauth_token] do
  55         assert {:ok, "public"} = Streamer.get_topic("public", user, oauth_token)
  56         assert {:ok, "public:local"} = Streamer.get_topic("public:local", user, oauth_token)
  57         assert {:ok, "public:media"} = Streamer.get_topic("public:media", user, oauth_token)
  58
  59         assert {:ok, "public:local:media"} =
  60                  Streamer.get_topic("public:local:media", user, oauth_token)
  61       end
  62     end
  63
  64     test "allows user streams (with proper OAuth token scopes)", %{
  65       user: user,
  66       token: read_oauth_token
  67     } do
  68       %{token: read_notifications_token} = oauth_access(["read:notifications"], user: user)
  69       %{token: read_statuses_token} = oauth_access(["read:statuses"], user: user)
  70       %{token: badly_scoped_token} = oauth_access(["irrelevant:scope"], user: user)
  71
  72       expected_user_topic = "user:#{user.id}"
  73       expected_notification_topic = "user:notification:#{user.id}"
  74       expected_direct_topic = "direct:#{user.id}"
  75       expected_pleroma_chat_topic = "user:pleroma_chat:#{user.id}"
  76
  77       for valid_user_token <- [read_oauth_token, read_statuses_token] do
  78         assert {:ok, ^expected_user_topic} = Streamer.get_topic("user", user, valid_user_token)
  79
  80         assert {:ok, ^expected_direct_topic} =
  81                  Streamer.get_topic("direct", user, valid_user_token)
  82
  83         assert {:ok, ^expected_pleroma_chat_topic} =
  84                  Streamer.get_topic("user:pleroma_chat", user, valid_user_token)
  85       end
  86
  87       for invalid_user_token <- [read_notifications_token, badly_scoped_token],
  88           user_topic <- ["user", "direct", "user:pleroma_chat"] do
  89         assert {:error, :unauthorized} = Streamer.get_topic(user_topic, user, invalid_user_token)
  90       end
  91
  92       for valid_notification_token <- [read_oauth_token, read_notifications_token] do
  93         assert {:ok, ^expected_notification_topic} =
  94                  Streamer.get_topic("user:notification", user, valid_notification_token)
  95       end
  96
  97       for invalid_notification_token <- [read_statuses_token, badly_scoped_token] do
  98         assert {:error, :unauthorized} =
  99                  Streamer.get_topic("user:notification", user, invalid_notification_token)
 100       end
 101     end
 102
 103     test "allows hashtag streams (regardless of OAuth token scopes)", %{
 104       user: user,
 105       token: read_oauth_token
 106     } do
 107       for oauth_token <- [nil, read_oauth_token] do
 108         assert {:ok, "hashtag:cofe"} =
 109                  Streamer.get_topic("hashtag", user, oauth_token, %{"tag" => "cofe"})
 110       end
 111     end
 112
 113     test "disallows registering to another user's stream", %{user: user, token: read_oauth_token} do
 114       another_user = insert(:user)
 115       assert {:error, _} = Streamer.get_topic("user:#{another_user.id}", user, read_oauth_token)
 116
 117       assert {:error, _} =
 118                Streamer.get_topic("user:notification:#{another_user.id}", user, read_oauth_token)
 119
 120       assert {:error, _} = Streamer.get_topic("direct:#{another_user.id}", user, read_oauth_token)
 121     end
 122
 123     test "allows list stream that are owned by the user (with `read` or `read:lists` scopes)", %{
 124       user: user,
 125       token: read_oauth_token
 126     } do
 127       %{token: read_lists_token} = oauth_access(["read:lists"], user: user)
 128       %{token: invalid_token} = oauth_access(["irrelevant:scope"], user: user)
 129       {:ok, list} = List.create("Test", user)
 130
 131       assert {:error, _} = Streamer.get_topic("list:#{list.id}", user, read_oauth_token)
 132
 133       for valid_token <- [read_oauth_token, read_lists_token] do
 134         assert {:ok, _} = Streamer.get_topic("list", user, valid_token, %{"list" => list.id})
 135       end
 136
 137       assert {:error, _} = Streamer.get_topic("list", user, invalid_token, %{"list" => list.id})
 138     end
 139
 140     test "disallows list stream that are not owned by the user", %{user: user, token: oauth_token} do
 141       another_user = insert(:user)
 142       {:ok, list} = List.create("Test", another_user)
 143
 144       assert {:error, _} = Streamer.get_topic("list:#{list.id}", user, oauth_token)
 145       assert {:error, _} = Streamer.get_topic("list", user, oauth_token, %{"list" => list.id})
 146     end
 147   end
 148
 149   describe "user streams" do
 150     setup do
 151       %{user: user, token: token} = oauth_access(["read"])
 152       notify = insert(:notification, user: user, activity: build(:note_activity))
 153       {:ok, %{user: user, notify: notify, token: token}}
 154     end
 155
 156     test "it streams the user's post in the 'user' stream", %{user: user, token: oauth_token} do
 157       Streamer.get_topic_and_add_socket("user", user, oauth_token)
 158       {:ok, activity} = CommonAPI.post(user, %{status: "hey"})
 159
 160       assert_receive {:render_with_user, _, _, ^activity}
 161       refute Streamer.filtered_by_user?(user, activity)
 162     end
 163
 164     test "it streams boosts of the user in the 'user' stream", %{user: user, token: oauth_token} do
 165       Streamer.get_topic_and_add_socket("user", user, oauth_token)
 166
 167       other_user = insert(:user)
 168       {:ok, activity} = CommonAPI.post(other_user, %{status: "hey"})
 169       {:ok, announce} = CommonAPI.repeat(activity.id, user)
 170
 171       assert_receive {:render_with_user, Pleroma.Web.StreamerView, "update.json", ^announce}
 172       refute Streamer.filtered_by_user?(user, announce)
 173     end
 174
 175     test "it does not stream announces of the user's own posts in the 'user' stream", %{
 176       user: user,
 177       token: oauth_token
 178     } do
 179       Streamer.get_topic_and_add_socket("user", user, oauth_token)
 180
 181       other_user = insert(:user)
 182       {:ok, activity} = CommonAPI.post(user, %{status: "hey"})
 183       {:ok, announce} = CommonAPI.repeat(activity.id, other_user)
 184
 185       assert Streamer.filtered_by_user?(user, announce)
 186     end
 187
 188     test "it does stream notifications announces of the user's own posts in the 'user' stream", %{
 189       user: user,
 190       token: oauth_token
 191     } do
 192       Streamer.get_topic_and_add_socket("user", user, oauth_token)
 193
 194       other_user = insert(:user)
 195       {:ok, activity} = CommonAPI.post(user, %{status: "hey"})
 196       {:ok, announce} = CommonAPI.repeat(activity.id, other_user)
 197
 198       notification =
 199         Pleroma.Notification
 200         |> Repo.get_by(%{user_id: user.id, activity_id: announce.id})
 201         |> Repo.preload(:activity)
 202
 203       refute Streamer.filtered_by_user?(user, notification)
 204     end
 205
 206     test "it streams boosts of mastodon user in the 'user' stream", %{
 207       user: user,
 208       token: oauth_token
 209     } do
 210       Streamer.get_topic_and_add_socket("user", user, oauth_token)
 211
 212       other_user = insert(:user)
 213       {:ok, activity} = CommonAPI.post(other_user, %{status: "hey"})
 214
 215       data =
 216         File.read!("test/fixtures/mastodon-announce.json")
 217         |> Poison.decode!()
 218         |> Map.put("object", activity.data["object"])
 219         |> Map.put("actor", user.ap_id)
 220
 221       {:ok, %Pleroma.Activity{data: _data, local: false} = announce} =
 222         Pleroma.Web.ActivityPub.Transmogrifier.handle_incoming(data)
 223
 224       assert_receive {:render_with_user, Pleroma.Web.StreamerView, "update.json", ^announce}
 225       refute Streamer.filtered_by_user?(user, announce)
 226     end
 227
 228     test "it sends notify to in the 'user' stream", %{
 229       user: user,
 230       token: oauth_token,
 231       notify: notify
 232     } do
 233       Streamer.get_topic_and_add_socket("user", user, oauth_token)
 234       Streamer.stream("user", notify)
 235
 236       assert_receive {:render_with_user, _, _, ^notify}
 237       refute Streamer.filtered_by_user?(user, notify)
 238     end
 239
 240     test "it sends notify to in the 'user:notification' stream", %{
 241       user: user,
 242       token: oauth_token,
 243       notify: notify
 244     } do
 245       Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 246       Streamer.stream("user:notification", notify)
 247
 248       assert_receive {:render_with_user, _, _, ^notify}
 249       refute Streamer.filtered_by_user?(user, notify)
 250     end
 251
 252     test "it sends chat messages to the 'user:pleroma_chat' stream", %{
 253       user: user,
 254       token: oauth_token
 255     } do
 256       other_user = insert(:user)
 257
 258       {:ok, create_activity} = CommonAPI.post_chat_message(other_user, user, "hey cirno")
 259       object = Object.normalize(create_activity, false)
 260       chat = Chat.get(user.id, other_user.ap_id)
 261       cm_ref = MessageReference.for_chat_and_object(chat, object)
 262       cm_ref = %{cm_ref | chat: chat, object: object}
 263
 264       Streamer.get_topic_and_add_socket("user:pleroma_chat", user, oauth_token)
 265       Streamer.stream("user:pleroma_chat", {user, cm_ref})
 266
 267       text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref})
 268
 269       assert text =~ "hey cirno"
 270       assert_receive {:text, ^text}
 271     end
 272
 273     test "it sends chat messages to the 'user' stream", %{user: user, token: oauth_token} do
 274       other_user = insert(:user)
 275
 276       {:ok, create_activity} = CommonAPI.post_chat_message(other_user, user, "hey cirno")
 277       object = Object.normalize(create_activity, false)
 278       chat = Chat.get(user.id, other_user.ap_id)
 279       cm_ref = MessageReference.for_chat_and_object(chat, object)
 280       cm_ref = %{cm_ref | chat: chat, object: object}
 281
 282       Streamer.get_topic_and_add_socket("user", user, oauth_token)
 283       Streamer.stream("user", {user, cm_ref})
 284
 285       text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref})
 286
 287       assert text =~ "hey cirno"
 288       assert_receive {:text, ^text}
 289     end
 290
 291     test "it sends chat message notifications to the 'user:notification' stream", %{
 292       user: user,
 293       token: oauth_token
 294     } do
 295       other_user = insert(:user)
 296
 297       {:ok, create_activity} = CommonAPI.post_chat_message(other_user, user, "hey")
 298
 299       notify =
 300         Repo.get_by(Pleroma.Notification, user_id: user.id, activity_id: create_activity.id)
 301         |> Repo.preload(:activity)
 302
 303       Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 304       Streamer.stream("user:notification", notify)
 305
 306       assert_receive {:render_with_user, _, _, ^notify}
 307       refute Streamer.filtered_by_user?(user, notify)
 308     end
 309
 310     test "it doesn't send notify to the 'user:notification' stream when a user is blocked", %{
 311       user: user,
 312       token: oauth_token
 313     } do
 314       blocked = insert(:user)
 315       {:ok, _user_relationship} = User.block(user, blocked)
 316
 317       Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 318
 319       {:ok, activity} = CommonAPI.post(user, %{status: ":("})
 320       {:ok, _} = CommonAPI.favorite(blocked, activity.id)
 321
 322       refute_receive _
 323     end
 324
 325     test "it doesn't send notify to the 'user:notification' stream when a thread is muted", %{
 326       user: user,
 327       token: oauth_token
 328     } do
 329       user2 = insert(:user)
 330
 331       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
 332       {:ok, _} = CommonAPI.add_mute(user, activity)
 333
 334       Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 335
 336       {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
 337
 338       refute_receive _
 339       assert Streamer.filtered_by_user?(user, favorite_activity)
 340     end
 341
 342     test "it sends favorite to 'user:notification' stream'", %{
 343       user: user,
 344       token: oauth_token
 345     } do
 346       user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
 347
 348       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
 349       Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 350       {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
 351
 352       assert_receive {:render_with_user, _, "notification.json", notif}
 353       assert notif.activity.id == favorite_activity.id
 354       refute Streamer.filtered_by_user?(user, notif)
 355     end
 356
 357     test "it doesn't send the 'user:notification' stream' when a domain is blocked", %{
 358       user: user,
 359       token: oauth_token
 360     } do
 361       user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
 362
 363       {:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
 364       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
 365       Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 366       {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
 367
 368       refute_receive _
 369       assert Streamer.filtered_by_user?(user, favorite_activity)
 370     end
 371
 372     test "it sends follow activities to the 'user:notification' stream", %{
 373       user: user,
 374       token: oauth_token
 375     } do
 376       user_url = user.ap_id
 377       user2 = insert(:user)
 378
 379       body =
 380         File.read!("test/fixtures/users_mock/localhost.json")
 381         |> String.replace("{{nickname}}", user.nickname)
 382         |> Jason.encode!()
 383
 384       Tesla.Mock.mock_global(fn
 385         %{method: :get, url: ^user_url} ->
 386           %Tesla.Env{status: 200, body: body}
 387       end)
 388
 389       Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 390       {:ok, _follower, _followed, follow_activity} = CommonAPI.follow(user2, user)
 391
 392       assert_receive {:render_with_user, _, "notification.json", notif}
 393       assert notif.activity.id == follow_activity.id
 394       refute Streamer.filtered_by_user?(user, notif)
 395     end
 396   end
 397
 398   describe "public streams" do
 399     test "it sends to public (authenticated)" do
 400       %{user: user, token: oauth_token} = oauth_access(["read"])
 401       other_user = insert(:user)
 402
 403       Streamer.get_topic_and_add_socket("public", user, oauth_token)
 404
 405       {:ok, activity} = CommonAPI.post(other_user, %{status: "Test"})
 406       assert_receive {:render_with_user, _, _, ^activity}
 407       refute Streamer.filtered_by_user?(other_user, activity)
 408     end
 409
 410     test "it sends to public (unauthenticated)" do
 411       user = insert(:user)
 412
 413       Streamer.get_topic_and_add_socket("public", nil, nil)
 414
 415       {:ok, activity} = CommonAPI.post(user, %{status: "Test"})
 416       activity_id = activity.id
 417       assert_receive {:text, event}
 418       assert %{"event" => "update", "payload" => payload} = Jason.decode!(event)
 419       assert %{"id" => ^activity_id} = Jason.decode!(payload)
 420
 421       {:ok, _} = CommonAPI.delete(activity.id, user)
 422       assert_receive {:text, event}
 423       assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
 424     end
 425
 426     test "handles deletions" do
 427       %{user: user, token: oauth_token} = oauth_access(["read"])
 428       other_user = insert(:user)
 429       {:ok, activity} = CommonAPI.post(other_user, %{status: "Test"})
 430
 431       Streamer.get_topic_and_add_socket("public", user, oauth_token)
 432
 433       {:ok, _} = CommonAPI.delete(activity.id, other_user)
 434       activity_id = activity.id
 435       assert_receive {:text, event}
 436       assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
 437     end
 438   end
 439
 440   describe "thread_containment/2" do
 441     test "it filters to user if recipients invalid and thread containment is enabled" do
 442       Pleroma.Config.put([:instance, :skip_thread_containment], false)
 443       author = insert(:user)
 444       %{user: user, token: oauth_token} = oauth_access(["read"])
 445       User.follow(user, author, :follow_accept)
 446
 447       activity =
 448         insert(:note_activity,
 449           note:
 450             insert(:note,
 451               user: author,
 452               data: %{"to" => ["TEST-FFF"]}
 453             )
 454         )
 455
 456       Streamer.get_topic_and_add_socket("public", user, oauth_token)
 457       Streamer.stream("public", activity)
 458       assert_receive {:render_with_user, _, _, ^activity}
 459       assert Streamer.filtered_by_user?(user, activity)
 460     end
 461
 462     test "it sends message if recipients invalid and thread containment is disabled" do
 463       Pleroma.Config.put([:instance, :skip_thread_containment], true)
 464       author = insert(:user)
 465       %{user: user, token: oauth_token} = oauth_access(["read"])
 466       User.follow(user, author, :follow_accept)
 467
 468       activity =
 469         insert(:note_activity,
 470           note:
 471             insert(:note,
 472               user: author,
 473               data: %{"to" => ["TEST-FFF"]}
 474             )
 475         )
 476
 477       Streamer.get_topic_and_add_socket("public", user, oauth_token)
 478       Streamer.stream("public", activity)
 479
 480       assert_receive {:render_with_user, _, _, ^activity}
 481       refute Streamer.filtered_by_user?(user, activity)
 482     end
 483
 484     test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
 485       Pleroma.Config.put([:instance, :skip_thread_containment], false)
 486       author = insert(:user)
 487       user = insert(:user, skip_thread_containment: true)
 488       %{token: oauth_token} = oauth_access(["read"], user: user)
 489       User.follow(user, author, :follow_accept)
 490
 491       activity =
 492         insert(:note_activity,
 493           note:
 494             insert(:note,
 495               user: author,
 496               data: %{"to" => ["TEST-FFF"]}
 497             )
 498         )
 499
 500       Streamer.get_topic_and_add_socket("public", user, oauth_token)
 501       Streamer.stream("public", activity)
 502
 503       assert_receive {:render_with_user, _, _, ^activity}
 504       refute Streamer.filtered_by_user?(user, activity)
 505     end
 506   end
 507
 508   describe "blocks" do
 509     setup do: oauth_access(["read"])
 510
 511     test "it filters messages involving blocked users", %{user: user, token: oauth_token} do
 512       blocked_user = insert(:user)
 513       {:ok, _user_relationship} = User.block(user, blocked_user)
 514
 515       Streamer.get_topic_and_add_socket("public", user, oauth_token)
 516       {:ok, activity} = CommonAPI.post(blocked_user, %{status: "Test"})
 517       assert_receive {:render_with_user, _, _, ^activity}
 518       assert Streamer.filtered_by_user?(user, activity)
 519     end
 520
 521     test "it filters messages transitively involving blocked users", %{
 522       user: blocker,
 523       token: blocker_token
 524     } do
 525       blockee = insert(:user)
 526       friend = insert(:user)
 527
 528       Streamer.get_topic_and_add_socket("public", blocker, blocker_token)
 529
 530       {:ok, _user_relationship} = User.block(blocker, blockee)
 531
 532       {:ok, activity_one} = CommonAPI.post(friend, %{status: "hey! @#{blockee.nickname}"})
 533
 534       assert_receive {:render_with_user, _, _, ^activity_one}
 535       assert Streamer.filtered_by_user?(blocker, activity_one)
 536
 537       {:ok, activity_two} = CommonAPI.post(blockee, %{status: "hey! @#{friend.nickname}"})
 538
 539       assert_receive {:render_with_user, _, _, ^activity_two}
 540       assert Streamer.filtered_by_user?(blocker, activity_two)
 541
 542       {:ok, activity_three} = CommonAPI.post(blockee, %{status: "hey! @#{blocker.nickname}"})
 543
 544       assert_receive {:render_with_user, _, _, ^activity_three}
 545       assert Streamer.filtered_by_user?(blocker, activity_three)
 546     end
 547   end
 548
 549   describe "lists" do
 550     setup do: oauth_access(["read"])
 551
 552     test "it doesn't send unwanted DMs to list", %{user: user_a, token: user_a_token} do
 553       user_b = insert(:user)
 554       user_c = insert(:user)
 555
 556       {:ok, user_a} = User.follow(user_a, user_b)
 557
 558       {:ok, list} = List.create("Test", user_a)
 559       {:ok, list} = List.follow(list, user_b)
 560
 561       Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})
 562
 563       {:ok, _activity} =
 564         CommonAPI.post(user_b, %{
 565           status: "@#{user_c.nickname} Test",
 566           visibility: "direct"
 567         })
 568
 569       refute_receive _
 570     end
 571
 572     test "it doesn't send unwanted private posts to list", %{user: user_a, token: user_a_token} do
 573       user_b = insert(:user)
 574
 575       {:ok, list} = List.create("Test", user_a)
 576       {:ok, list} = List.follow(list, user_b)
 577
 578       Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})
 579
 580       {:ok, _activity} =
 581         CommonAPI.post(user_b, %{
 582           status: "Test",
 583           visibility: "private"
 584         })
 585
 586       refute_receive _
 587     end
 588
 589     test "it sends wanted private posts to list", %{user: user_a, token: user_a_token} do
 590       user_b = insert(:user)
 591
 592       {:ok, user_a} = User.follow(user_a, user_b)
 593
 594       {:ok, list} = List.create("Test", user_a)
 595       {:ok, list} = List.follow(list, user_b)
 596
 597       Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})
 598
 599       {:ok, activity} =
 600         CommonAPI.post(user_b, %{
 601           status: "Test",
 602           visibility: "private"
 603         })
 604
 605       assert_receive {:render_with_user, _, _, ^activity}
 606       refute Streamer.filtered_by_user?(user_a, activity)
 607     end
 608   end
 609
 610   describe "muted reblogs" do
 611     setup do: oauth_access(["read"])
 612
 613     test "it filters muted reblogs", %{user: user1, token: user1_token} do
 614       user2 = insert(:user)
 615       user3 = insert(:user)
 616       CommonAPI.follow(user1, user2)
 617       CommonAPI.hide_reblogs(user1, user2)
 618
 619       {:ok, create_activity} = CommonAPI.post(user3, %{status: "I'm kawen"})
 620
 621       Streamer.get_topic_and_add_socket("user", user1, user1_token)
 622       {:ok, announce_activity} = CommonAPI.repeat(create_activity.id, user2)
 623       assert_receive {:render_with_user, _, _, ^announce_activity}
 624       assert Streamer.filtered_by_user?(user1, announce_activity)
 625     end
 626
 627     test "it filters reblog notification for reblog-muted actors", %{
 628       user: user1,
 629       token: user1_token
 630     } do
 631       user2 = insert(:user)
 632       CommonAPI.follow(user1, user2)
 633       CommonAPI.hide_reblogs(user1, user2)
 634
 635       {:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
 636       Streamer.get_topic_and_add_socket("user", user1, user1_token)
 637       {:ok, _announce_activity} = CommonAPI.repeat(create_activity.id, user2)
 638
 639       assert_receive {:render_with_user, _, "notification.json", notif}
 640       assert Streamer.filtered_by_user?(user1, notif)
 641     end
 642
 643     test "it send non-reblog notification for reblog-muted actors", %{
 644       user: user1,
 645       token: user1_token
 646     } do
 647       user2 = insert(:user)
 648       CommonAPI.follow(user1, user2)
 649       CommonAPI.hide_reblogs(user1, user2)
 650
 651       {:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
 652       Streamer.get_topic_and_add_socket("user", user1, user1_token)
 653       {:ok, _favorite_activity} = CommonAPI.favorite(user2, create_activity.id)
 654
 655       assert_receive {:render_with_user, _, "notification.json", notif}
 656       refute Streamer.filtered_by_user?(user1, notif)
 657     end
 658   end
 659
 660   describe "muted threads" do
 661     test "it filters posts from muted threads" do
 662       user = insert(:user)
 663       %{user: user2, token: user2_token} = oauth_access(["read"])
 664       Streamer.get_topic_and_add_socket("user", user2, user2_token)
 665
 666       {:ok, user2, user, _activity} = CommonAPI.follow(user2, user)
 667       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
 668       {:ok, _} = CommonAPI.add_mute(user2, activity)
 669
 670       assert_receive {:render_with_user, _, _, ^activity}
 671       assert Streamer.filtered_by_user?(user2, activity)
 672     end
 673   end
 674
 675   describe "direct streams" do
 676     setup do: oauth_access(["read"])
 677
 678     test "it sends conversation update to the 'direct' stream", %{user: user, token: oauth_token} do
 679       another_user = insert(:user)
 680
 681       Streamer.get_topic_and_add_socket("direct", user, oauth_token)
 682
 683       {:ok, _create_activity} =
 684         CommonAPI.post(another_user, %{
 685           status: "hey @#{user.nickname}",
 686           visibility: "direct"
 687         })
 688
 689       assert_receive {:text, received_event}
 690
 691       assert %{"event" => "conversation", "payload" => received_payload} =
 692                Jason.decode!(received_event)
 693
 694       assert %{"last_status" => last_status} = Jason.decode!(received_payload)
 695       [participation] = Participation.for_user(user)
 696       assert last_status["pleroma"]["direct_conversation_id"] == participation.id
 697     end
 698
 699     test "it doesn't send conversation update to the 'direct' stream when the last message in the conversation is deleted",
 700          %{user: user, token: oauth_token} do
 701       another_user = insert(:user)
 702
 703       Streamer.get_topic_and_add_socket("direct", user, oauth_token)
 704
 705       {:ok, create_activity} =
 706         CommonAPI.post(another_user, %{
 707           status: "hi @#{user.nickname}",
 708           visibility: "direct"
 709         })
 710
 711       create_activity_id = create_activity.id
 712       assert_receive {:render_with_user, _, _, ^create_activity}
 713       assert_receive {:text, received_conversation1}
 714       assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
 715
 716       {:ok, _} = CommonAPI.delete(create_activity_id, another_user)
 717
 718       assert_receive {:text, received_event}
 719
 720       assert %{"event" => "delete", "payload" => ^create_activity_id} =
 721                Jason.decode!(received_event)
 722
 723       refute_receive _
 724     end
 725
 726     test "it sends conversation update to the 'direct' stream when a message is deleted", %{
 727       user: user,
 728       token: oauth_token
 729     } do
 730       another_user = insert(:user)
 731       Streamer.get_topic_and_add_socket("direct", user, oauth_token)
 732
 733       {:ok, create_activity} =
 734         CommonAPI.post(another_user, %{
 735           status: "hi @#{user.nickname}",
 736           visibility: "direct"
 737         })
 738
 739       {:ok, create_activity2} =
 740         CommonAPI.post(another_user, %{
 741           status: "hi @#{user.nickname} 2",
 742           in_reply_to_status_id: create_activity.id,
 743           visibility: "direct"
 744         })
 745
 746       assert_receive {:render_with_user, _, _, ^create_activity}
 747       assert_receive {:render_with_user, _, _, ^create_activity2}
 748       assert_receive {:text, received_conversation1}
 749       assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
 750       assert_receive {:text, received_conversation1}
 751       assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
 752
 753       {:ok, _} = CommonAPI.delete(create_activity2.id, another_user)
 754
 755       assert_receive {:text, received_event}
 756       assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
 757
 758       assert_receive {:text, received_event}
 759
 760       assert %{"event" => "conversation", "payload" => received_payload} =
 761                Jason.decode!(received_event)
 762
 763       assert %{"last_status" => last_status} = Jason.decode!(received_payload)
 764       assert last_status["id"] == to_string(create_activity.id)
 765     end
 766   end
 767 end