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