1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
5 defmodule Pleroma.Web.StreamerTest do
10 alias Pleroma.Conversation.Participation
13 alias Pleroma.Web.CommonAPI
14 alias Pleroma.Web.Streamer
15 alias Pleroma.Web.Streamer.StreamerSocket
16 alias Pleroma.Web.Streamer.Worker
18 @moduletag needs_streamer: true, capture_log: true
21 @streamer_start_wait 10
23 clear_config_all([:instance, :skip_thread_containment])
25 describe "user streams" do
28 notify = insert(:notification, user: user, activity: build(:note_activity))
29 {:ok, %{user: user, notify: notify}}
32 test "it sends notify to in the 'user' stream", %{user: user, notify: notify} do
35 assert_receive {:text, _}, @streamer_timeout
40 %{transport_pid: task.pid, assigns: %{user: user}}
43 Streamer.stream("user", notify)
47 test "it sends notify to in the 'user:notification' stream", %{user: user, notify: notify} do
50 assert_receive {:text, _}, @streamer_timeout
55 %{transport_pid: task.pid, assigns: %{user: user}}
58 Streamer.stream("user:notification", notify)
62 test "it doesn't send notify to the 'user:notification' stream when a user is blocked", %{
65 blocked = insert(:user)
66 {:ok, _user_relationship} = User.block(user, blocked)
68 task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
72 %{transport_pid: task.pid, assigns: %{user: user}}
75 {:ok, activity} = CommonAPI.post(user, %{"status" => ":("})
76 {:ok, notif, _} = CommonAPI.favorite(activity.id, blocked)
78 Streamer.stream("user:notification", notif)
82 test "it doesn't send notify to the 'user:notification' stream when a thread is muted", %{
86 task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
90 %{transport_pid: task.pid, assigns: %{user: user}}
93 {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
94 {:ok, activity} = CommonAPI.add_mute(user, activity)
95 {:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
96 Streamer.stream("user:notification", notif)
100 test "it doesn't send notify to the 'user:notification' stream' when a domain is blocked", %{
103 user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
104 task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
108 %{transport_pid: task.pid, assigns: %{user: user}}
111 {:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
112 {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
113 {:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
115 Streamer.stream("user:notification", notif)
119 test "it sends follow activities to the 'user:notification' stream", %{
122 user2 = insert(:user)
123 task = Task.async(fn -> assert_receive {:text, _}, @streamer_timeout end)
125 Process.sleep(@streamer_start_wait)
129 %{transport_pid: task.pid, assigns: %{user: user}}
132 {:ok, _follower, _followed, _activity} = CommonAPI.follow(user2, user)
134 # We don't directly pipe the notification to the streamer as it's already
135 # generated as a side effect of CommonAPI.follow().
140 test "it sends to public" do
142 other_user = insert(:user)
146 assert_receive {:text, _}, @streamer_timeout
149 fake_socket = %StreamerSocket{
150 transport_pid: task.pid,
154 {:ok, activity} = CommonAPI.post(other_user, %{"status" => "Test"})
157 "public" => [fake_socket]
160 Worker.push_to_socket(topics, "public", activity)
169 "payload" => activity.id
173 assert_receive {:text, received_event}, @streamer_timeout
174 assert received_event == expected_event
177 fake_socket = %StreamerSocket{
178 transport_pid: task.pid,
182 {:ok, activity} = CommonAPI.delete(activity.id, other_user)
185 "public" => [fake_socket]
188 Worker.push_to_socket(topics, "public", activity)
193 describe "thread_containment" do
194 test "it doesn't send to user if recipients invalid and thread containment is enabled" do
195 Pleroma.Config.put([:instance, :skip_thread_containment], false)
196 author = insert(:user)
198 User.follow(user, author, "accept")
201 insert(:note_activity,
205 data: %{"to" => ["TEST-FFF"]}
209 task = Task.async(fn -> refute_receive {:text, _}, 1_000 end)
210 fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
211 topics = %{"public" => [fake_socket]}
212 Worker.push_to_socket(topics, "public", activity)
217 test "it sends message if recipients invalid and thread containment is disabled" do
218 Pleroma.Config.put([:instance, :skip_thread_containment], true)
219 author = insert(:user)
221 User.follow(user, author, "accept")
224 insert(:note_activity,
228 data: %{"to" => ["TEST-FFF"]}
232 task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
233 fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
234 topics = %{"public" => [fake_socket]}
235 Worker.push_to_socket(topics, "public", activity)
240 test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
241 Pleroma.Config.put([:instance, :skip_thread_containment], false)
242 author = insert(:user)
243 user = insert(:user, skip_thread_containment: true)
244 User.follow(user, author, "accept")
247 insert(:note_activity,
251 data: %{"to" => ["TEST-FFF"]}
255 task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
256 fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
257 topics = %{"public" => [fake_socket]}
258 Worker.push_to_socket(topics, "public", activity)
265 test "it doesn't send messages involving blocked users" do
267 blocked_user = insert(:user)
268 {:ok, _user_relationship} = User.block(user, blocked_user)
272 refute_receive {:text, _}, 1_000
275 fake_socket = %StreamerSocket{
276 transport_pid: task.pid,
280 {:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})
283 "public" => [fake_socket]
286 Worker.push_to_socket(topics, "public", activity)
291 test "it doesn't send messages transitively involving blocked users" do
292 blocker = insert(:user)
293 blockee = insert(:user)
294 friend = insert(:user)
298 refute_receive {:text, _}, 1_000
301 fake_socket = %StreamerSocket{
302 transport_pid: task.pid,
307 "public" => [fake_socket]
310 {:ok, _user_relationship} = User.block(blocker, blockee)
312 {:ok, activity_one} = CommonAPI.post(friend, %{"status" => "hey! @#{blockee.nickname}"})
314 Worker.push_to_socket(topics, "public", activity_one)
316 {:ok, activity_two} = CommonAPI.post(blockee, %{"status" => "hey! @#{friend.nickname}"})
318 Worker.push_to_socket(topics, "public", activity_two)
320 {:ok, activity_three} = CommonAPI.post(blockee, %{"status" => "hey! @#{blocker.nickname}"})
322 Worker.push_to_socket(topics, "public", activity_three)
328 test "it doesn't send unwanted DMs to list" do
329 user_a = insert(:user)
330 user_b = insert(:user)
331 user_c = insert(:user)
333 {:ok, user_a} = User.follow(user_a, user_b)
335 {:ok, list} = List.create("Test", user_a)
336 {:ok, list} = List.follow(list, user_b)
340 refute_receive {:text, _}, 1_000
343 fake_socket = %StreamerSocket{
344 transport_pid: task.pid,
349 CommonAPI.post(user_b, %{
350 "status" => "@#{user_c.nickname} Test",
351 "visibility" => "direct"
355 "list:#{list.id}" => [fake_socket]
358 Worker.handle_call({:stream, "list", activity}, self(), topics)
363 test "it doesn't send unwanted private posts to list" do
364 user_a = insert(:user)
365 user_b = insert(:user)
367 {:ok, list} = List.create("Test", user_a)
368 {:ok, list} = List.follow(list, user_b)
372 refute_receive {:text, _}, 1_000
375 fake_socket = %StreamerSocket{
376 transport_pid: task.pid,
381 CommonAPI.post(user_b, %{
383 "visibility" => "private"
387 "list:#{list.id}" => [fake_socket]
390 Worker.handle_call({:stream, "list", activity}, self(), topics)
395 test "it sends wanted private posts to list" do
396 user_a = insert(:user)
397 user_b = insert(:user)
399 {:ok, user_a} = User.follow(user_a, user_b)
401 {:ok, list} = List.create("Test", user_a)
402 {:ok, list} = List.follow(list, user_b)
406 assert_receive {:text, _}, 1_000
409 fake_socket = %StreamerSocket{
410 transport_pid: task.pid,
415 CommonAPI.post(user_b, %{
417 "visibility" => "private"
425 Worker.handle_call({:stream, "list", activity}, self(), %{})
430 test "it doesn't send muted reblogs" do
431 user1 = insert(:user)
432 user2 = insert(:user)
433 user3 = insert(:user)
434 CommonAPI.hide_reblogs(user1, user2)
438 refute_receive {:text, _}, 1_000
441 fake_socket = %StreamerSocket{
442 transport_pid: task.pid,
446 {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
447 {:ok, announce_activity, _} = CommonAPI.repeat(create_activity.id, user2)
450 "public" => [fake_socket]
453 Worker.push_to_socket(topics, "public", announce_activity)
458 test "it doesn't send posts from muted threads" do
460 user2 = insert(:user)
461 {:ok, user2, user, _activity} = CommonAPI.follow(user2, user)
463 {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
465 {:ok, activity} = CommonAPI.add_mute(user2, activity)
467 task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
471 %{transport_pid: task.pid, assigns: %{user: user2}}
474 Streamer.stream("user", activity)
478 describe "direct streams" do
483 test "it sends conversation update to the 'direct' stream", %{} do
485 another_user = insert(:user)
489 assert_receive {:text, received_event}, @streamer_timeout
491 assert %{"event" => "conversation", "payload" => received_payload} =
492 Jason.decode!(received_event)
494 assert %{"last_status" => last_status} = Jason.decode!(received_payload)
495 [participation] = Participation.for_user(user)
496 assert last_status["pleroma"]["direct_conversation_id"] == participation.id
501 %{transport_pid: task.pid, assigns: %{user: user}}
504 {:ok, _create_activity} =
505 CommonAPI.post(another_user, %{
506 "status" => "hey @#{user.nickname}",
507 "visibility" => "direct"
513 test "it doesn't send conversation update to the 'direct' stream when the last message in the conversation is deleted" do
515 another_user = insert(:user)
517 {:ok, create_activity} =
518 CommonAPI.post(another_user, %{
519 "status" => "hi @#{user.nickname}",
520 "visibility" => "direct"
525 assert_receive {:text, received_event}, @streamer_timeout
526 assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
528 refute_receive {:text, _}, @streamer_timeout
531 Process.sleep(@streamer_start_wait)
535 %{transport_pid: task.pid, assigns: %{user: user}}
538 {:ok, _} = CommonAPI.delete(create_activity.id, another_user)
543 test "it sends conversation update to the 'direct' stream when a message is deleted" do
545 another_user = insert(:user)
547 {:ok, create_activity} =
548 CommonAPI.post(another_user, %{
549 "status" => "hi @#{user.nickname}",
550 "visibility" => "direct"
553 {:ok, create_activity2} =
554 CommonAPI.post(another_user, %{
555 "status" => "hi @#{user.nickname}",
556 "in_reply_to_status_id" => create_activity.id,
557 "visibility" => "direct"
562 assert_receive {:text, received_event}, @streamer_timeout
563 assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
565 assert_receive {:text, received_event}, @streamer_timeout
567 assert %{"event" => "conversation", "payload" => received_payload} =
568 Jason.decode!(received_event)
570 assert %{"last_status" => last_status} = Jason.decode!(received_payload)
571 assert last_status["id"] == to_string(create_activity.id)
574 Process.sleep(@streamer_start_wait)
578 %{transport_pid: task.pid, assigns: %{user: user}}
581 {:ok, _} = CommonAPI.delete(create_activity2.id, another_user)