hj/pleroma
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Revert "Merge branch 'streamer-refactoring' into 'develop'"

Browse source 
This reverts merge request !1653
This commit is contained in:
kaniini 2019-09-16 09:09:21 +00:00
commit c623b4324d
26 changed files with 447 additions and 888 deletions
Download patch file Download diff file Expand all files Collapse all files

141
test/activity/ir/topics_test.exs
View file

@ -1,141 +0,0 @@
defmodule Pleroma.Activity.Ir.TopicsTest do
use Pleroma.DataCase
alias Pleroma.Activity
alias Pleroma.Activity.Ir.Topics
alias Pleroma.Object
require Pleroma.Constants
describe "poll answer" do
test "produce no topics" do
activity = %Activity{object: %Object{data: %{"type" => "Answer"}}}
assert [] == Topics.get_activity_topics(activity)
end
end
describe "non poll answer" do
test "always add user and list topics" do
activity = %Activity{object: %Object{data: %{"type" => "FooBar"}}}
topics = Topics.get_activity_topics(activity)
assert Enum.member?(topics, "user")
assert Enum.member?(topics, "list")
end
end
describe "public visibility" do
setup do
activity = %Activity{
object: %Object{data: %{"type" => "Note"}},
data: %{"to" => [Pleroma.Constants.as_public()]}
}
{:ok, activity: activity}
end
test "produces public topic", %{activity: activity} do
topics = Topics.get_activity_topics(activity)
assert Enum.member?(topics, "public")
end
test "local action produces public:local topic", %{activity: activity} do
activity = %{activity | local: true}
topics = Topics.get_activity_topics(activity)
assert Enum.member?(topics, "public:local")
end
test "non-local action does not produce public:local topic", %{activity: activity} do
activity = %{activity | local: false}
topics = Topics.get_activity_topics(activity)
refute Enum.member?(topics, "public:local")
end
end
describe "public visibility create events" do
setup do
activity = %Activity{
object: %Object{data: %{"type" => "Create", "attachment" => []}},
data: %{"to" => [Pleroma.Constants.as_public()]}
}
{:ok, activity: activity}
end
test "with no attachments doesn't produce public:media topics", %{activity: activity} do
topics = Topics.get_activity_topics(activity)
refute Enum.member?(topics, "public:media")
refute Enum.member?(topics, "public:local:media")
end
test "converts tags to hash tags", %{activity: %{object: %{data: data} = object} = activity} do
tagged_data = Map.put(data, "tag", ["foo", "bar"])
activity = %{activity | object: %{object | data: tagged_data}}
topics = Topics.get_activity_topics(activity)
assert Enum.member?(topics, "hashtag:foo")
assert Enum.member?(topics, "hashtag:bar")
end
test "only converts strinngs to hash tags", %{
activity: %{object: %{data: data} = object} = activity
} do
tagged_data = Map.put(data, "tag", [2])
activity = %{activity | object: %{object | data: tagged_data}}
topics = Topics.get_activity_topics(activity)
refute Enum.member?(topics, "hashtag:2")
end
end
describe "public visibility create events with attachments" do
setup do
activity = %Activity{
object: %Object{data: %{"type" => "Create", "attachment" => ["foo"]}},
data: %{"to" => [Pleroma.Constants.as_public()]}
}
{:ok, activity: activity}
end
test "produce public:media topics", %{activity: activity} do
topics = Topics.get_activity_topics(activity)
assert Enum.member?(topics, "public:media")
end
test "local produces public:local:media topics", %{activity: activity} do
topics = Topics.get_activity_topics(activity)
assert Enum.member?(topics, "public:local:media")
end
test "non-local doesn't produce public:local:media topics", %{activity: activity} do
activity = %{activity | local: false}
topics = Topics.get_activity_topics(activity)
refute Enum.member?(topics, "public:local:media")
end
end
describe "non-public visibility" do
test "produces direct topic" do
activity = %Activity{object: %Object{data: %{"type" => "Note"}}, data: %{"to" => []}}
topics = Topics.get_activity_topics(activity)
assert Enum.member?(topics, "direct")
refute Enum.member?(topics, "public")
refute Enum.member?(topics, "public:local")
refute Enum.member?(topics, "public:media")
refute Enum.member?(topics, "public:local:media")
end
end
end

16
test/integration/mastodon_websocket_test.exs
View file

@ -11,6 +11,7 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
alias Pleroma.Integration.WebsocketClient
alias Pleroma.Web.CommonAPI
alias Pleroma.Web.OAuth
alias Pleroma.Web.Streamer
@path Pleroma.Web.Endpoint.url()
|> URI.parse()
@ -18,6 +19,16 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
|> Map.put(:path, "/api/v1/streaming")
|> URI.to_string()
setup do
GenServer.start(Streamer, %{}, name: Streamer)
on_exit(fn ->
if pid = Process.whereis(Streamer) do
Process.exit(pid, :kill)
end
end)
end
def start_socket(qs \\ nil, headers \\ []) do
path =
case qs do
@ -42,14 +53,12 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
end)
end
@tag needs_streamer: true
test "allows public streams without authentication" do
assert {:ok, _} = start_socket("?stream=public")
assert {:ok, _} = start_socket("?stream=public:local")
assert {:ok, _} = start_socket("?stream=hashtag&tag=lain")
end
@tag needs_streamer: true
test "receives well formatted events" do
user = insert(:user)
{:ok, _} = start_socket("?stream=public")
@ -94,7 +103,6 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
assert {:ok, _} = start_socket("?stream=user&access_token=#{state.token.token}")
end
@tag needs_streamer: true
test "accepts the 'user' stream", %{token: token} = _state do
assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")
@ -103,7 +111,6 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
end) =~ ":badarg"
end
@tag needs_streamer: true
test "accepts the 'user:notification' stream", %{token: token} = _state do
assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")
@ -112,7 +119,6 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
end) =~ ":badarg"
end
@tag needs_streamer: true
test "accepts valid token on Sec-WebSocket-Protocol header", %{token: token} do
assert {:ok, _} = start_socket("?stream=user", [{"Sec-WebSocket-Protocol", token.token}])

11
test/notification_test.exs
View file

@ -69,7 +69,16 @@ defmodule Pleroma.NotificationTest do
end
describe "create_notification" do
@tag needs_streamer: true
setup do
GenServer.start(Streamer, %{}, name: Streamer)
on_exit(fn ->
if pid = Process.whereis(Streamer) do
Process.exit(pid, :kill)
end
end)
end
test "it creates a notification for user and send to the 'user' and the 'user:notification' stream" do
user = insert(:user)
task = Task.async(fn -> assert_receive {:text, _}, 4_000 end)

4
test/support/conn_case.ex
View file

@ -40,10 +40,6 @@ defmodule Pleroma.Web.ConnCase do
Ecto.Adapters.SQL.Sandbox.mode(Pleroma.Repo, {:shared, self()})
end
if tags[:needs_streamer] do
start_supervised(Pleroma.Web.Streamer.supervisor())
end
{:ok, conn: Phoenix.ConnTest.build_conn()}
end
end

4
test/support/data_case.ex
View file

@ -39,10 +39,6 @@ defmodule Pleroma.DataCase do
Ecto.Adapters.SQL.Sandbox.mode(Pleroma.Repo, {:shared, self()})
end
if tags[:needs_streamer] do
start_supervised(Pleroma.Web.Streamer.supervisor())
end
:ok
end

4
test/web/activity_pub/activity_pub_test.exs
View file

@ -38,7 +38,9 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
stream: fn _, _ -> nil end do
ActivityPub.stream_out_participations(conversation.participations)
assert called(Pleroma.Web.Streamer.stream("participation", participations))
Enum.each(participations, fn participation ->
assert called(Pleroma.Web.Streamer.stream("participation", participation))
end)
end
end
end

36
test/web/streamer/ping_test.exs
View file

@ -1,36 +0,0 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.PingTest do
use Pleroma.DataCase
import Pleroma.Factory
alias Pleroma.Web.Streamer
setup do
start_supervised({Streamer.supervisor(), [ping_interval: 30]})
:ok
end
describe "sockets" do
setup do
user = insert(:user)
{:ok, %{user: user}}
end
test "it sends pings", %{user: user} do
task =
Task.async(fn ->
assert_receive {:text, received_event}, 40
assert_receive {:text, received_event}, 40
assert_receive {:text, received_event}, 40
end)
Streamer.add_socket("public", %{transport_pid: task.pid, assigns: %{user: user}})
Task.await(task)
end
end
end

54
test/web/streamer/state_test.exs
View file

@ -1,54 +0,0 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Web.StateTest do
use Pleroma.DataCase
import Pleroma.Factory
alias Pleroma.Web.Streamer
alias Pleroma.Web.Streamer.StreamerSocket
@moduletag needs_streamer: true
describe "sockets" do
setup do
user = insert(:user)
user2 = insert(:user)
{:ok, %{user: user, user2: user2}}
end
test "it can add a socket", %{user: user} do
Streamer.add_socket("public", %{transport_pid: 1, assigns: %{user: user}})
assert(%{"public" => [%StreamerSocket{transport_pid: 1}]} = Streamer.get_sockets())
end
test "it can add multiple sockets per user", %{user: user} do
Streamer.add_socket("public", %{transport_pid: 1, assigns: %{user: user}})
Streamer.add_socket("public", %{transport_pid: 2, assigns: %{user: user}})
assert(
%{
"public" => [
%StreamerSocket{transport_pid: 2},
%StreamerSocket{transport_pid: 1}
]
} = Streamer.get_sockets()
)
end
test "it will not add a duplicate socket", %{user: user} do
Streamer.add_socket("activity", %{transport_pid: 1, assigns: %{user: user}})
Streamer.add_socket("activity", %{transport_pid: 1, assigns: %{user: user}})
assert(
%{
"activity" => [
%StreamerSocket{transport_pid: 1}
]
} = Streamer.get_sockets()
)
end
end
end

105
test/web/streamer/streamer_test.exs → test/web/streamer_test.exs
View file

@ -5,20 +5,24 @@
defmodule Pleroma.Web.StreamerTest do
use Pleroma.DataCase
import Pleroma.Factory
alias Pleroma.List
alias Pleroma.User
alias Pleroma.Web.CommonAPI
alias Pleroma.Web.Streamer
alias Pleroma.Web.Streamer.StreamerSocket
alias Pleroma.Web.Streamer.Worker
import Pleroma.Factory
@moduletag needs_streamer: true
clear_config_all([:instance, :skip_thread_containment])
describe "user streams" do
setup do
GenServer.start(Streamer, %{}, name: Streamer)
on_exit(fn ->
if pid = Process.whereis(Streamer) do
Process.exit(pid, :kill)
end
end)
user = insert(:user)
notify = insert(:notification, user: user, activity: build(:note_activity))
{:ok, %{user: user, notify: notify}}
@ -121,9 +125,11 @@ defmodule Pleroma.Web.StreamerTest do
assert_receive {:text, _}, 4_000
end)
fake_socket = %StreamerSocket{
fake_socket = %{
transport_pid: task.pid,
user: user
assigns: %{
user: user
}
}
{:ok, activity} = CommonAPI.post(other_user, %{"status" => "Test"})
@ -132,7 +138,7 @@ defmodule Pleroma.Web.StreamerTest do
"public" => [fake_socket]
}
Worker.push_to_socket(topics, "public", activity)
Streamer.push_to_socket(topics, "public", activity)
Task.await(task)
@ -149,9 +155,11 @@ defmodule Pleroma.Web.StreamerTest do
assert received_event == expected_event
end)
fake_socket = %StreamerSocket{
fake_socket = %{
transport_pid: task.pid,
user: user
assigns: %{
user: user
}
}
{:ok, activity} = CommonAPI.delete(activity.id, other_user)
@ -160,7 +168,7 @@ defmodule Pleroma.Web.StreamerTest do
"public" => [fake_socket]
}
Worker.push_to_socket(topics, "public", activity)
Streamer.push_to_socket(topics, "public", activity)
Task.await(task)
end
@ -181,9 +189,9 @@ defmodule Pleroma.Web.StreamerTest do
)
task = Task.async(fn -> refute_receive {:text, _}, 1_000 end)
fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
topics = %{"public" => [fake_socket]}
Worker.push_to_socket(topics, "public", activity)
Streamer.push_to_socket(topics, "public", activity)
Task.await(task)
end
@ -203,9 +211,9 @@ defmodule Pleroma.Web.StreamerTest do
)
task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
topics = %{"public" => [fake_socket]}
Worker.push_to_socket(topics, "public", activity)
Streamer.push_to_socket(topics, "public", activity)
Task.await(task)
end
@ -225,9 +233,9 @@ defmodule Pleroma.Web.StreamerTest do
)
task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
topics = %{"public" => [fake_socket]}
Worker.push_to_socket(topics, "public", activity)
Streamer.push_to_socket(topics, "public", activity)
Task.await(task)
end
@ -243,9 +251,11 @@ defmodule Pleroma.Web.StreamerTest do
refute_receive {:text, _}, 1_000
end)
fake_socket = %StreamerSocket{
fake_socket = %{
transport_pid: task.pid,
user: user
assigns: %{
user: user
}
}
{:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})
@ -254,7 +264,7 @@ defmodule Pleroma.Web.StreamerTest do
"public" => [fake_socket]
}
Worker.push_to_socket(topics, "public", activity)
Streamer.push_to_socket(topics, "public", activity)
Task.await(task)
end
@ -274,9 +284,11 @@ defmodule Pleroma.Web.StreamerTest do
refute_receive {:text, _}, 1_000
end)
fake_socket = %StreamerSocket{
fake_socket = %{
transport_pid: task.pid,
user: user_a
assigns: %{
user: user_a
}
}
{:ok, activity} =
@ -289,7 +301,7 @@ defmodule Pleroma.Web.StreamerTest do
"list:#{list.id}" => [fake_socket]
}
Worker.handle_call({:stream, "list", activity}, self(), topics)
Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
Task.await(task)
end
@ -306,9 +318,11 @@ defmodule Pleroma.Web.StreamerTest do
refute_receive {:text, _}, 1_000
end)
fake_socket = %StreamerSocket{
fake_socket = %{
transport_pid: task.pid,
user: user_a
assigns: %{
user: user_a
}
}
{:ok, activity} =
@ -321,12 +335,12 @@ defmodule Pleroma.Web.StreamerTest do
"list:#{list.id}" => [fake_socket]
}
Worker.handle_call({:stream, "list", activity}, self(), topics)
Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
Task.await(task)
end
test "it sends wanted private posts to list" do
test "it send wanted private posts to list" do
user_a = insert(:user)
user_b = insert(:user)
@ -340,9 +354,11 @@ defmodule Pleroma.Web.StreamerTest do
assert_receive {:text, _}, 1_000
end)
fake_socket = %StreamerSocket{
fake_socket = %{
transport_pid: task.pid,
user: user_a
assigns: %{
user: user_a
}
}
{:ok, activity} =
@ -351,12 +367,11 @@ defmodule Pleroma.Web.StreamerTest do
"visibility" => "private"
})
Streamer.add_socket(
"list:#{list.id}",
fake_socket
)
topics = %{
"list:#{list.id}" => [fake_socket]
}
Worker.handle_call({:stream, "list", activity}, self(), %{})
Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
Task.await(task)
end
@ -372,9 +387,11 @@ defmodule Pleroma.Web.StreamerTest do
refute_receive {:text, _}, 1_000
end)
fake_socket = %StreamerSocket{
fake_socket = %{
transport_pid: task.pid,
user: user1
assigns: %{
user: user1
}
}
{:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
@ -384,7 +401,7 @@ defmodule Pleroma.Web.StreamerTest do
"public" => [fake_socket]
}
Worker.push_to_socket(topics, "public", announce_activity)
Streamer.push_to_socket(topics, "public", announce_activity)
Task.await(task)
end
@ -400,8 +417,6 @@ defmodule Pleroma.Web.StreamerTest do
task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)
Process.sleep(4000)
Streamer.add_socket(
"user",
%{transport_pid: task.pid, assigns: %{user: user2}}
@ -413,6 +428,14 @@ defmodule Pleroma.Web.StreamerTest do
describe "direct streams" do
setup do
GenServer.start(Streamer, %{}, name: Streamer)
on_exit(fn ->
if pid = Process.whereis(Streamer) do
Process.exit(pid, :kill)
end
end)
:ok
end
@ -457,8 +480,6 @@ defmodule Pleroma.Web.StreamerTest do
refute_receive {:text, _}, 4_000
end)
Process.sleep(1000)
Streamer.add_socket(
"direct",
%{transport_pid: task.pid, assigns: %{user: user}}
@ -500,8 +521,6 @@ defmodule Pleroma.Web.StreamerTest do
assert last_status["id"] == to_string(create_activity.id)
end)
Process.sleep(1000)
Streamer.add_socket(
"direct",
%{transport_pid: task.pid, assigns: %{user: user}}