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} =
 259         CommonAPI.post_chat_message(other_user, user, "hey cirno", idempotency_key: "123")
 260
 261       object = Object.normalize(create_activity, false)
 262       chat = Chat.get(user.id, other_user.ap_id)
 263       cm_ref = MessageReference.for_chat_and_object(chat, object)
 264       cm_ref = %{cm_ref | chat: chat, object: object}
 265
 266       Streamer.get_topic_and_add_socket("user:pleroma_chat", user, oauth_token)
 267       Streamer.stream("user:pleroma_chat", {user, cm_ref})
 268
 269       text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref})
 270
 271       assert text =~ "hey cirno"
 272       assert_receive {:text, ^text}
 273     end
 274
 275     test "it sends chat messages to the 'user' stream", %{user: user, token: oauth_token} do
 276       other_user = insert(:user)
 277
 278       {:ok, create_activity} = CommonAPI.post_chat_message(other_user, user, "hey cirno")
 279       object = Object.normalize(create_activity, false)
 280       chat = Chat.get(user.id, other_user.ap_id)
 281       cm_ref = MessageReference.for_chat_and_object(chat, object)
 282       cm_ref = %{cm_ref | chat: chat, object: object}
 283
 284       Streamer.get_topic_and_add_socket("user", user, oauth_token)
 285       Streamer.stream("user", {user, cm_ref})
 286
 287       text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref})
 288
 289       assert text =~ "hey cirno"
 290       assert_receive {:text, ^text}
 291     end
 292
 293     test "it sends chat message notifications to the 'user:notification' stream", %{
 294       user: user,
 295       token: oauth_token
 296     } do
 297       other_user = insert(:user)
 298
 299       {:ok, create_activity} = CommonAPI.post_chat_message(other_user, user, "hey")
 300
 301       notify =
 302         Repo.get_by(Pleroma.Notification, user_id: user.id, activity_id: create_activity.id)
 303         |> Repo.preload(:activity)
 304
 305       Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 306       Streamer.stream("user:notification", notify)
 307
 308       assert_receive {:render_with_user, _, _, ^notify}
 309       refute Streamer.filtered_by_user?(user, notify)
 310     end
 311
 312     test "it doesn't send notify to the 'user:notification' stream when a user is blocked", %{
 313       user: user,
 314       token: oauth_token
 315     } do
 316       blocked = insert(:user)
 317       {:ok, _user_relationship} = User.block(user, blocked)
 318
 319       Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 320
 321       {:ok, activity} = CommonAPI.post(user, %{status: ":("})
 322       {:ok, _} = CommonAPI.favorite(blocked, activity.id)
 323
 324       refute_receive _
 325     end
 326
 327     test "it doesn't send notify to the 'user:notification' stream when a thread is muted", %{
 328       user: user,
 329       token: oauth_token
 330     } do
 331       user2 = insert(:user)
 332
 333       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
 334       {:ok, _} = CommonAPI.add_mute(user, activity)
 335
 336       Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 337
 338       {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
 339
 340       refute_receive _
 341       assert Streamer.filtered_by_user?(user, favorite_activity)
 342     end
 343
 344     test "it sends favorite to 'user:notification' stream'", %{
 345       user: user,
 346       token: oauth_token
 347     } do
 348       user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
 349
 350       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
 351       Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 352       {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
 353
 354       assert_receive {:render_with_user, _, "notification.json", notif}
 355       assert notif.activity.id == favorite_activity.id
 356       refute Streamer.filtered_by_user?(user, notif)
 357     end
 358
 359     test "it doesn't send the 'user:notification' stream' when a domain is blocked", %{
 360       user: user,
 361       token: oauth_token
 362     } do
 363       user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
 364
 365       {:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
 366       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
 367       Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 368       {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
 369
 370       refute_receive _
 371       assert Streamer.filtered_by_user?(user, favorite_activity)
 372     end
 373
 374     test "it sends follow activities to the 'user:notification' stream", %{
 375       user: user,
 376       token: oauth_token
 377     } do
 378       user_url = user.ap_id
 379       user2 = insert(:user)
 380
 381       body =
 382         File.read!("test/fixtures/users_mock/localhost.json")
 383         |> String.replace("{{nickname}}", user.nickname)
 384         |> Jason.encode!()
 385
 386       Tesla.Mock.mock_global(fn
 387         %{method: :get, url: ^user_url} ->
 388           %Tesla.Env{status: 200, body: body}
 389       end)
 390
 391       Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
 392       {:ok, _follower, _followed, follow_activity} = CommonAPI.follow(user2, user)
 393
 394       assert_receive {:render_with_user, _, "notification.json", notif}
 395       assert notif.activity.id == follow_activity.id
 396       refute Streamer.filtered_by_user?(user, notif)
 397     end
 398   end
 399
 400   describe "public streams" do
 401     test "it sends to public (authenticated)" do
 402       %{user: user, token: oauth_token} = oauth_access(["read"])
 403       other_user = insert(:user)
 404
 405       Streamer.get_topic_and_add_socket("public", user, oauth_token)
 406
 407       {:ok, activity} = CommonAPI.post(other_user, %{status: "Test"})
 408       assert_receive {:render_with_user, _, _, ^activity}
 409       refute Streamer.filtered_by_user?(other_user, activity)
 410     end
 411
 412     test "it sends to public (unauthenticated)" do
 413       user = insert(:user)
 414
 415       Streamer.get_topic_and_add_socket("public", nil, nil)
 416
 417       {:ok, activity} = CommonAPI.post(user, %{status: "Test"})
 418       activity_id = activity.id
 419       assert_receive {:text, event}
 420       assert %{"event" => "update", "payload" => payload} = Jason.decode!(event)
 421       assert %{"id" => ^activity_id} = Jason.decode!(payload)
 422
 423       {:ok, _} = CommonAPI.delete(activity.id, user)
 424       assert_receive {:text, event}
 425       assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
 426     end
 427
 428     test "handles deletions" do
 429       %{user: user, token: oauth_token} = oauth_access(["read"])
 430       other_user = insert(:user)
 431       {:ok, activity} = CommonAPI.post(other_user, %{status: "Test"})
 432
 433       Streamer.get_topic_and_add_socket("public", user, oauth_token)
 434
 435       {:ok, _} = CommonAPI.delete(activity.id, other_user)
 436       activity_id = activity.id
 437       assert_receive {:text, event}
 438       assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
 439     end
 440   end
 441
 442   describe "thread_containment/2" do
 443     test "it filters to user if recipients invalid and thread containment is enabled" do
 444       Pleroma.Config.put([:instance, :skip_thread_containment], false)
 445       author = insert(:user)
 446       %{user: user, token: oauth_token} = oauth_access(["read"])
 447       User.follow(user, author, :follow_accept)
 448
 449       activity =
 450         insert(:note_activity,
 451           note:
 452             insert(:note,
 453               user: author,
 454               data: %{"to" => ["TEST-FFF"]}
 455             )
 456         )
 457
 458       Streamer.get_topic_and_add_socket("public", user, oauth_token)
 459       Streamer.stream("public", activity)
 460       assert_receive {:render_with_user, _, _, ^activity}
 461       assert Streamer.filtered_by_user?(user, activity)
 462     end
 463
 464     test "it sends message if recipients invalid and thread containment is disabled" do
 465       Pleroma.Config.put([:instance, :skip_thread_containment], true)
 466       author = insert(:user)
 467       %{user: user, token: oauth_token} = oauth_access(["read"])
 468       User.follow(user, author, :follow_accept)
 469
 470       activity =
 471         insert(:note_activity,
 472           note:
 473             insert(:note,
 474               user: author,
 475               data: %{"to" => ["TEST-FFF"]}
 476             )
 477         )
 478
 479       Streamer.get_topic_and_add_socket("public", user, oauth_token)
 480       Streamer.stream("public", activity)
 481
 482       assert_receive {:render_with_user, _, _, ^activity}
 483       refute Streamer.filtered_by_user?(user, activity)
 484     end
 485
 486     test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
 487       Pleroma.Config.put([:instance, :skip_thread_containment], false)
 488       author = insert(:user)
 489       user = insert(:user, skip_thread_containment: true)
 490       %{token: oauth_token} = oauth_access(["read"], user: user)
 491       User.follow(user, author, :follow_accept)
 492
 493       activity =
 494         insert(:note_activity,
 495           note:
 496             insert(:note,
 497               user: author,
 498               data: %{"to" => ["TEST-FFF"]}
 499             )
 500         )
 501
 502       Streamer.get_topic_and_add_socket("public", user, oauth_token)
 503       Streamer.stream("public", activity)
 504
 505       assert_receive {:render_with_user, _, _, ^activity}
 506       refute Streamer.filtered_by_user?(user, activity)
 507     end
 508   end
 509
 510   describe "blocks" do
 511     setup do: oauth_access(["read"])
 512
 513     test "it filters messages involving blocked users", %{user: user, token: oauth_token} do
 514       blocked_user = insert(:user)
 515       {:ok, _user_relationship} = User.block(user, blocked_user)
 516
 517       Streamer.get_topic_and_add_socket("public", user, oauth_token)
 518       {:ok, activity} = CommonAPI.post(blocked_user, %{status: "Test"})
 519       assert_receive {:render_with_user, _, _, ^activity}
 520       assert Streamer.filtered_by_user?(user, activity)
 521     end
 522
 523     test "it filters messages transitively involving blocked users", %{
 524       user: blocker,
 525       token: blocker_token
 526     } do
 527       blockee = insert(:user)
 528       friend = insert(:user)
 529
 530       Streamer.get_topic_and_add_socket("public", blocker, blocker_token)
 531
 532       {:ok, _user_relationship} = User.block(blocker, blockee)
 533
 534       {:ok, activity_one} = CommonAPI.post(friend, %{status: "hey! @#{blockee.nickname}"})
 535
 536       assert_receive {:render_with_user, _, _, ^activity_one}
 537       assert Streamer.filtered_by_user?(blocker, activity_one)
 538
 539       {:ok, activity_two} = CommonAPI.post(blockee, %{status: "hey! @#{friend.nickname}"})
 540
 541       assert_receive {:render_with_user, _, _, ^activity_two}
 542       assert Streamer.filtered_by_user?(blocker, activity_two)
 543
 544       {:ok, activity_three} = CommonAPI.post(blockee, %{status: "hey! @#{blocker.nickname}"})
 545
 546       assert_receive {:render_with_user, _, _, ^activity_three}
 547       assert Streamer.filtered_by_user?(blocker, activity_three)
 548     end
 549   end
 550
 551   describe "lists" do
 552     setup do: oauth_access(["read"])
 553
 554     test "it doesn't send unwanted DMs to list", %{user: user_a, token: user_a_token} do
 555       user_b = insert(:user)
 556       user_c = insert(:user)
 557
 558       {:ok, user_a} = User.follow(user_a, user_b)
 559
 560       {:ok, list} = List.create("Test", user_a)
 561       {:ok, list} = List.follow(list, user_b)
 562
 563       Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})
 564
 565       {:ok, _activity} =
 566         CommonAPI.post(user_b, %{
 567           status: "@#{user_c.nickname} Test",
 568           visibility: "direct"
 569         })
 570
 571       refute_receive _
 572     end
 573
 574     test "it doesn't send unwanted private posts to list", %{user: user_a, token: user_a_token} do
 575       user_b = insert(:user)
 576
 577       {:ok, list} = List.create("Test", user_a)
 578       {:ok, list} = List.follow(list, user_b)
 579
 580       Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})
 581
 582       {:ok, _activity} =
 583         CommonAPI.post(user_b, %{
 584           status: "Test",
 585           visibility: "private"
 586         })
 587
 588       refute_receive _
 589     end
 590
 591     test "it sends wanted private posts to list", %{user: user_a, token: user_a_token} do
 592       user_b = insert(:user)
 593
 594       {:ok, user_a} = User.follow(user_a, user_b)
 595
 596       {:ok, list} = List.create("Test", user_a)
 597       {:ok, list} = List.follow(list, user_b)
 598
 599       Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})
 600
 601       {:ok, activity} =
 602         CommonAPI.post(user_b, %{
 603           status: "Test",
 604           visibility: "private"
 605         })
 606
 607       assert_receive {:render_with_user, _, _, ^activity}
 608       refute Streamer.filtered_by_user?(user_a, activity)
 609     end
 610   end
 611
 612   describe "muted reblogs" do
 613     setup do: oauth_access(["read"])
 614
 615     test "it filters muted reblogs", %{user: user1, token: user1_token} do
 616       user2 = insert(:user)
 617       user3 = insert(:user)
 618       CommonAPI.follow(user1, user2)
 619       CommonAPI.hide_reblogs(user1, user2)
 620
 621       {:ok, create_activity} = CommonAPI.post(user3, %{status: "I'm kawen"})
 622
 623       Streamer.get_topic_and_add_socket("user", user1, user1_token)
 624       {:ok, announce_activity} = CommonAPI.repeat(create_activity.id, user2)
 625       assert_receive {:render_with_user, _, _, ^announce_activity}
 626       assert Streamer.filtered_by_user?(user1, announce_activity)
 627     end
 628
 629     test "it filters reblog notification for reblog-muted actors", %{
 630       user: user1,
 631       token: user1_token
 632     } do
 633       user2 = insert(:user)
 634       CommonAPI.follow(user1, user2)
 635       CommonAPI.hide_reblogs(user1, user2)
 636
 637       {:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
 638       Streamer.get_topic_and_add_socket("user", user1, user1_token)
 639       {:ok, _announce_activity} = CommonAPI.repeat(create_activity.id, user2)
 640
 641       assert_receive {:render_with_user, _, "notification.json", notif}
 642       assert Streamer.filtered_by_user?(user1, notif)
 643     end
 644
 645     test "it send non-reblog notification for reblog-muted actors", %{
 646       user: user1,
 647       token: user1_token
 648     } do
 649       user2 = insert(:user)
 650       CommonAPI.follow(user1, user2)
 651       CommonAPI.hide_reblogs(user1, user2)
 652
 653       {:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
 654       Streamer.get_topic_and_add_socket("user", user1, user1_token)
 655       {:ok, _favorite_activity} = CommonAPI.favorite(user2, create_activity.id)
 656
 657       assert_receive {:render_with_user, _, "notification.json", notif}
 658       refute Streamer.filtered_by_user?(user1, notif)
 659     end
 660   end
 661
 662   describe "muted threads" do
 663     test "it filters posts from muted threads" do
 664       user = insert(:user)
 665       %{user: user2, token: user2_token} = oauth_access(["read"])
 666       Streamer.get_topic_and_add_socket("user", user2, user2_token)
 667
 668       {:ok, user2, user, _activity} = CommonAPI.follow(user2, user)
 669       {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
 670       {:ok, _} = CommonAPI.add_mute(user2, activity)
 671
 672       assert_receive {:render_with_user, _, _, ^activity}
 673       assert Streamer.filtered_by_user?(user2, activity)
 674     end
 675   end
 676
 677   describe "direct streams" do
 678     setup do: oauth_access(["read"])
 679
 680     test "it sends conversation update to the 'direct' stream", %{user: user, token: oauth_token} do
 681       another_user = insert(:user)
 682
 683       Streamer.get_topic_and_add_socket("direct", user, oauth_token)
 684
 685       {:ok, _create_activity} =
 686         CommonAPI.post(another_user, %{
 687           status: "hey @#{user.nickname}",
 688           visibility: "direct"
 689         })
 690
 691       assert_receive {:text, received_event}
 692
 693       assert %{"event" => "conversation", "payload" => received_payload} =
 694                Jason.decode!(received_event)
 695
 696       assert %{"last_status" => last_status} = Jason.decode!(received_payload)
 697       [participation] = Participation.for_user(user)
 698       assert last_status["pleroma"]["direct_conversation_id"] == participation.id
 699     end
 700
 701     test "it doesn't send conversation update to the 'direct' stream when the last message in the conversation is deleted",
 702          %{user: user, token: oauth_token} do
 703       another_user = insert(:user)
 704
 705       Streamer.get_topic_and_add_socket("direct", user, oauth_token)
 706
 707       {:ok, create_activity} =
 708         CommonAPI.post(another_user, %{
 709           status: "hi @#{user.nickname}",
 710           visibility: "direct"
 711         })
 712
 713       create_activity_id = create_activity.id
 714       assert_receive {:render_with_user, _, _, ^create_activity}
 715       assert_receive {:text, received_conversation1}
 716       assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
 717
 718       {:ok, _} = CommonAPI.delete(create_activity_id, another_user)
 719
 720       assert_receive {:text, received_event}
 721
 722       assert %{"event" => "delete", "payload" => ^create_activity_id} =
 723                Jason.decode!(received_event)
 724
 725       refute_receive _
 726     end
 727
 728     test "it sends conversation update to the 'direct' stream when a message is deleted", %{
 729       user: user,
 730       token: oauth_token
 731     } do
 732       another_user = insert(:user)
 733       Streamer.get_topic_and_add_socket("direct", user, oauth_token)
 734
 735       {:ok, create_activity} =
 736         CommonAPI.post(another_user, %{
 737           status: "hi @#{user.nickname}",
 738           visibility: "direct"
 739         })
 740
 741       {:ok, create_activity2} =
 742         CommonAPI.post(another_user, %{
 743           status: "hi @#{user.nickname} 2",
 744           in_reply_to_status_id: create_activity.id,
 745           visibility: "direct"
 746         })
 747
 748       assert_receive {:render_with_user, _, _, ^create_activity}
 749       assert_receive {:render_with_user, _, _, ^create_activity2}
 750       assert_receive {:text, received_conversation1}
 751       assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
 752       assert_receive {:text, received_conversation1}
 753       assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
 754
 755       {:ok, _} = CommonAPI.delete(create_activity2.id, another_user)
 756
 757       assert_receive {:text, received_event}
 758       assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
 759
 760       assert_receive {:text, received_event}
 761
 762       assert %{"event" => "conversation", "payload" => received_payload} =
 763                Jason.decode!(received_event)
 764
 765       assert %{"last_status" => last_status} = Jason.decode!(received_payload)
 766       assert last_status["id"] == to_string(create_activity.id)
 767     end
 768   end
 769 end