No OneTemporary
Actions

Size

33 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/pleroma/web/streamer/streamer.ex b/lib/pleroma/web/streamer/streamer.ex
	index 73ee3e1e1..d1d70e556 100644
	--- a/lib/pleroma/web/streamer/streamer.ex
	+++ b/lib/pleroma/web/streamer/streamer.ex
	@@ -1,291 +1,295 @@
	# 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
	require Logger

	alias Pleroma.Activity
	alias Pleroma.Chat.MessageReference
	alias Pleroma.Config
	alias Pleroma.Conversation.Participation
	alias Pleroma.Notification
	alias Pleroma.Object
	alias Pleroma.User
	alias Pleroma.Web.ActivityPub.ActivityPub
	alias Pleroma.Web.ActivityPub.Visibility
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.StreamerView

	@mix_env Mix.env()
	@registry Pleroma.Web.StreamerRegistry

	def registry, do: @registry

	@public_streams ["public", "public:local", "public:media", "public:local:media"]
	@user_streams ["user", "user:notification", "direct", "user:pleroma_chat"]

	@doc "Expands and authorizes a stream, and registers the process for streaming."
	@spec get_topic_and_add_socket(stream :: String.t(), User.t() \| nil, Map.t() \| nil) ::
	{:ok, topic :: String.t()} \| {:error, :bad_topic} \| {:error, :unauthorized}
	def get_topic_and_add_socket(stream, user, params \\ %{}) do
	case get_topic(stream, user, params) do
	{:ok, topic} -> add_socket(topic, user)
	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

	@doc "Registers the process for streaming. Use `get_topic/3` to get the full authorized topic."
	def add_socket(topic, user) do
	if should_env_send?() do
	auth? = if user, do: true
	Registry.register(@registry, topic, auth?)
	end

	{:ok, topic}
	end

	def remove_socket(topic) do
	if should_env_send?(), do: Registry.unregister(@registry, topic)
	end

	def stream(topics, items) do
	if should_env_send?() do
	List.wrap(topics)
	\|> Enum.each(fn topic ->
	List.wrap(items)
	\|> Enum.each(fn item ->
	spawn(fn -> do_stream(topic, item) end)
	end)
	end)
	end

	:ok
	end

	- def filtered_by_user?(%User{} = user, %Activity{} = item) do
	+ def filtered_by_user?(user, item, streamed_type \\ :activity)
	+
	+ def filtered_by_user?(%User{} = user, %Activity{} = item, streamed_type) do
	%{block: blocked_ap_ids, mute: muted_ap_ids, reblog_mute: reblog_muted_ap_ids} =
	User.outgoing_relationships_ap_ids(user, [:block, :mute, :reblog_mute])

	recipient_blocks = MapSet.new(blocked_ap_ids ++ muted_ap_ids)
	recipients = MapSet.new(item.recipients)
	domain_blocks = Pleroma.Web.ActivityPub.MRF.subdomains_regex(user.domain_blocks)

	with parent <- Object.normalize(item) \|\| item,
	true <-
	Enum.all?([blocked_ap_ids, muted_ap_ids], &(item.actor not in &1)),
	true <- item.data["type"] != "Announce" \|\| item.actor not in reblog_muted_ap_ids,
	- true <- !(item.data["type"] == "Announce" && parent.data["actor"] == user.ap_id),
	+ true <-
	+ !(streamed_type == :activity && item.data["type"] == "Announce" &&
	+ parent.data["actor"] == user.ap_id),
	true <- Enum.all?([blocked_ap_ids, muted_ap_ids], &(parent.data["actor"] not in &1)),
	true <- MapSet.disjoint?(recipients, recipient_blocks),
	%{host: item_host} <- URI.parse(item.actor),
	%{host: parent_host} <- URI.parse(parent.data["actor"]),
	false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, item_host),
	false <- Pleroma.Web.ActivityPub.MRF.subdomain_match?(domain_blocks, parent_host),
	true <- thread_containment(item, user),
	false <- CommonAPI.thread_muted?(user, parent) do
	false
	else
	_ -> true
	end
	end

	- def filtered_by_user?(%User{} = user, %Notification{activity: activity}) do
	- filtered_by_user?(user, activity)
	+ def filtered_by_user?(%User{} = user, %Notification{activity: activity}, _) do
	+ filtered_by_user?(user, activity, :notification)
	end

	defp do_stream("direct", item) do
	recipient_topics =
	User.get_recipients_from_activity(item)
	\|> Enum.map(fn %{id: id} -> "direct:#{id}" end)

	Enum.each(recipient_topics, fn user_topic ->
	Logger.debug("Trying to push direct message to #{user_topic}\n\n")
	push_to_socket(user_topic, item)
	end)
	end

	defp do_stream("participation", participation) do
	user_topic = "direct:#{participation.user_id}"
	Logger.debug("Trying to push a conversation participation to #{user_topic}\n\n")

	push_to_socket(user_topic, participation)
	end

	defp do_stream("list", item) do
	# filter the recipient list if the activity is not public, see #270.
	recipient_lists =
	case Visibility.is_public?(item) do
	true ->
	Pleroma.List.get_lists_from_activity(item)

	_ ->
	Pleroma.List.get_lists_from_activity(item)
	\|> Enum.filter(fn list ->
	owner = User.get_cached_by_id(list.user_id)

	Visibility.visible_for_user?(item, owner)
	end)
	end

	recipient_topics =
	recipient_lists
	\|> Enum.map(fn %{id: id} -> "list:#{id}" end)

	Enum.each(recipient_topics, fn list_topic ->
	Logger.debug("Trying to push message to #{list_topic}\n\n")
	push_to_socket(list_topic, item)
	end)
	end

	defp do_stream(topic, %Notification{} = item)
	when topic in ["user", "user:notification"] do
	Registry.dispatch(@registry, "#{topic}:#{item.user_id}", fn list ->
	Enum.each(list, fn {pid, _auth} ->
	send(pid, {:render_with_user, StreamerView, "notification.json", item})
	end)
	end)
	end

	defp do_stream(topic, {user, %MessageReference{} = cm_ref})
	when topic in ["user", "user:pleroma_chat"] do
	topic = "#{topic}:#{user.id}"

	text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref})

	Registry.dispatch(@registry, topic, fn list ->
	Enum.each(list, fn {pid, _auth} ->
	send(pid, {:text, text})
	end)
	end)
	end

	defp do_stream("user", item) do
	Logger.debug("Trying to push to users")

	recipient_topics =
	User.get_recipients_from_activity(item)
	\|> Enum.map(fn %{id: id} -> "user:#{id}" end)

	Enum.each(recipient_topics, fn topic ->
	push_to_socket(topic, item)
	end)
	end

	defp do_stream(topic, item) do
	Logger.debug("Trying to push to #{topic}")
	Logger.debug("Pushing item to #{topic}")
	push_to_socket(topic, item)
	end

	defp push_to_socket(topic, %Participation{} = participation) do
	rendered = StreamerView.render("conversation.json", participation)

	Registry.dispatch(@registry, topic, fn list ->
	Enum.each(list, fn {pid, _} ->
	send(pid, {:text, rendered})
	end)
	end)
	end

	defp push_to_socket(topic, %Activity{
	data: %{"type" => "Delete", "deleted_activity_id" => deleted_activity_id}
	}) do
	rendered = Jason.encode!(%{event: "delete", payload: to_string(deleted_activity_id)})

	Registry.dispatch(@registry, topic, fn list ->
	Enum.each(list, fn {pid, _} ->
	send(pid, {:text, rendered})
	end)
	end)
	end

	defp push_to_socket(_topic, %Activity{data: %{"type" => "Delete"}}), do: :noop

	defp push_to_socket(topic, item) do
	anon_render = StreamerView.render("update.json", item)

	Registry.dispatch(@registry, topic, fn list ->
	Enum.each(list, fn {pid, auth?} ->
	if auth? do
	send(pid, {:render_with_user, StreamerView, "update.json", item})
	else
	send(pid, {:text, anon_render})
	end
	end)
	end)
	end

	defp thread_containment(_activity, %User{skip_thread_containment: true}), do: true

	defp thread_containment(activity, user) do
	if Config.get([:instance, :skip_thread_containment]) do
	true
	else
	ActivityPub.contain_activity(activity, user)
	end
	end

	# In test environement, only return true if the registry is started.
	# In benchmark environment, returns false.
	# In any other environment, always returns true.
	cond do
	@mix_env == :test ->
	def should_env_send? do
	case Process.whereis(@registry) do
	nil ->
	false

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

	@mix_env == :benchmark ->
	def should_env_send?, do: false

	true ->
	def should_env_send?, do: true
	end
	end
	diff --git a/test/web/streamer/streamer_test.exs b/test/web/streamer/streamer_test.exs
	index dfe341b34..d56d74464 100644
	--- a/test/web/streamer/streamer_test.exs
	+++ b/test/web/streamer/streamer_test.exs
	@@ -1,654 +1,671 @@
	# 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.Chat
	alias Pleroma.Chat.MessageReference
	alias Pleroma.Conversation.Participation
	alias Pleroma.List
	alias Pleroma.Object
	alias Pleroma.User
	alias Pleroma.Web.CommonAPI
	alias Pleroma.Web.Streamer
	alias Pleroma.Web.StreamerView

	@moduletag needs_streamer: true, capture_log: true

	setup do: 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 streams the user's post in the 'user' stream", %{user: user} do
	Streamer.get_topic_and_add_socket("user", user)
	{:ok, activity} = CommonAPI.post(user, %{status: "hey"})
	assert_receive {:render_with_user, _, _, ^activity}
	refute Streamer.filtered_by_user?(user, activity)
	end

	test "it streams boosts of the user in the 'user' stream", %{user: user} do
	Streamer.get_topic_and_add_socket("user", user)

	other_user = insert(:user)
	{:ok, activity} = CommonAPI.post(other_user, %{status: "hey"})
	{:ok, announce} = CommonAPI.repeat(activity.id, user)

	assert_receive {:render_with_user, Pleroma.Web.StreamerView, "update.json", ^announce}
	refute Streamer.filtered_by_user?(user, announce)
	end

	test "it does not stream announces of the user's own posts in the 'user' stream", %{
	user: user
	} do
	Streamer.get_topic_and_add_socket("user", user)

	other_user = insert(:user)
	{:ok, activity} = CommonAPI.post(user, %{status: "hey"})
	{:ok, announce} = CommonAPI.repeat(activity.id, other_user)

	assert Streamer.filtered_by_user?(user, announce)
	end

	+ test "it does stream notifications announces of the user's own posts in the 'user' stream", %{
	+ user: user
	+ } do
	+ Streamer.get_topic_and_add_socket("user", user)
	+
	+ other_user = insert(:user)
	+ {:ok, activity} = CommonAPI.post(user, %{status: "hey"})
	+ {:ok, announce} = CommonAPI.repeat(activity.id, other_user)
	+
	+ notification =
	+ Pleroma.Notification
	+ \|> Repo.get_by(%{user_id: user.id, activity_id: announce.id})
	+ \|> Repo.preload(:activity)
	+
	+ refute Streamer.filtered_by_user?(user, notification)
	+ end
	+
	test "it streams boosts of mastodon user in the 'user' stream", %{user: user} do
	Streamer.get_topic_and_add_socket("user", user)

	other_user = insert(:user)
	{:ok, activity} = CommonAPI.post(other_user, %{status: "hey"})

	data =
	File.read!("test/fixtures/mastodon-announce.json")
	\|> Poison.decode!()
	\|> Map.put("object", activity.data["object"])
	\|> Map.put("actor", user.ap_id)

	{:ok, %Pleroma.Activity{data: _data, local: false} = announce} =
	Pleroma.Web.ActivityPub.Transmogrifier.handle_incoming(data)

	assert_receive {:render_with_user, Pleroma.Web.StreamerView, "update.json", ^announce}
	refute Streamer.filtered_by_user?(user, announce)
	end

	test "it sends notify to in the 'user' stream", %{user: user, notify: notify} do
	Streamer.get_topic_and_add_socket("user", user)
	Streamer.stream("user", notify)
	assert_receive {:render_with_user, _, _, ^notify}
	refute Streamer.filtered_by_user?(user, notify)
	end

	test "it sends notify to in the 'user:notification' stream", %{user: user, notify: notify} do
	Streamer.get_topic_and_add_socket("user:notification", user)
	Streamer.stream("user:notification", notify)
	assert_receive {:render_with_user, _, _, ^notify}
	refute Streamer.filtered_by_user?(user, notify)
	end

	test "it sends chat messages to the 'user:pleroma_chat' stream", %{user: user} do
	other_user = insert(:user)

	{:ok, create_activity} = CommonAPI.post_chat_message(other_user, user, "hey cirno")
	object = Object.normalize(create_activity, false)
	chat = Chat.get(user.id, other_user.ap_id)
	cm_ref = MessageReference.for_chat_and_object(chat, object)
	cm_ref = %{cm_ref \| chat: chat, object: object}

	Streamer.get_topic_and_add_socket("user:pleroma_chat", user)
	Streamer.stream("user:pleroma_chat", {user, cm_ref})

	text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref})

	assert text =~ "hey cirno"
	assert_receive {:text, ^text}
	end

	test "it sends chat messages to the 'user' stream", %{user: user} do
	other_user = insert(:user)

	{:ok, create_activity} = CommonAPI.post_chat_message(other_user, user, "hey cirno")
	object = Object.normalize(create_activity, false)
	chat = Chat.get(user.id, other_user.ap_id)
	cm_ref = MessageReference.for_chat_and_object(chat, object)
	cm_ref = %{cm_ref \| chat: chat, object: object}

	Streamer.get_topic_and_add_socket("user", user)
	Streamer.stream("user", {user, cm_ref})

	text = StreamerView.render("chat_update.json", %{chat_message_reference: cm_ref})

	assert text =~ "hey cirno"
	assert_receive {:text, ^text}
	end

	test "it sends chat message notifications to the 'user:notification' stream", %{user: user} do
	other_user = insert(:user)

	{:ok, create_activity} = CommonAPI.post_chat_message(other_user, user, "hey")

	notify =
	Repo.get_by(Pleroma.Notification, user_id: user.id, activity_id: create_activity.id)
	\|> Repo.preload(:activity)

	Streamer.get_topic_and_add_socket("user:notification", user)
	Streamer.stream("user:notification", notify)
	assert_receive {:render_with_user, _, _, ^notify}
	refute Streamer.filtered_by_user?(user, notify)
	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)

	Streamer.get_topic_and_add_socket("user:notification", user)

	{:ok, activity} = CommonAPI.post(user, %{status: ":("})
	{:ok, _} = CommonAPI.favorite(blocked, activity.id)

	refute_receive _
	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, _} = CommonAPI.add_mute(user, activity)

	Streamer.get_topic_and_add_socket("user:notification", user)

	{:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)

	refute_receive _
	assert Streamer.filtered_by_user?(user, favorite_activity)
	end

	test "it sends favorite to 'user:notification' stream'", %{
	user: user
	} do
	user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})

	{:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
	Streamer.get_topic_and_add_socket("user:notification", user)
	{:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)

	assert_receive {:render_with_user, _, "notification.json", notif}
	assert notif.activity.id == favorite_activity.id
	refute Streamer.filtered_by_user?(user, notif)
	end

	test "it doesn't send 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"})
	Streamer.get_topic_and_add_socket("user:notification", user)
	{:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)

	refute_receive _
	assert Streamer.filtered_by_user?(user, favorite_activity)
	end

	test "it sends follow activities to the 'user:notification' stream", %{
	user: user
	} do
	user_url = user.ap_id
	user2 = insert(:user)

	body =
	File.read!("test/fixtures/users_mock/localhost.json")
	\|> String.replace("{{nickname}}", user.nickname)
	\|> Jason.encode!()

	Tesla.Mock.mock_global(fn
	%{method: :get, url: ^user_url} ->
	%Tesla.Env{status: 200, body: body}
	end)

	Streamer.get_topic_and_add_socket("user:notification", user)
	{:ok, _follower, _followed, follow_activity} = CommonAPI.follow(user2, user)

	assert_receive {:render_with_user, _, "notification.json", notif}
	assert notif.activity.id == follow_activity.id
	refute Streamer.filtered_by_user?(user, notif)
	end
	end

	test "it sends to public authenticated" do
	user = insert(:user)
	other_user = insert(:user)

	Streamer.get_topic_and_add_socket("public", other_user)

	{:ok, activity} = CommonAPI.post(user, %{status: "Test"})
	assert_receive {:render_with_user, _, _, ^activity}
	refute Streamer.filtered_by_user?(user, activity)
	end

	test "works for deletions" do
	user = insert(:user)
	other_user = insert(:user)
	{:ok, activity} = CommonAPI.post(other_user, %{status: "Test"})

	Streamer.get_topic_and_add_socket("public", user)

	{:ok, _} = CommonAPI.delete(activity.id, other_user)
	activity_id = activity.id
	assert_receive {:text, event}
	assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
	end

	test "it sends to public unauthenticated" do
	user = insert(:user)

	Streamer.get_topic_and_add_socket("public", nil)

	{:ok, activity} = CommonAPI.post(user, %{status: "Test"})
	activity_id = activity.id
	assert_receive {:text, event}
	assert %{"event" => "update", "payload" => payload} = Jason.decode!(event)
	assert %{"id" => ^activity_id} = Jason.decode!(payload)

	{:ok, _} = CommonAPI.delete(activity.id, user)
	assert_receive {:text, event}
	assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
	end

	describe "thread_containment" do
	test "it filters 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"]}
	)
	)

	Streamer.get_topic_and_add_socket("public", user)
	Streamer.stream("public", activity)
	assert_receive {:render_with_user, _, _, ^activity}
	assert Streamer.filtered_by_user?(user, activity)
	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"]}
	)
	)

	Streamer.get_topic_and_add_socket("public", user)
	Streamer.stream("public", activity)

	assert_receive {:render_with_user, _, _, ^activity}
	refute Streamer.filtered_by_user?(user, activity)
	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"]}
	)
	)

	Streamer.get_topic_and_add_socket("public", user)
	Streamer.stream("public", activity)

	assert_receive {:render_with_user, _, _, ^activity}
	refute Streamer.filtered_by_user?(user, activity)
	end
	end

	describe "blocks" do
	test "it filters messages involving blocked users" do
	user = insert(:user)
	blocked_user = insert(:user)
	{:ok, _user_relationship} = User.block(user, blocked_user)

	Streamer.get_topic_and_add_socket("public", user)
	{:ok, activity} = CommonAPI.post(blocked_user, %{status: "Test"})
	assert_receive {:render_with_user, _, _, ^activity}
	assert Streamer.filtered_by_user?(user, activity)
	end

	test "it filters messages transitively involving blocked users" do
	blocker = insert(:user)
	blockee = insert(:user)
	friend = insert(:user)

	Streamer.get_topic_and_add_socket("public", blocker)

	{:ok, _user_relationship} = User.block(blocker, blockee)

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

	assert_receive {:render_with_user, _, _, ^activity_one}
	assert Streamer.filtered_by_user?(blocker, activity_one)

	{:ok, activity_two} = CommonAPI.post(blockee, %{status: "hey! @#{friend.nickname}"})

	assert_receive {:render_with_user, _, _, ^activity_two}
	assert Streamer.filtered_by_user?(blocker, activity_two)

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

	assert_receive {:render_with_user, _, _, ^activity_three}
	assert Streamer.filtered_by_user?(blocker, activity_three)
	end
	end

	describe "lists" do
	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)

	Streamer.get_topic_and_add_socket("list", user_a, %{"list" => list.id})

	{:ok, _activity} =
	CommonAPI.post(user_b, %{
	status: "@#{user_c.nickname} Test",
	visibility: "direct"
	})

	refute_receive _
	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)

	Streamer.get_topic_and_add_socket("list", user_a, %{"list" => list.id})

	{:ok, _activity} =
	CommonAPI.post(user_b, %{
	status: "Test",
	visibility: "private"
	})

	refute_receive _
	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)

	Streamer.get_topic_and_add_socket("list", user_a, %{"list" => list.id})

	{:ok, activity} =
	CommonAPI.post(user_b, %{
	status: "Test",
	visibility: "private"
	})

	assert_receive {:render_with_user, _, _, ^activity}
	refute Streamer.filtered_by_user?(user_a, activity)
	end
	end

	describe "muted reblogs" do
	test "it filters muted reblogs" do
	user1 = insert(:user)
	user2 = insert(:user)
	user3 = insert(:user)
	CommonAPI.follow(user1, user2)
	CommonAPI.hide_reblogs(user1, user2)

	{:ok, create_activity} = CommonAPI.post(user3, %{status: "I'm kawen"})

	Streamer.get_topic_and_add_socket("user", user1)
	{:ok, announce_activity} = CommonAPI.repeat(create_activity.id, user2)
	assert_receive {:render_with_user, _, _, ^announce_activity}
	assert Streamer.filtered_by_user?(user1, announce_activity)
	end

	test "it filters reblog notification for reblog-muted actors" do
	user1 = insert(:user)
	user2 = insert(:user)
	CommonAPI.follow(user1, user2)
	CommonAPI.hide_reblogs(user1, user2)

	{:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
	Streamer.get_topic_and_add_socket("user", user1)
	{:ok, _announce_activity} = CommonAPI.repeat(create_activity.id, user2)

	assert_receive {:render_with_user, _, "notification.json", notif}
	assert Streamer.filtered_by_user?(user1, notif)
	end

	test "it send non-reblog notification for reblog-muted actors" do
	user1 = insert(:user)
	user2 = insert(:user)
	CommonAPI.follow(user1, user2)
	CommonAPI.hide_reblogs(user1, user2)

	{:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
	Streamer.get_topic_and_add_socket("user", user1)
	{:ok, _favorite_activity} = CommonAPI.favorite(user2, create_activity.id)

	assert_receive {:render_with_user, _, "notification.json", notif}
	refute Streamer.filtered_by_user?(user1, notif)
	end
	end

	test "it filters posts from muted threads" do
	user = insert(:user)
	user2 = insert(:user)
	Streamer.get_topic_and_add_socket("user", user2)
	{:ok, user2, user, _activity} = CommonAPI.follow(user2, user)
	{:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
	{:ok, _} = CommonAPI.add_mute(user2, activity)
	assert_receive {:render_with_user, _, _, ^activity}
	assert Streamer.filtered_by_user?(user2, activity)
	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)

	Streamer.get_topic_and_add_socket("direct", user)

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

	assert_receive {:text, received_event}

	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

	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)

	Streamer.get_topic_and_add_socket("direct", user)

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

	create_activity_id = create_activity.id
	assert_receive {:render_with_user, _, _, ^create_activity}
	assert_receive {:text, received_conversation1}
	assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)

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

	assert_receive {:text, received_event}

	assert %{"event" => "delete", "payload" => ^create_activity_id} =
	Jason.decode!(received_event)

	refute_receive _
	end

	test "it sends conversation update to the 'direct' stream when a message is deleted" do
	user = insert(:user)
	another_user = insert(:user)
	Streamer.get_topic_and_add_socket("direct", 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} 2",
	in_reply_to_status_id: create_activity.id,
	visibility: "direct"
	})

	assert_receive {:render_with_user, _, _, ^create_activity}
	assert_receive {:render_with_user, _, _, ^create_activity2}
	assert_receive {:text, received_conversation1}
	assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
	assert_receive {:text, received_conversation1}
	assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)

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

	assert_receive {:text, received_event}
	assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)

	assert_receive {:text, received_event}

	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
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Sat, Aug 8, 12:18 PM (1 d, 10 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1723167
Default Alt Text: (33 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions