No OneTemporary
Actions

Size

68 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/web/mastodon_api/websocket_handler.ex b/lib/pleroma/web/mastodon_api/websocket_handler.ex
	index 5652a37c1..b1aebe014 100644
	--- a/lib/pleroma/web/mastodon_api/websocket_handler.ex
	+++ b/lib/pleroma/web/mastodon_api/websocket_handler.ex
	@@ -1,143 +1,102 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.MastodonAPI.WebsocketHandler do
	require Logger

	alias Pleroma.Repo
	alias Pleroma.User
	alias Pleroma.Web.OAuth.Token
	alias Pleroma.Web.Streamer

	@behaviour :cowboy_websocket

	- @streams [
	- "public",
	- "public:local",
	- "public:media",
	- "public:local:media",
	- "user",
	- "user:notification",
	- "direct",
	- "list",
	- "hashtag"
	- ]
	- @anonymous_streams ["public", "public:local", "hashtag"]
	-
	# Handled by periodic keepalive in Pleroma.Web.Streamer.Ping.
	@timeout :infinity

	def init(%{qs: qs} = req, state) do
	- with params <- :cow_qs.parse_qs(qs),
	+ with params <- Enum.into(:cow_qs.parse_qs(qs), %{}),
	sec_websocket <- :cowboy_req.header("sec-websocket-protocol", req, nil),
	- access_token <- List.keyfind(params, "access_token", 0),
	- {_, stream} <- List.keyfind(params, "stream", 0),
	- {:ok, user} <- allow_request(stream, [access_token, sec_websocket]),
	- topic when is_binary(topic) <- expand_topic(stream, params) do
	+ access_token <- Map.get(params, "access_token"),
	+ {:ok, user} <- authenticate_request(access_token, sec_websocket),
	+ {:ok, topic} <- Streamer.get_topic(Map.get(params, "stream"), user, params) do
	req =
	if sec_websocket do
	:cowboy_req.set_resp_header("sec-websocket-protocol", sec_websocket, req)
	else
	req
	end

	{:cowboy_websocket, req, %{user: user, topic: topic}, %{idle_timeout: @timeout}}
	else
	- {:error, code} ->
	- Logger.debug("#{__MODULE__} denied connection: #{inspect(code)} - #{inspect(req)}")
	- {:ok, req} = :cowboy_req.reply(code, req)
	+ {:error, :bad_topic} ->
	+ Logger.debug("#{__MODULE__} bad topic #{inspect(req)}")
	+ {:ok, req} = :cowboy_req.reply(404, req)
	{:ok, req, state}

	- error ->
	- Logger.debug("#{__MODULE__} denied connection: #{inspect(error)} - #{inspect(req)}")
	- {:ok, req} = :cowboy_req.reply(400, req)
	+ {:error, :unauthorized} ->
	+ Logger.debug("#{__MODULE__} authentication error: #{inspect(req)}")
	+ {:ok, req} = :cowboy_req.reply(401, req)
	{:ok, req, state}
	end
	end

	def websocket_init(state) do
	send(self(), :subscribe)
	{:ok, state}
	end

	# We never receive messages.
	def websocket_handle(_frame, state) do
	{:ok, state}
	end

	def websocket_info(:subscribe, state) do
	Logger.debug(
	"#{__MODULE__} accepted websocket connection for user #{
	(state.user \|\| %{id: "anonymous"}).id
	}, topic #{state.topic}"
	)

	Streamer.add_socket(state.topic, streamer_socket(state))
	{:ok, state}
	end

	def websocket_info({:text, message}, state) do
	{:reply, {:text, message}, state}
	end

	def terminate(reason, _req, state) do
	Logger.debug(
	"#{__MODULE__} terminating websocket connection for user #{
	(state.user \|\| %{id: "anonymous"}).id
	}, topic #{state.topic \|\| "?"}: #{inspect(reason)}"
	)

	Streamer.remove_socket(state.topic, streamer_socket(state))
	:ok
	end

	# Public streams without authentication.
	- defp allow_request(stream, [nil, nil]) when stream in @anonymous_streams do
	+ defp authenticate_request(nil, nil) do
	{:ok, nil}
	end

	# Authenticated streams.
	- defp allow_request(stream, [access_token, sec_websocket]) when stream in @streams do
	- token =
	- with {"access_token", token} <- access_token do
	- token
	- else
	- _ -> sec_websocket
	- end
	+ defp authenticate_request(access_token, sec_websocket) do
	+ token = access_token \|\| sec_websocket

	with true <- is_bitstring(token),
	%Token{user_id: user_id} <- Repo.get_by(Token, token: token),
	user = %User{} <- User.get_cached_by_id(user_id) do
	{:ok, user}
	else
	- _ -> {:error, 403}
	- end
	- end
	-
	- # Not authenticated.
	- defp allow_request(stream, _) when stream in @streams, do: {:error, 403}
	-
	- # No matching stream.
	- defp allow_request(_, _), do: {:error, 404}
	-
	- defp expand_topic("hashtag", params) do
	- case List.keyfind(params, "tag", 0) do
	- {_, tag} -> "hashtag:#{tag}"
	- _ -> nil
	+ _ -> {:error, :unauthorized}
	end
	end

	- defp expand_topic("list", params) do
	- case List.keyfind(params, "list", 0) do
	- {_, list} -> "list:#{list}"
	- _ -> nil
	- end
	- end
	-
	- defp expand_topic(topic, _), do: topic
	-
	defp streamer_socket(state) do
	%{transport_pid: self(), assigns: state}
	end
	end
	diff --git a/lib/pleroma/web/streamer/state.ex b/lib/pleroma/web/streamer/state.ex
	index 999550b88..4eb462a1a 100644
	--- a/lib/pleroma/web/streamer/state.ex
	+++ b/lib/pleroma/web/streamer/state.ex
	@@ -1,82 +1,71 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Streamer.State do
	use GenServer
	require Logger

	alias Pleroma.Web.Streamer.StreamerSocket

	@env Mix.env()

	def start_link(_) do
	GenServer.start_link(__MODULE__, %{sockets: %{}}, name: __MODULE__)
	end

	def add_socket(topic, socket) do
	GenServer.call(__MODULE__, {:add, topic, socket})
	end

	def remove_socket(topic, socket) do
	do_remove_socket(@env, topic, socket)
	end

	def get_sockets do
	%{sockets: stream_sockets} = GenServer.call(__MODULE__, :get_state)
	stream_sockets
	end

	def init(init_arg) do
	{:ok, init_arg}
	end

	def handle_call(:get_state, _from, state) do
	{:reply, state, state}
	end

	def handle_call({:add, topic, socket}, _from, %{sockets: sockets} = state) do
	- internal_topic = internal_topic(topic, socket)
	stream_socket = StreamerSocket.from_socket(socket)

	sockets_for_topic =
	sockets
	- \|> Map.get(internal_topic, [])
	+ \|> Map.get(topic, [])
	\|> List.insert_at(0, stream_socket)
	\|> Enum.uniq()

	- state = put_in(state, [:sockets, internal_topic], sockets_for_topic)
	+ state = put_in(state, [:sockets, topic], sockets_for_topic)
	Logger.debug("Got new conn for #{topic}")
	{:reply, state, state}
	end

	def handle_call({:remove, topic, socket}, _from, %{sockets: sockets} = state) do
	- internal_topic = internal_topic(topic, socket)
	stream_socket = StreamerSocket.from_socket(socket)

	sockets_for_topic =
	sockets
	- \|> Map.get(internal_topic, [])
	+ \|> Map.get(topic, [])
	\|> List.delete(stream_socket)

	- state = Kernel.put_in(state, [:sockets, internal_topic], sockets_for_topic)
	+ state = Kernel.put_in(state, [:sockets, topic], sockets_for_topic)
	{:reply, state, state}
	end

	defp do_remove_socket(:test, _, _) do
	:ok
	end

	defp do_remove_socket(_env, topic, socket) do
	GenServer.call(__MODULE__, {:remove, topic, socket})
	end
	-
	- defp internal_topic(topic, socket)
	- when topic in ~w[user user:notification direct] do
	- "#{topic}:#{socket.assigns[:user].id}"
	- end
	-
	- defp internal_topic(topic, _) do
	- topic
	- end
	end
	diff --git a/lib/pleroma/web/streamer/streamer.ex b/lib/pleroma/web/streamer/streamer.ex
	index 814d5a729..1e5700b6a 100644
	--- a/lib/pleroma/web/streamer/streamer.ex
	+++ b/lib/pleroma/web/streamer/streamer.ex
	@@ -1,55 +1,120 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.Streamer do
	alias Pleroma.Web.Streamer.State
	alias Pleroma.Web.Streamer.Worker
	+ alias Pleroma.User

	@timeout 60_000
	@mix_env Mix.env()

	+ @public_streams ["public", "public:local", "public:media", "public:local:media"]
	+ @user_streams ["user", "user:notification", "direct"]
	+
	+ @doc "Expands and authorizes a stream, and registers the process for streaming."
	+ @spec get_topic_and_add_socket(stream :: String.t(), State.t(), Map.t() \| nil) ::
	+ {:ok, topic :: String.t()} \| {:error, :bad_topic} \| {:error, :unauthorized}
	+ def get_topic_and_add_socket(stream, socket, params \\ %{}) do
	+ user =
	+ case socket do
	+ %{assigns: %{user: user}} -> user
	+ _ -> nil
	+ end
	+
	+ case get_topic(stream, user, params) do
	+ {:ok, topic} ->
	+ add_socket(topic, socket)
	+ {:ok, topic}
	+
	+ error ->
	+ error
	+ end
	+ end
	+
	+ @doc "Expand and authorizes a stream"
	+ @spec get_topic(stream :: String.t(), User.t() \| nil, Map.t()) ::
	+ {:ok, topic :: String.t()} \| {:error, :bad_topic}
	+ def get_topic(stream, user, params \\ %{})
	+
	+ # Allow all public steams.
	+ def get_topic(stream, _, _) when stream in @public_streams do
	+ {:ok, stream}
	+ end
	+
	+ # Allow all hashtags streams.
	+ def get_topic("hashtag", _, %{"tag" => tag}) do
	+ {:ok, "hashtag:" <> tag}
	+ end
	+
	+ # Expand user streams.
	+ def get_topic(stream, %User{} = user, _) when stream in @user_streams do
	+ {:ok, stream <> ":" <> to_string(user.id)}
	+ end
	+
	+ def get_topic(stream, _, _) when stream in @user_streams do
	+ {:error, :unauthorized}
	+ end
	+
	+ # List streams.
	+ def get_topic("list", %User{} = user, %{"list" => id}) do
	+ if Pleroma.List.get(id, user) do
	+ {:ok, "list:" <> to_string(id)}
	+ else
	+ {:error, :bad_topic}
	+ end
	+ end
	+
	+ def get_topic("list", _, _) do
	+ {:error, :unauthorized}
	+ end
	+
	+ def get_topic(_, _, _) do
	+ {:error, :bad_topic}
	+ end
	+
	def add_socket(topic, socket) do
	State.add_socket(topic, socket)
	end

	def remove_socket(topic, socket) do
	State.remove_socket(topic, socket)
	end

	def get_sockets do
	State.get_sockets()
	end

	def stream(topics, items) do
	if should_send?() do
	Task.async(fn ->
	:poolboy.transaction(
	:streamer_worker,
	&Worker.stream(&1, topics, items),
	@timeout
	)
	end)
	end
	end

	def supervisor, do: Pleroma.Web.Streamer.Supervisor

	defp should_send? do
	handle_should_send(@mix_env)
	end

	defp handle_should_send(:test) do
	case Process.whereis(:streamer_worker) do
	nil ->
	false

	pid ->
	Process.alive?(pid)
	end
	end

	defp handle_should_send(:benchmark), do: false

	defp handle_should_send(_), do: true
	end
	diff --git a/test/integration/mastodon_websocket_test.exs b/test/integration/mastodon_websocket_test.exs
	index bd229c55f..39be5ad08 100644
	--- a/test/integration/mastodon_websocket_test.exs
	+++ b/test/integration/mastodon_websocket_test.exs
	@@ -1,131 +1,131 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Integration.MastodonWebsocketTest do
	use Pleroma.DataCase

	import ExUnit.CaptureLog
	import Pleroma.Factory

	alias Pleroma.Integration.WebsocketClient
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.OAuth

	@path Pleroma.Web.Endpoint.url()
	\|> URI.parse()
	\|> Map.put(:scheme, "ws")
	\|> Map.put(:path, "/api/v1/streaming")
	\|> URI.to_string()

	setup_all do
	start_supervised(Pleroma.Web.Streamer.supervisor())
	:ok
	end

	def start_socket(qs \\ nil, headers \\ []) do
	path =
	case qs do
	nil -> @path
	qs -> @path <> qs
	end

	WebsocketClient.start_link(self(), path, headers)
	end

	test "refuses invalid requests" do
	capture_log(fn ->
	- assert {:error, {400, _}} = start_socket()
	+ assert {:error, {404, _}} = start_socket()
	assert {:error, {404, _}} = start_socket("?stream=ncjdk")
	Process.sleep(30)
	end)
	end

	test "requires authentication and a valid token for protected streams" do
	capture_log(fn ->
	- assert {:error, {403, _}} = start_socket("?stream=user&access_token=aaaaaaaaaaaa")
	- assert {:error, {403, _}} = start_socket("?stream=user")
	+ assert {:error, {401, _}} = start_socket("?stream=user&access_token=aaaaaaaaaaaa")
	+ assert {:error, {401, _}} = start_socket("?stream=user")
	Process.sleep(30)
	end)
	end

	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

	test "receives well formatted events" do
	user = insert(:user)
	{:ok, _} = start_socket("?stream=public")
	{:ok, activity} = CommonAPI.post(user, %{"status" => "nice echo chamber"})

	assert_receive {:text, raw_json}, 1_000
	assert {:ok, json} = Jason.decode(raw_json)

	assert "update" == json["event"]
	assert json["payload"]
	assert {:ok, json} = Jason.decode(json["payload"])

	view_json =
	Pleroma.Web.MastodonAPI.StatusView.render("show.json", activity: activity, for: nil)
	\|> Jason.encode!()
	\|> Jason.decode!()

	assert json == view_json
	end

	describe "with a valid user token" do
	setup do
	{:ok, app} =
	Pleroma.Repo.insert(
	OAuth.App.register_changeset(%OAuth.App{}, %{
	client_name: "client",
	scopes: ["scope"],
	redirect_uris: "url"
	})
	)

	user = insert(:user)

	{:ok, auth} = OAuth.Authorization.create_authorization(app, user)

	{:ok, token} = OAuth.Token.exchange_token(app, auth)

	%{user: user, token: token}
	end

	test "accepts valid tokens", state do
	assert {:ok, _} = start_socket("?stream=user&access_token=#{state.token.token}")
	end

	test "accepts the 'user' stream", %{token: token} = _state do
	assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")

	assert capture_log(fn ->
	- assert {:error, {403, "Forbidden"}} = start_socket("?stream=user")
	+ assert {:error, {401, _}} = start_socket("?stream=user")
	Process.sleep(30)
	end) =~ ":badarg"
	end

	test "accepts the 'user:notification' stream", %{token: token} = _state do
	assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")

	assert capture_log(fn ->
	- assert {:error, {403, "Forbidden"}} = start_socket("?stream=user:notification")
	+ assert {:error, {401, _}} = start_socket("?stream=user:notification")
	Process.sleep(30)
	end) =~ ":badarg"
	end

	test "accepts valid token on Sec-WebSocket-Protocol header", %{token: token} do
	assert {:ok, _} = start_socket("?stream=user", [{"Sec-WebSocket-Protocol", token.token}])

	assert capture_log(fn ->
	- assert {:error, {403, "Forbidden"}} =
	+ assert {:error, {401, _}} =
	start_socket("?stream=user", [{"Sec-WebSocket-Protocol", "I am a friend"}])

	Process.sleep(30)
	end) =~ ":badarg"
	end
	end
	end
	diff --git a/test/notification_test.exs b/test/notification_test.exs
	index e12418db3..d04754a9d 100644
	--- a/test/notification_test.exs
	+++ b/test/notification_test.exs
	@@ -1,970 +1,971 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.NotificationTest do
	use Pleroma.DataCase

	import Pleroma.Factory
	import Mock

	alias Pleroma.FollowingRelationship
	alias Pleroma.Notification
	alias Pleroma.Tests.ObanHelpers
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.Transmogrifier
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.MastodonAPI.NotificationView
	alias Pleroma.Web.Push
	alias Pleroma.Web.Streamer

	describe "create_notifications" do
	test "creates a notification for an emoji reaction" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "yeah"})
	{:ok, activity, _object} = CommonAPI.react_with_emoji(activity.id, other_user, "☕")

	{:ok, [notification]} = Notification.create_notifications(activity)

	assert notification.user_id == user.id
	end

	test "notifies someone when they are directly addressed" do
	user = insert(:user)
	other_user = insert(:user)
	third_user = insert(:user)

	{:ok, activity} =
	CommonAPI.post(user, %{
	"status" => "hey @#{other_user.nickname} and @#{third_user.nickname}"
	})

	{:ok, [notification, other_notification]} = Notification.create_notifications(activity)

	notified_ids = Enum.sort([notification.user_id, other_notification.user_id])
	assert notified_ids == [other_user.id, third_user.id]
	assert notification.activity_id == activity.id
	assert other_notification.activity_id == activity.id
	end

	test "it creates a notification for subscribed users" do
	user = insert(:user)
	subscriber = insert(:user)

	User.subscribe(subscriber, user)

	{:ok, status} = CommonAPI.post(user, %{"status" => "Akariiiin"})
	{:ok, [notification]} = Notification.create_notifications(status)

	assert notification.user_id == subscriber.id
	end

	test "does not create a notification for subscribed users if status is a reply" do
	user = insert(:user)
	other_user = insert(:user)
	subscriber = insert(:user)

	User.subscribe(subscriber, other_user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	{:ok, _reply_activity} =
	CommonAPI.post(other_user, %{
	"status" => "test reply",
	"in_reply_to_status_id" => activity.id
	})

	user_notifications = Notification.for_user(user)
	assert length(user_notifications) == 1

	subscriber_notifications = Notification.for_user(subscriber)
	assert Enum.empty?(subscriber_notifications)
	end
	end

	describe "CommonApi.post/2 notification-related functionality" do
	test_with_mock "creates but does NOT send notification to blocker user",
	Push,
	[:passthrough],
	[] do
	user = insert(:user)
	blocker = insert(:user)
	{:ok, _user_relationship} = User.block(blocker, user)

	{:ok, _activity} = CommonAPI.post(user, %{"status" => "hey @#{blocker.nickname}!"})

	blocker_id = blocker.id
	assert [%Notification{user_id: ^blocker_id}] = Repo.all(Notification)
	refute called(Push.send(:_))
	end

	test_with_mock "creates but does NOT send notification to notification-muter user",
	Push,
	[:passthrough],
	[] do
	user = insert(:user)
	muter = insert(:user)
	{:ok, _user_relationships} = User.mute(muter, user)

	{:ok, _activity} = CommonAPI.post(user, %{"status" => "hey @#{muter.nickname}!"})

	muter_id = muter.id
	assert [%Notification{user_id: ^muter_id}] = Repo.all(Notification)
	refute called(Push.send(:_))
	end

	test_with_mock "creates but does NOT send notification to thread-muter user",
	Push,
	[:passthrough],
	[] do
	user = insert(:user)
	thread_muter = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{thread_muter.nickname}!"})

	{:ok, _} = CommonAPI.add_mute(thread_muter, activity)

	{:ok, _same_context_activity} =
	CommonAPI.post(user, %{
	"status" => "hey-hey-hey @#{thread_muter.nickname}!",
	"in_reply_to_status_id" => activity.id
	})

	[pre_mute_notification, post_mute_notification] =
	Repo.all(from(n in Notification, where: n.user_id == ^thread_muter.id, order_by: n.id))

	pre_mute_notification_id = pre_mute_notification.id
	post_mute_notification_id = post_mute_notification.id

	assert called(
	Push.send(
	:meck.is(fn
	%Notification{id: ^pre_mute_notification_id} -> true
	_ -> false
	end)
	)
	)

	refute called(
	Push.send(
	:meck.is(fn
	%Notification{id: ^post_mute_notification_id} -> true
	_ -> false
	end)
	)
	)
	end
	end

	describe "create_notification" do
	@tag needs_streamer: true
	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)
	task_user_notification = Task.async(fn -> assert_receive {:text, _}, 4_000 end)
	- Streamer.add_socket("user", %{transport_pid: task.pid, assigns: %{user: user}})

	- Streamer.add_socket(
	- "user:notification",
	- %{transport_pid: task_user_notification.pid, assigns: %{user: user}}
	- )
	+ Streamer.get_topic_and_add_socket("user", %{transport_pid: task.pid, assigns: %{user: user}})
	+
	+ Streamer.get_topic_and_add_socket("user:notification", %{
	+ transport_pid: task_user_notification.pid,
	+ assigns: %{user: user}
	+ })

	activity = insert(:note_activity)

	notify = Notification.create_notification(activity, user)
	assert notify.user_id == user.id
	Task.await(task)
	Task.await(task_user_notification)
	end

	test "it creates a notification for user if the user blocks the activity author" do
	activity = insert(:note_activity)
	author = User.get_cached_by_ap_id(activity.data["actor"])
	user = insert(:user)
	{:ok, _user_relationship} = User.block(user, author)

	assert Notification.create_notification(activity, user)
	end

	test "it creates a notification for the user if the user mutes the activity author" do
	muter = insert(:user)
	muted = insert(:user)
	{:ok, _} = User.mute(muter, muted)
	muter = Repo.get(User, muter.id)
	{:ok, activity} = CommonAPI.post(muted, %{"status" => "Hi @#{muter.nickname}"})

	assert Notification.create_notification(activity, muter)
	end

	test "notification created if user is muted without notifications" do
	muter = insert(:user)
	muted = insert(:user)

	{:ok, _user_relationships} = User.mute(muter, muted, false)

	{:ok, activity} = CommonAPI.post(muted, %{"status" => "Hi @#{muter.nickname}"})

	assert Notification.create_notification(activity, muter)
	end

	test "it creates a notification for an activity from a muted thread" do
	muter = insert(:user)
	other_user = insert(:user)
	{:ok, activity} = CommonAPI.post(muter, %{"status" => "hey"})
	CommonAPI.add_mute(muter, activity)

	{:ok, activity} =
	CommonAPI.post(other_user, %{
	"status" => "Hi @#{muter.nickname}",
	"in_reply_to_status_id" => activity.id
	})

	assert Notification.create_notification(activity, muter)
	end

	test "it disables notifications from followers" do
	follower = insert(:user)

	followed =
	insert(:user, notification_settings: %Pleroma.User.NotificationSetting{followers: false})

	User.follow(follower, followed)
	{:ok, activity} = CommonAPI.post(follower, %{"status" => "hey @#{followed.nickname}"})
	refute Notification.create_notification(activity, followed)
	end

	test "it disables notifications from non-followers" do
	follower = insert(:user)

	followed =
	insert(:user,
	notification_settings: %Pleroma.User.NotificationSetting{non_followers: false}
	)

	{:ok, activity} = CommonAPI.post(follower, %{"status" => "hey @#{followed.nickname}"})
	refute Notification.create_notification(activity, followed)
	end

	test "it disables notifications from people the user follows" do
	follower =
	insert(:user, notification_settings: %Pleroma.User.NotificationSetting{follows: false})

	followed = insert(:user)
	User.follow(follower, followed)
	follower = Repo.get(User, follower.id)
	{:ok, activity} = CommonAPI.post(followed, %{"status" => "hey @#{follower.nickname}"})
	refute Notification.create_notification(activity, follower)
	end

	test "it disables notifications from people the user does not follow" do
	follower =
	insert(:user, notification_settings: %Pleroma.User.NotificationSetting{non_follows: false})

	followed = insert(:user)
	{:ok, activity} = CommonAPI.post(followed, %{"status" => "hey @#{follower.nickname}"})
	refute Notification.create_notification(activity, follower)
	end

	test "it doesn't create a notification for user if he is the activity author" do
	activity = insert(:note_activity)
	author = User.get_cached_by_ap_id(activity.data["actor"])

	refute Notification.create_notification(activity, author)
	end

	test "it doesn't create duplicate notifications for follow+subscribed users" do
	user = insert(:user)
	subscriber = insert(:user)

	{:ok, _, _, _} = CommonAPI.follow(subscriber, user)
	User.subscribe(subscriber, user)
	{:ok, status} = CommonAPI.post(user, %{"status" => "Akariiiin"})
	{:ok, [_notif]} = Notification.create_notifications(status)
	end

	test "it doesn't create subscription notifications if the recipient cannot see the status" do
	user = insert(:user)
	subscriber = insert(:user)

	User.subscribe(subscriber, user)

	{:ok, status} = CommonAPI.post(user, %{"status" => "inwisible", "visibility" => "direct"})

	assert {:ok, []} == Notification.create_notifications(status)
	end
	end

	describe "follow / follow_request notifications" do
	test "it creates `follow` notification for approved Follow activity" do
	user = insert(:user)
	followed_user = insert(:user, locked: false)

	{:ok, _, _, _activity} = CommonAPI.follow(user, followed_user)
	assert FollowingRelationship.following?(user, followed_user)
	assert [notification] = Notification.for_user(followed_user)

	assert %{type: "follow"} =
	NotificationView.render("show.json", %{
	notification: notification,
	for: followed_user
	})
	end

	test "it creates `follow_request` notification for pending Follow activity" do
	user = insert(:user)
	followed_user = insert(:user, locked: true)

	{:ok, _, _, _activity} = CommonAPI.follow(user, followed_user)
	refute FollowingRelationship.following?(user, followed_user)
	assert [notification] = Notification.for_user(followed_user)

	render_opts = %{notification: notification, for: followed_user}
	assert %{type: "follow_request"} = NotificationView.render("show.json", render_opts)

	# After request is accepted, the same notification is rendered with type "follow":
	assert {:ok, _} = CommonAPI.accept_follow_request(user, followed_user)

	notification_id = notification.id
	assert [%{id: ^notification_id}] = Notification.for_user(followed_user)
	assert %{type: "follow"} = NotificationView.render("show.json", render_opts)
	end

	test "it doesn't create a notification for follow-unfollow-follow chains" do
	user = insert(:user)
	followed_user = insert(:user, locked: false)

	{:ok, _, _, _activity} = CommonAPI.follow(user, followed_user)
	assert FollowingRelationship.following?(user, followed_user)
	assert [notification] = Notification.for_user(followed_user)

	CommonAPI.unfollow(user, followed_user)
	{:ok, _, _, _activity_dupe} = CommonAPI.follow(user, followed_user)

	notification_id = notification.id
	assert [%{id: ^notification_id}] = Notification.for_user(followed_user)
	end

	test "dismisses the notification on follow request rejection" do
	user = insert(:user, locked: true)
	follower = insert(:user)
	{:ok, _, _, _follow_activity} = CommonAPI.follow(follower, user)
	assert [notification] = Notification.for_user(user)
	{:ok, _follower} = CommonAPI.reject_follow_request(follower, user)
	assert [] = Notification.for_user(user)
	end
	end

	describe "get notification" do
	test "it gets a notification that belongs to the user" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}"})

	{:ok, [notification]} = Notification.create_notifications(activity)
	{:ok, notification} = Notification.get(other_user, notification.id)

	assert notification.user_id == other_user.id
	end

	test "it returns error if the notification doesn't belong to the user" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}"})

	{:ok, [notification]} = Notification.create_notifications(activity)
	{:error, _notification} = Notification.get(user, notification.id)
	end
	end

	describe "dismiss notification" do
	test "it dismisses a notification that belongs to the user" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}"})

	{:ok, [notification]} = Notification.create_notifications(activity)
	{:ok, notification} = Notification.dismiss(other_user, notification.id)

	assert notification.user_id == other_user.id
	end

	test "it returns error if the notification doesn't belong to the user" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}"})

	{:ok, [notification]} = Notification.create_notifications(activity)
	{:error, _notification} = Notification.dismiss(user, notification.id)
	end
	end

	describe "clear notification" do
	test "it clears all notifications belonging to the user" do
	user = insert(:user)
	other_user = insert(:user)
	third_user = insert(:user)

	{:ok, activity} =
	CommonAPI.post(user, %{
	"status" => "hey @#{other_user.nickname} and @#{third_user.nickname} !"
	})

	{:ok, _notifs} = Notification.create_notifications(activity)

	{:ok, activity} =
	CommonAPI.post(user, %{
	"status" => "hey again @#{other_user.nickname} and @#{third_user.nickname} !"
	})

	{:ok, _notifs} = Notification.create_notifications(activity)
	Notification.clear(other_user)

	assert Notification.for_user(other_user) == []
	assert Notification.for_user(third_user) != []
	end
	end

	describe "set_read_up_to()" do
	test "it sets all notifications as read up to a specified notification ID" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, _activity} =
	CommonAPI.post(user, %{
	"status" => "hey @#{other_user.nickname}!"
	})

	{:ok, _activity} =
	CommonAPI.post(user, %{
	"status" => "hey again @#{other_user.nickname}!"
	})

	[n2, n1] = notifs = Notification.for_user(other_user)
	assert length(notifs) == 2

	assert n2.id > n1.id

	{:ok, _activity} =
	CommonAPI.post(user, %{
	"status" => "hey yet again @#{other_user.nickname}!"
	})

	Notification.set_read_up_to(other_user, n2.id)

	[n3, n2, n1] = Notification.for_user(other_user)

	assert n1.seen == true
	assert n2.seen == true
	assert n3.seen == false
	end
	end

	describe "for_user_since/2" do
	defp days_ago(days) do
	NaiveDateTime.add(
	NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second),
	-days * 60 * 60 * 24,
	:second
	)
	end

	test "Returns recent notifications" do
	user1 = insert(:user)
	user2 = insert(:user)

	Enum.each(0..10, fn i ->
	{:ok, _activity} =
	CommonAPI.post(user1, %{
	"status" => "hey ##{i} @#{user2.nickname}!"
	})
	end)

	{old, new} = Enum.split(Notification.for_user(user2), 5)

	Enum.each(old, fn notification ->
	notification
	\|> cast(%{updated_at: days_ago(10)}, [:updated_at])
	\|> Pleroma.Repo.update!()
	end)

	recent_notifications_ids =
	user2
	\|> Notification.for_user_since(
	NaiveDateTime.add(NaiveDateTime.utc_now(), -5 * 86_400, :second)
	)
	\|> Enum.map(& &1.id)

	Enum.each(old, fn %{id: id} ->
	refute id in recent_notifications_ids
	end)

	Enum.each(new, fn %{id: id} ->
	assert id in recent_notifications_ids
	end)
	end
	end

	describe "notification target determination / get_notified_from_activity/2" do
	test "it sends notifications to addressed users in new messages" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} =
	CommonAPI.post(user, %{
	"status" => "hey @#{other_user.nickname}!"
	})

	{enabled_receivers, _disabled_receivers} = Notification.get_notified_from_activity(activity)

	assert other_user in enabled_receivers
	end

	test "it sends notifications to mentioned users in new messages" do
	user = insert(:user)
	other_user = insert(:user)

	create_activity = %{
	"@context" => "https://www.w3.org/ns/activitystreams",
	"type" => "Create",
	"to" => ["https://www.w3.org/ns/activitystreams#Public"],
	"actor" => user.ap_id,
	"object" => %{
	"type" => "Note",
	"content" => "message with a Mention tag, but no explicit tagging",
	"tag" => [
	%{
	"type" => "Mention",
	"href" => other_user.ap_id,
	"name" => other_user.nickname
	}
	],
	"attributedTo" => user.ap_id
	}
	}

	{:ok, activity} = Transmogrifier.handle_incoming(create_activity)

	{enabled_receivers, _disabled_receivers} = Notification.get_notified_from_activity(activity)

	assert other_user in enabled_receivers
	end

	test "it does not send notifications to users who are only cc in new messages" do
	user = insert(:user)
	other_user = insert(:user)

	create_activity = %{
	"@context" => "https://www.w3.org/ns/activitystreams",
	"type" => "Create",
	"to" => ["https://www.w3.org/ns/activitystreams#Public"],
	"cc" => [other_user.ap_id],
	"actor" => user.ap_id,
	"object" => %{
	"type" => "Note",
	"content" => "hi everyone",
	"attributedTo" => user.ap_id
	}
	}

	{:ok, activity} = Transmogrifier.handle_incoming(create_activity)

	{enabled_receivers, _disabled_receivers} = Notification.get_notified_from_activity(activity)

	assert other_user not in enabled_receivers
	end

	test "it does not send notification to mentioned users in likes" do
	user = insert(:user)
	other_user = insert(:user)
	third_user = insert(:user)

	{:ok, activity_one} =
	CommonAPI.post(user, %{
	"status" => "hey @#{other_user.nickname}!"
	})

	{:ok, activity_two, _} = CommonAPI.favorite(activity_one.id, third_user)

	{enabled_receivers, _disabled_receivers} =
	Notification.get_notified_from_activity(activity_two)

	assert other_user not in enabled_receivers
	end

	test "it does not send notification to mentioned users in announces" do
	user = insert(:user)
	other_user = insert(:user)
	third_user = insert(:user)

	{:ok, activity_one} =
	CommonAPI.post(user, %{
	"status" => "hey @#{other_user.nickname}!"
	})

	{:ok, activity_two, _} = CommonAPI.repeat(activity_one.id, third_user)

	{enabled_receivers, _disabled_receivers} =
	Notification.get_notified_from_activity(activity_two)

	assert other_user not in enabled_receivers
	end

	test "it returns blocking recipient in disabled recipients list" do
	user = insert(:user)
	other_user = insert(:user)
	{:ok, _user_relationship} = User.block(other_user, user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}!"})

	{enabled_receivers, disabled_receivers} = Notification.get_notified_from_activity(activity)

	assert [] == enabled_receivers
	assert [other_user] == disabled_receivers
	end

	test "it returns notification-muting recipient in disabled recipients list" do
	user = insert(:user)
	other_user = insert(:user)
	{:ok, _user_relationships} = User.mute(other_user, user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}!"})

	{enabled_receivers, disabled_receivers} = Notification.get_notified_from_activity(activity)

	assert [] == enabled_receivers
	assert [other_user] == disabled_receivers
	end

	test "it returns thread-muting recipient in disabled recipients list" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "hey @#{other_user.nickname}!"})

	{:ok, _} = CommonAPI.add_mute(other_user, activity)

	{:ok, same_context_activity} =
	CommonAPI.post(user, %{
	"status" => "hey-hey-hey @#{other_user.nickname}!",
	"in_reply_to_status_id" => activity.id
	})

	{enabled_receivers, disabled_receivers} =
	Notification.get_notified_from_activity(same_context_activity)

	assert [other_user] == disabled_receivers
	refute other_user in enabled_receivers
	end
	end

	describe "notification lifecycle" do
	test "liking an activity results in 1 notification, then 0 if the activity is deleted" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	assert Enum.empty?(Notification.for_user(user))

	{:ok, _, _} = CommonAPI.favorite(activity.id, other_user)

	assert length(Notification.for_user(user)) == 1

	{:ok, _} = CommonAPI.delete(activity.id, user)

	assert Enum.empty?(Notification.for_user(user))
	end

	test "liking an activity results in 1 notification, then 0 if the activity is unliked" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	assert Enum.empty?(Notification.for_user(user))

	{:ok, _, _} = CommonAPI.favorite(activity.id, other_user)

	assert length(Notification.for_user(user)) == 1

	{:ok, _, _, _} = CommonAPI.unfavorite(activity.id, other_user)

	assert Enum.empty?(Notification.for_user(user))
	end

	test "repeating an activity results in 1 notification, then 0 if the activity is deleted" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	assert Enum.empty?(Notification.for_user(user))

	{:ok, _, _} = CommonAPI.repeat(activity.id, other_user)

	assert length(Notification.for_user(user)) == 1

	{:ok, _} = CommonAPI.delete(activity.id, user)

	assert Enum.empty?(Notification.for_user(user))
	end

	test "repeating an activity results in 1 notification, then 0 if the activity is unrepeated" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	assert Enum.empty?(Notification.for_user(user))

	{:ok, _, _} = CommonAPI.repeat(activity.id, other_user)

	assert length(Notification.for_user(user)) == 1

	{:ok, _, _} = CommonAPI.unrepeat(activity.id, other_user)

	assert Enum.empty?(Notification.for_user(user))
	end

	test "liking an activity which is already deleted does not generate a notification" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	assert Enum.empty?(Notification.for_user(user))

	{:ok, _deletion_activity} = CommonAPI.delete(activity.id, user)

	assert Enum.empty?(Notification.for_user(user))

	{:error, _} = CommonAPI.favorite(activity.id, other_user)

	assert Enum.empty?(Notification.for_user(user))
	end

	test "repeating an activity which is already deleted does not generate a notification" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})

	assert Enum.empty?(Notification.for_user(user))

	{:ok, _deletion_activity} = CommonAPI.delete(activity.id, user)

	assert Enum.empty?(Notification.for_user(user))

	{:error, _} = CommonAPI.repeat(activity.id, other_user)

	assert Enum.empty?(Notification.for_user(user))
	end

	test "replying to a deleted post without tagging does not generate a notification" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "test post"})
	{:ok, _deletion_activity} = CommonAPI.delete(activity.id, user)

	{:ok, _reply_activity} =
	CommonAPI.post(other_user, %{
	"status" => "test reply",
	"in_reply_to_status_id" => activity.id
	})

	assert Enum.empty?(Notification.for_user(user))
	end

	test "notifications are deleted if a local user is deleted" do
	user = insert(:user)
	other_user = insert(:user)

	{:ok, _activity} =
	CommonAPI.post(user, %{"status" => "hi @#{other_user.nickname}", "visibility" => "direct"})

	refute Enum.empty?(Notification.for_user(other_user))

	{:ok, job} = User.delete(user)
	ObanHelpers.perform(job)

	assert Enum.empty?(Notification.for_user(other_user))
	end

	test "notifications are deleted if a remote user is deleted" do
	remote_user = insert(:user)
	local_user = insert(:user)

	dm_message = %{
	"@context" => "https://www.w3.org/ns/activitystreams",
	"type" => "Create",
	"actor" => remote_user.ap_id,
	"id" => remote_user.ap_id <> "/activities/test",
	"to" => [local_user.ap_id],
	"cc" => [],
	"object" => %{
	"type" => "Note",
	"content" => "Hello!",
	"tag" => [
	%{
	"type" => "Mention",
	"href" => local_user.ap_id,
	"name" => "@#{local_user.nickname}"
	}
	],
	"to" => [local_user.ap_id],
	"cc" => [],
	"attributedTo" => remote_user.ap_id
	}
	}

	{:ok, _dm_activity} = Transmogrifier.handle_incoming(dm_message)

	refute Enum.empty?(Notification.for_user(local_user))

	delete_user_message = %{
	"@context" => "https://www.w3.org/ns/activitystreams",
	"id" => remote_user.ap_id <> "/activities/delete",
	"actor" => remote_user.ap_id,
	"type" => "Delete",
	"object" => remote_user.ap_id
	}

	{:ok, _delete_activity} = Transmogrifier.handle_incoming(delete_user_message)
	ObanHelpers.perform_all()

	assert Enum.empty?(Notification.for_user(local_user))
	end

	test "move activity generates a notification" do
	%{ap_id: old_ap_id} = old_user = insert(:user)
	%{ap_id: new_ap_id} = new_user = insert(:user, also_known_as: [old_ap_id])
	follower = insert(:user)
	other_follower = insert(:user, %{allow_following_move: false})

	User.follow(follower, old_user)
	User.follow(other_follower, old_user)

	Pleroma.Web.ActivityPub.ActivityPub.move(old_user, new_user)
	ObanHelpers.perform_all()

	assert [] = Notification.for_user(follower)

	assert [
	%{
	activity: %{
	data: %{"type" => "Move", "actor" => ^old_ap_id, "target" => ^new_ap_id}
	}
	}
	] = Notification.for_user(follower, %{with_move: true})

	assert [] = Notification.for_user(other_follower)

	assert [
	%{
	activity: %{
	data: %{"type" => "Move", "actor" => ^old_ap_id, "target" => ^new_ap_id}
	}
	}
	] = Notification.for_user(other_follower, %{with_move: true})
	end
	end

	describe "for_user" do
	test "it returns notifications for muted user without notifications" do
	user = insert(:user)
	muted = insert(:user)
	{:ok, _user_relationships} = User.mute(user, muted, false)

	{:ok, _activity} = CommonAPI.post(muted, %{"status" => "hey @#{user.nickname}"})

	assert length(Notification.for_user(user)) == 1
	end

	test "it doesn't return notifications for muted user with notifications" do
	user = insert(:user)
	muted = insert(:user)
	{:ok, _user_relationships} = User.mute(user, muted)

	{:ok, _activity} = CommonAPI.post(muted, %{"status" => "hey @#{user.nickname}"})

	assert Notification.for_user(user) == []
	end

	test "it doesn't return notifications for blocked user" do
	user = insert(:user)
	blocked = insert(:user)
	{:ok, _user_relationship} = User.block(user, blocked)

	{:ok, _activity} = CommonAPI.post(blocked, %{"status" => "hey @#{user.nickname}"})

	assert Notification.for_user(user) == []
	end

	test "it doesn't return notifications for blocked domain" do
	user = insert(:user)
	blocked = insert(:user, ap_id: "http://some-domain.com")
	{:ok, user} = User.block_domain(user, "some-domain.com")

	{:ok, _activity} = CommonAPI.post(blocked, %{"status" => "hey @#{user.nickname}"})

	assert Notification.for_user(user) == []
	end

	test "it doesn't return notifications for muted thread" do
	user = insert(:user)
	another_user = insert(:user)

	{:ok, activity} = CommonAPI.post(another_user, %{"status" => "hey @#{user.nickname}"})

	{:ok, _} = Pleroma.ThreadMute.add_mute(user.id, activity.data["context"])
	assert Notification.for_user(user) == []
	end

	test "it returns notifications from a muted user when with_muted is set" do
	user = insert(:user)
	muted = insert(:user)
	{:ok, _user_relationships} = User.mute(user, muted)

	{:ok, _activity} = CommonAPI.post(muted, %{"status" => "hey @#{user.nickname}"})

	assert length(Notification.for_user(user, %{with_muted: true})) == 1
	end

	test "it doesn't return notifications from a blocked user when with_muted is set" do
	user = insert(:user)
	blocked = insert(:user)
	{:ok, _user_relationship} = User.block(user, blocked)

	{:ok, _activity} = CommonAPI.post(blocked, %{"status" => "hey @#{user.nickname}"})

	assert Enum.empty?(Notification.for_user(user, %{with_muted: true}))
	end

	test "it doesn't return notifications from a domain-blocked user when with_muted is set" do
	user = insert(:user)
	blocked = insert(:user, ap_id: "http://some-domain.com")
	{:ok, user} = User.block_domain(user, "some-domain.com")

	{:ok, _activity} = CommonAPI.post(blocked, %{"status" => "hey @#{user.nickname}"})

	assert Enum.empty?(Notification.for_user(user, %{with_muted: true}))
	end

	test "it returns notifications from muted threads when with_muted is set" do
	user = insert(:user)
	another_user = insert(:user)

	{:ok, activity} = CommonAPI.post(another_user, %{"status" => "hey @#{user.nickname}"})

	{:ok, _} = Pleroma.ThreadMute.add_mute(user.id, activity.data["context"])
	assert length(Notification.for_user(user, %{with_muted: true})) == 1
	end
	end
	end
	diff --git a/test/web/streamer/streamer_test.exs b/test/web/streamer/streamer_test.exs
	index 720f8fa44..cbd5ec462 100644
	--- a/test/web/streamer/streamer_test.exs
	+++ b/test/web/streamer/streamer_test.exs
	@@ -1,617 +1,683 @@
	# Pleroma: A lightweight social networking server
	# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
	# SPDX-License-Identifier: AGPL-3.0-only

	defmodule Pleroma.Web.StreamerTest do
	use Pleroma.DataCase

	import Pleroma.Factory

	alias Pleroma.Conversation.Participation
	alias Pleroma.List
	alias Pleroma.User
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.Streamer
	alias Pleroma.Web.Streamer.StreamerSocket
	alias Pleroma.Web.Streamer.Worker

	@moduletag needs_streamer: true, capture_log: true

	- @streamer_timeout 150
	+ @streamer_timeout 300
	@streamer_start_wait 10

	clear_config([:instance, :skip_thread_containment])

	+ describe "get_topic without an user" do
	+ test "allows public" do
	+ assert {:ok, "public"} = Streamer.get_topic("public", nil)
	+ assert {:ok, "public:local"} = Streamer.get_topic("public:local", nil)
	+ assert {:ok, "public:media"} = Streamer.get_topic("public:media", nil)
	+ assert {:ok, "public:local:media"} = Streamer.get_topic("public:local:media", nil)
	+ end
	+
	+ test "allows hashtag streams" do
	+ assert {:ok, "hashtag:cofe"} = Streamer.get_topic("hashtag", nil, %{"tag" => "cofe"})
	+ end
	+
	+ test "disallows user streams" do
	+ assert {:error, _} = Streamer.get_topic("user", nil)
	+ assert {:error, _} = Streamer.get_topic("user:notification", nil)
	+ assert {:error, _} = Streamer.get_topic("direct", nil)
	+ end
	+
	+ test "disallows list streams" do
	+ assert {:error, _} = Streamer.get_topic("list", nil, %{"list" => 42})
	+ end
	+ end
	+
	+ describe "get_topic with an user" do
	+ setup do
	+ user = insert(:user)
	+ {:ok, %{user: user}}
	+ end
	+
	+ test "allows public streams", %{user: user} do
	+ assert {:ok, "public"} = Streamer.get_topic("public", user)
	+ assert {:ok, "public:local"} = Streamer.get_topic("public:local", user)
	+ assert {:ok, "public:media"} = Streamer.get_topic("public:media", user)
	+ assert {:ok, "public:local:media"} = Streamer.get_topic("public:local:media", user)
	+ end
	+
	+ test "allows user streams", %{user: user} do
	+ expected_user_topic = "user:#{user.id}"
	+ expected_notif_topic = "user:notification:#{user.id}"
	+ expected_direct_topic = "direct:#{user.id}"
	+ assert {:ok, ^expected_user_topic} = Streamer.get_topic("user", user)
	+ assert {:ok, ^expected_notif_topic} = Streamer.get_topic("user:notification", user)
	+ assert {:ok, ^expected_direct_topic} = Streamer.get_topic("direct", user)
	+ end
	+
	+ test "allows hashtag streams", %{user: user} do
	+ assert {:ok, "hashtag:cofe"} = Streamer.get_topic("hashtag", user, %{"tag" => "cofe"})
	+ end
	+
	+ test "disallows registering to an user stream", %{user: user} do
	+ another_user = insert(:user)
	+ assert {:error, _} = Streamer.get_topic("user:#{another_user.id}", user)
	+ assert {:error, _} = Streamer.get_topic("user:notification:#{another_user.id}", user)
	+ assert {:error, _} = Streamer.get_topic("direct:#{another_user.id}", user)
	+ end
	+
	+ test "allows list stream that are owned by the user", %{user: user} do
	+ {:ok, list} = List.create("Test", user)
	+ assert {:error, _} = Streamer.get_topic("list:#{list.id}", user)
	+ assert {:ok, _} = Streamer.get_topic("list", user, %{"list" => list.id})
	+ end
	+
	+ test "disallows list stream that are not owned by the user", %{user: user} do
	+ another_user = insert(:user)
	+ {:ok, list} = List.create("Test", another_user)
	+ assert {:error, _} = Streamer.get_topic("list:#{list.id}", user)
	+ assert {:error, _} = Streamer.get_topic("list", user, %{"list" => list.id})
	+ end
	+ end
	+
	describe "user streams" do
	setup do
	user = insert(:user)
	notify = insert(:notification, user: user, activity: build(:note_activity))
	{:ok, %{user: user, notify: notify}}
	end

	test "it sends notify to in the 'user' stream", %{user: user, notify: notify} do
	task =
	Task.async(fn ->
	assert_receive {:text, _}, @streamer_timeout
	end)

	- Streamer.add_socket(
	+ Streamer.get_topic_and_add_socket(
	"user",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	Streamer.stream("user", notify)
	Task.await(task)
	end

	test "it sends notify to in the 'user:notification' stream", %{user: user, notify: notify} do
	task =
	Task.async(fn ->
	assert_receive {:text, _}, @streamer_timeout
	end)

	- Streamer.add_socket(
	+ Streamer.get_topic_and_add_socket(
	"user:notification",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	Streamer.stream("user:notification", notify)
	Task.await(task)
	end

	test "it doesn't send notify to the 'user:notification' stream when a user is blocked", %{
	user: user
	} do
	blocked = insert(:user)
	{: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, _}, @streamer_timeout end)

	- Streamer.add_socket(
	+ Streamer.get_topic_and_add_socket(
	"user:notification",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	Streamer.stream("user:notification", notif)
	Task.await(task)
	end

	test "it doesn't send notify to the 'user:notification' stream when a thread is muted", %{
	user: user
	} do
	user2 = insert(:user)

	{: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(
	+ Streamer.get_topic_and_add_socket(
	"user:notification",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	Streamer.stream("user:notification", notif)
	Task.await(task)
	end

	test "it doesn't send notify to the 'user:notification' stream' when a domain is blocked", %{
	user: user
	} do
	user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})

	{: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(
	+ Streamer.get_topic_and_add_socket(
	"user:notification",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	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(
	+ Streamer.get_topic_and_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
	user = insert(:user)
	other_user = insert(:user)

	task =
	Task.async(fn ->
	assert_receive {:text, _}, @streamer_timeout
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user
	}

	{:ok, activity} = CommonAPI.post(other_user, %{"status" => "Test"})

	topics = %{
	"public" => [fake_socket]
	}

	Worker.push_to_socket(topics, "public", activity)

	Task.await(task)

	task =
	Task.async(fn ->
	expected_event =
	%{
	"event" => "delete",
	"payload" => activity.id
	}
	\|> Jason.encode!()

	assert_receive {:text, received_event}, @streamer_timeout
	assert received_event == expected_event
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user
	}

	{:ok, activity} = CommonAPI.delete(activity.id, other_user)

	topics = %{
	"public" => [fake_socket]
	}

	Worker.push_to_socket(topics, "public", activity)

	Task.await(task)
	end

	describe "thread_containment" 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)
	User.follow(user, author, :follow_accept)

	activity =
	insert(:note_activity,
	note:
	insert(:note,
	user: author,
	data: %{"to" => ["TEST-FFF"]}
	)
	)

	task = Task.async(fn -> refute_receive {:text, _}, 1_000 end)
	fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
	topics = %{"public" => [fake_socket]}
	Worker.push_to_socket(topics, "public", activity)

	Task.await(task)
	end

	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)
	User.follow(user, author, :follow_accept)

	activity =
	insert(:note_activity,
	note:
	insert(:note,
	user: author,
	data: %{"to" => ["TEST-FFF"]}
	)
	)

	task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
	fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
	topics = %{"public" => [fake_socket]}
	Worker.push_to_socket(topics, "public", activity)

	Task.await(task)
	end

	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, skip_thread_containment: true)
	User.follow(user, author, :follow_accept)

	activity =
	insert(:note_activity,
	note:
	insert(:note,
	user: author,
	data: %{"to" => ["TEST-FFF"]}
	)
	)

	task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
	fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
	topics = %{"public" => [fake_socket]}
	Worker.push_to_socket(topics, "public", activity)

	Task.await(task)
	end
	end

	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)

	{:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})

	task =
	Task.async(fn ->
	refute_receive {:text, _}, 1_000
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user
	}

	topics = %{
	"public" => [fake_socket]
	}

	Worker.push_to_socket(topics, "public", activity)

	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
	user_a = insert(:user)
	user_b = insert(:user)
	user_c = insert(:user)

	{:ok, user_a} = User.follow(user_a, user_b)

	{: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
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user_a
	}

	topics = %{
	"list:#{list.id}" => [fake_socket]
	}

	Worker.handle_call({:stream, "list", activity}, self(), topics)

	Task.await(task)
	end

	test "it doesn't send unwanted private posts to list" do
	user_a = insert(:user)
	user_b = insert(:user)

	{: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
	end)

	fake_socket = %StreamerSocket{
	transport_pid: task.pid,
	user: user_a
	}

	topics = %{
	"list:#{list.id}" => [fake_socket]
	}

	Worker.handle_call({:stream, "list", activity}, self(), topics)

	Task.await(task)
	end

	test "it sends wanted private posts to list" do
	user_a = insert(:user)
	user_b = insert(:user)

	{:ok, user_a} = User.follow(user_a, user_b)

	{: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
	end)

	- fake_socket = %StreamerSocket{
	- transport_pid: task.pid,
	- user: user_a
	- }
	-
	- Streamer.add_socket(
	- "list:#{list.id}",
	- fake_socket
	+ Streamer.get_topic_and_add_socket(
	+ "list",
	+ %{transport_pid: task.pid, assigns: %{user: user_a}},
	+ %{"list" => list.id}
	)

	Worker.handle_call({:stream, "list", activity}, self(), %{})

	Task.await(task)
	end

	test "it doesn't send muted reblogs" 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)

	task =
	Task.async(fn ->
	refute_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)

	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, 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", favorite_activity)

	Task.await(task)
	end

	test "it doesn't send posts from muted threads" do
	user = insert(:user)
	user2 = insert(:user)
	{:ok, user2, user, _activity} = CommonAPI.follow(user2, user)

	{:ok, activity} = CommonAPI.post(user, %{"status" => "super hot take"})

	{:ok, activity} = CommonAPI.add_mute(user2, activity)

	task = Task.async(fn -> refute_receive {:text, _}, @streamer_timeout end)

	- Streamer.add_socket(
	+ Streamer.get_topic_and_add_socket(
	"user",
	%{transport_pid: task.pid, assigns: %{user: user2}}
	)

	Streamer.stream("user", activity)
	Task.await(task)
	end

	describe "direct streams" do
	setup do
	:ok
	end

	test "it sends conversation update to the 'direct' stream", %{} do
	user = insert(:user)
	another_user = insert(:user)

	task =
	Task.async(fn ->
	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(
	+ Streamer.get_topic_and_add_socket(
	"direct",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	{:ok, _create_activity} =
	CommonAPI.post(another_user, %{
	"status" => "hey @#{user.nickname}",
	"visibility" => "direct"
	})

	Task.await(task)
	end

	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)

	{:ok, create_activity} =
	CommonAPI.post(another_user, %{
	"status" => "hi @#{user.nickname}",
	"visibility" => "direct"
	})

	task =
	Task.async(fn ->
	assert_receive {:text, received_event}, @streamer_timeout
	assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)

	refute_receive {:text, _}, @streamer_timeout
	end)

	Process.sleep(@streamer_start_wait)

	- Streamer.add_socket(
	+ Streamer.get_topic_and_add_socket(
	"direct",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	{:ok, _} = CommonAPI.delete(create_activity.id, another_user)

	Task.await(task)
	end

	test "it sends conversation update to the 'direct' stream when a message is deleted" do
	user = insert(:user)
	another_user = insert(:user)

	{:ok, create_activity} =
	CommonAPI.post(another_user, %{
	"status" => "hi @#{user.nickname}",
	"visibility" => "direct"
	})

	{:ok, create_activity2} =
	CommonAPI.post(another_user, %{
	"status" => "hi @#{user.nickname}",
	"in_reply_to_status_id" => create_activity.id,
	"visibility" => "direct"
	})

	task =
	Task.async(fn ->
	assert_receive {:text, received_event}, @streamer_timeout
	assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)

	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)
	assert last_status["id"] == to_string(create_activity.id)
	end)

	Process.sleep(@streamer_start_wait)

	- Streamer.add_socket(
	+ Streamer.get_topic_and_add_socket(
	"direct",
	%{transport_pid: task.pid, assigns: %{user: user}}
	)

	{:ok, _} = CommonAPI.delete(create_activity2.id, another_user)

	Task.await(task)
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Sat, Aug 29, 8:05 AM (1 d, 17 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1736259
Default Alt Text: (68 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions