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