X-Git-Url: http://git.squeep.com/?a=blobdiff_plain;f=test%2Fweb%2Fstreamer%2Fstreamer_test.exs;h=339f99bbf8b39f2dc2b13fc8971bac19995e6a77;hb=509c81e4b10bd8ba6d3a93889cd6fbbbfbcbab21;hp=b8fcd41fa31abb73588162f504c2d54125fbad61;hpb=447514dfa2759e3415399412e82bf772ff119e04;p=akkoma
diff --git a/test/web/streamer/streamer_test.exs b/test/web/streamer/streamer_test.exs
index b8fcd41fa..339f99bbf 100644
--- a/test/web/streamer/streamer_test.exs
+++ b/test/web/streamer/streamer_test.exs
@@ -1,5 +1,5 @@
# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors
+# Copyright © 2017-2020 Pleroma Authors
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.StreamerTest do
@@ -7,6 +7,7 @@ defmodule Pleroma.Web.StreamerTest do
import Pleroma.Factory
+ alias Pleroma.Conversation.Participation
alias Pleroma.List
alias Pleroma.User
alias Pleroma.Web.CommonAPI
@@ -14,8 +15,12 @@ defmodule Pleroma.Web.StreamerTest do
alias Pleroma.Web.Streamer.StreamerSocket
alias Pleroma.Web.Streamer.Worker
- @moduletag needs_streamer: true
- clear_config_all([:instance, :skip_thread_containment])
+ @moduletag needs_streamer: true, capture_log: true
+
+ @streamer_timeout 150
+ @streamer_start_wait 10
+
+ clear_config([:instance, :skip_thread_containment])
describe "user streams" do
setup do
@@ -27,7 +32,7 @@ defmodule Pleroma.Web.StreamerTest do
test "it sends notify to in the 'user' stream", %{user: user, notify: notify} do
task =
Task.async(fn ->
- assert_receive {:text, _}, 4_000
+ assert_receive {:text, _}, @streamer_timeout
end)
Streamer.add_socket(
@@ -42,7 +47,7 @@ defmodule Pleroma.Web.StreamerTest do
test "it sends notify to in the 'user:notification' stream", %{user: user, notify: notify} do
task =
Task.async(fn ->
- assert_receive {:text, _}, 4_000
+ assert_receive {:text, _}, @streamer_timeout
end)
Streamer.add_socket(
@@ -58,18 +63,18 @@ defmodule Pleroma.Web.StreamerTest do
user: user
} do
blocked = insert(:user)
- {:ok, user} = User.block(user, blocked)
+ {:ok, _user_relationship} = User.block(user, blocked)
+
+ {:ok, activity} = CommonAPI.post(user, %{"status" => ":("})
+ {:ok, notif, _} = CommonAPI.favorite(activity.id, blocked)
- task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)
+ task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
Streamer.add_socket(
"user:notification",
%{transport_pid: task.pid, assigns: %{user: user}}
)
- {:ok, activity} = CommonAPI.post(user, %{"status" => ":("})
- {:ok, notif, _} = CommonAPI.favorite(activity.id, blocked)
-
Streamer.stream("user:notification", notif)
Task.await(task)
end
@@ -78,16 +83,18 @@ defmodule Pleroma.Web.StreamerTest do
user: user
} do
user2 = insert(:user)
- task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)
+
+ {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
+ {:ok, activity} = CommonAPI.add_mute(user, activity)
+ {:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
+
+ task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
Streamer.add_socket(
"user:notification",
%{transport_pid: task.pid, assigns: %{user: user}}
)
- {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
- {:ok, activity} = CommonAPI.add_mute(user, activity)
- {:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
Streamer.stream("user:notification", notif)
Task.await(task)
end
@@ -96,20 +103,41 @@ defmodule Pleroma.Web.StreamerTest do
user: user
} do
user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
- task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)
+
+ {:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
+ {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
+ {:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
+
+ task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
Streamer.add_socket(
"user:notification",
%{transport_pid: task.pid, assigns: %{user: user}}
)
- {:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
- {:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})
- {:ok, notif, _} = CommonAPI.favorite(activity.id, user2)
-
Streamer.stream("user:notification", notif)
Task.await(task)
end
+
+ test "it sends follow activities to the 'user:notification' stream", %{
+ user: user
+ } do
+ user2 = insert(:user)
+ task = Task.async(fn -> assert_receive {:text, _}, @streamer_timeout end)
+
+ Process.sleep(@streamer_start_wait)
+
+ Streamer.add_socket(
+ "user:notification",
+ %{transport_pid: task.pid, assigns: %{user: user}}
+ )
+
+ {:ok, _follower, _followed, _activity} = CommonAPI.follow(user2, user)
+
+ # We don't directly pipe the notification to the streamer as it's already
+ # generated as a side effect of CommonAPI.follow().
+ Task.await(task)
+ end
end
test "it sends to public" do
@@ -118,7 +146,7 @@ defmodule Pleroma.Web.StreamerTest do
task =
Task.async(fn ->
- assert_receive {:text, _}, 4_000
+ assert_receive {:text, _}, @streamer_timeout
end)
fake_socket = %StreamerSocket{
@@ -145,7 +173,7 @@ defmodule Pleroma.Web.StreamerTest do
}
|> Jason.encode!()
- assert_receive {:text, received_event}, 4_000
+ assert_receive {:text, received_event}, @streamer_timeout
assert received_event == expected_event
end)
@@ -169,7 +197,8 @@ defmodule Pleroma.Web.StreamerTest do
test "it doesn't send to user if recipients invalid and thread containment is enabled" do
Pleroma.Config.put([:instance, :skip_thread_containment], false)
author = insert(:user)
- user = insert(:user, following: [author.ap_id])
+ user = insert(:user)
+ User.follow(user, author, "accept")
activity =
insert(:note_activity,
@@ -191,7 +220,8 @@ defmodule Pleroma.Web.StreamerTest do
test "it sends message if recipients invalid and thread containment is disabled" do
Pleroma.Config.put([:instance, :skip_thread_containment], true)
author = insert(:user)
- user = insert(:user, following: [author.ap_id])
+ user = insert(:user)
+ User.follow(user, author, "accept")
activity =
insert(:note_activity,
@@ -213,7 +243,8 @@ defmodule Pleroma.Web.StreamerTest do
test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
Pleroma.Config.put([:instance, :skip_thread_containment], false)
author = insert(:user)
- user = insert(:user, following: [author.ap_id], info: %{skip_thread_containment: true})
+ user = insert(:user, skip_thread_containment: true)
+ User.follow(user, author, "accept")
activity =
insert(:note_activity,
@@ -233,30 +264,68 @@ defmodule Pleroma.Web.StreamerTest do
end
end
- test "it doesn't send to blocked users" do
- user = insert(:user)
- blocked_user = insert(:user)
- {:ok, user} = User.block(user, blocked_user)
+ describe "blocks" do
+ test "it doesn't send messages involving blocked users" do
+ user = insert(:user)
+ blocked_user = insert(:user)
+ {:ok, _user_relationship} = User.block(user, blocked_user)
- task =
- Task.async(fn ->
- refute_receive {:text, _}, 1_000
- end)
+ {:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})
- fake_socket = %StreamerSocket{
- transport_pid: task.pid,
- user: user
- }
+ task =
+ Task.async(fn ->
+ refute_receive {:text, _}, 1_000
+ end)
- {:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})
+ fake_socket = %StreamerSocket{
+ transport_pid: task.pid,
+ user: user
+ }
- topics = %{
- "public" => [fake_socket]
- }
+ topics = %{
+ "public" => [fake_socket]
+ }
- Worker.push_to_socket(topics, "public", activity)
+ Worker.push_to_socket(topics, "public", activity)
- Task.await(task)
+ Task.await(task)
+ end
+
+ test "it doesn't send messages transitively involving blocked users" do
+ blocker = insert(:user)
+ blockee = insert(:user)
+ friend = insert(:user)
+
+ task =
+ Task.async(fn ->
+ refute_receive {:text, _}, 1_000
+ end)
+
+ fake_socket = %StreamerSocket{
+ transport_pid: task.pid,
+ user: blocker
+ }
+
+ topics = %{
+ "public" => [fake_socket]
+ }
+
+ {:ok, _user_relationship} = User.block(blocker, blockee)
+
+ {:ok, activity_one} = CommonAPI.post(friend, %{"status" => "hey! @#{blockee.nickname}"})
+
+ Worker.push_to_socket(topics, "public", activity_one)
+
+ {:ok, activity_two} = CommonAPI.post(blockee, %{"status" => "hey! @#{friend.nickname}"})
+
+ Worker.push_to_socket(topics, "public", activity_two)
+
+ {:ok, activity_three} = CommonAPI.post(blockee, %{"status" => "hey! @#{blocker.nickname}"})
+
+ Worker.push_to_socket(topics, "public", activity_three)
+
+ Task.await(task)
+ end
end
test "it doesn't send unwanted DMs to list" do
@@ -269,6 +338,12 @@ defmodule Pleroma.Web.StreamerTest do
{:ok, list} = List.create("Test", user_a)
{:ok, list} = List.follow(list, user_b)
+ {:ok, activity} =
+ CommonAPI.post(user_b, %{
+ "status" => "@#{user_c.nickname} Test",
+ "visibility" => "direct"
+ })
+
task =
Task.async(fn ->
refute_receive {:text, _}, 1_000
@@ -279,12 +354,6 @@ defmodule Pleroma.Web.StreamerTest do
user: user_a
}
- {:ok, activity} =
- CommonAPI.post(user_b, %{
- "status" => "@#{user_c.nickname} Test",
- "visibility" => "direct"
- })
-
topics = %{
"list:#{list.id}" => [fake_socket]
}
@@ -301,6 +370,12 @@ defmodule Pleroma.Web.StreamerTest do
{:ok, list} = List.create("Test", user_a)
{:ok, list} = List.follow(list, user_b)
+ {:ok, activity} =
+ CommonAPI.post(user_b, %{
+ "status" => "Test",
+ "visibility" => "private"
+ })
+
task =
Task.async(fn ->
refute_receive {:text, _}, 1_000
@@ -311,12 +386,6 @@ defmodule Pleroma.Web.StreamerTest do
user: user_a
}
- {:ok, activity} =
- CommonAPI.post(user_b, %{
- "status" => "Test",
- "visibility" => "private"
- })
-
topics = %{
"list:#{list.id}" => [fake_socket]
}
@@ -335,6 +404,12 @@ defmodule Pleroma.Web.StreamerTest do
{:ok, list} = List.create("Test", user_a)
{:ok, list} = List.follow(list, user_b)
+ {:ok, activity} =
+ CommonAPI.post(user_b, %{
+ "status" => "Test",
+ "visibility" => "private"
+ })
+
task =
Task.async(fn ->
assert_receive {:text, _}, 1_000
@@ -345,12 +420,6 @@ defmodule Pleroma.Web.StreamerTest do
user: user_a
}
- {:ok, activity} =
- CommonAPI.post(user_b, %{
- "status" => "Test",
- "visibility" => "private"
- })
-
Streamer.add_socket(
"list:#{list.id}",
fake_socket
@@ -367,6 +436,9 @@ defmodule Pleroma.Web.StreamerTest do
user3 = insert(:user)
CommonAPI.hide_reblogs(user1, user2)
+ {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
+ {:ok, announce_activity, _} = CommonAPI.repeat(create_activity.id, user2)
+
task =
Task.async(fn ->
refute_receive {:text, _}, 1_000
@@ -377,14 +449,39 @@ defmodule Pleroma.Web.StreamerTest do
user: user1
}
+ topics = %{
+ "public" => [fake_socket]
+ }
+
+ Worker.push_to_socket(topics, "public", announce_activity)
+
+ Task.await(task)
+ end
+
+ test "it does send non-reblog notification for reblog-muted actors" do
+ user1 = insert(:user)
+ user2 = insert(:user)
+ user3 = insert(:user)
+ CommonAPI.hide_reblogs(user1, user2)
+
{:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
- {:ok, announce_activity, _} = CommonAPI.repeat(create_activity.id, user2)
+ {:ok, favorite_activity, _} = CommonAPI.favorite(create_activity.id, user2)
+
+ task =
+ Task.async(fn ->
+ assert_receive {:text, _}, 1_000
+ end)
+
+ fake_socket = %StreamerSocket{
+ transport_pid: task.pid,
+ user: user1
+ }
topics = %{
"public" => [fake_socket]
}
- Worker.push_to_socket(topics, "public", announce_activity)
+ Worker.push_to_socket(topics, "public", favorite_activity)
Task.await(task)
end
@@ -398,9 +495,7 @@ defmodule Pleroma.Web.StreamerTest do
{:ok, activity} = CommonAPI.add_mute(user2, activity)
- task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)
-
- Process.sleep(4000)
+ task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)
Streamer.add_socket(
"user",
@@ -422,7 +517,14 @@ defmodule Pleroma.Web.StreamerTest do
task =
Task.async(fn ->
- assert_receive {:text, _received_event}, 4_000
+ assert_receive {:text, received_event}, @streamer_timeout
+
+ assert %{"event" => "conversation", "payload" => received_payload} =
+ Jason.decode!(received_event)
+
+ assert %{"last_status" => last_status} = Jason.decode!(received_payload)
+ [participation] = Participation.for_user(user)
+ assert last_status["pleroma"]["direct_conversation_id"] == participation.id
end)
Streamer.add_socket(
@@ -439,7 +541,7 @@ defmodule Pleroma.Web.StreamerTest do
Task.await(task)
end
- test "it doesn't send conversation update to the 'direct' streamj when the last message in the conversation is deleted" do
+ test "it doesn't send conversation update to the 'direct' stream when the last message in the conversation is deleted" do
user = insert(:user)
another_user = insert(:user)
@@ -451,13 +553,13 @@ defmodule Pleroma.Web.StreamerTest do
task =
Task.async(fn ->
- assert_receive {:text, received_event}, 4_000
+ assert_receive {:text, received_event}, @streamer_timeout
assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
- refute_receive {:text, _}, 4_000
+ refute_receive {:text, _}, @streamer_timeout
end)
- Process.sleep(1000)
+ Process.sleep(@streamer_start_wait)
Streamer.add_socket(
"direct",
@@ -488,10 +590,10 @@ defmodule Pleroma.Web.StreamerTest do
task =
Task.async(fn ->
- assert_receive {:text, received_event}, 4_000
+ assert_receive {:text, received_event}, @streamer_timeout
assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
- assert_receive {:text, received_event}, 4_000
+ assert_receive {:text, received_event}, @streamer_timeout
assert %{"event" => "conversation", "payload" => received_payload} =
Jason.decode!(received_event)
@@ -500,7 +602,7 @@ defmodule Pleroma.Web.StreamerTest do
assert last_status["id"] == to_string(create_activity.id)
end)
- Process.sleep(1000)
+ Process.sleep(@streamer_start_wait)
Streamer.add_socket(
"direct",