add changelog entry for MFM
[akkoma] / test / pleroma / web / streamer_test.exs
1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2021 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
4
5 defmodule Pleroma.Web.StreamerTest do
6 use Pleroma.DataCase, async: false
7
8 import Pleroma.Factory
9
10 alias Pleroma.Conversation.Participation
11 alias Pleroma.List
12 alias Pleroma.User
13 alias Pleroma.Web.CommonAPI
14 alias Pleroma.Web.Streamer
15
16 @moduletag needs_streamer: true, capture_log: true
17
18 setup do: clear_config([:instance, :skip_thread_containment])
19
20 describe "get_topic/_ (unauthenticated)" do
21 test "allows public" do
22 assert {:ok, "public"} = Streamer.get_topic("public", nil, nil)
23 assert {:ok, "public:local"} = Streamer.get_topic("public:local", nil, nil)
24 assert {:ok, "public:media"} = Streamer.get_topic("public:media", nil, nil)
25 assert {:ok, "public:local:media"} = Streamer.get_topic("public:local:media", nil, nil)
26 end
27
28 test "allows instance streams" do
29 assert {:ok, "public:remote:lain.com"} =
30 Streamer.get_topic("public:remote", nil, nil, %{"instance" => "lain.com"})
31
32 assert {:ok, "public:remote:media:lain.com"} =
33 Streamer.get_topic("public:remote:media", nil, nil, %{"instance" => "lain.com"})
34 end
35
36 test "allows hashtag streams" do
37 assert {:ok, "hashtag:cofe"} = Streamer.get_topic("hashtag", nil, nil, %{"tag" => "cofe"})
38 end
39
40 test "disallows user streams" do
41 assert {:error, _} = Streamer.get_topic("user", nil, nil)
42 assert {:error, _} = Streamer.get_topic("user:notification", nil, nil)
43 assert {:error, _} = Streamer.get_topic("direct", nil, nil)
44 end
45
46 test "disallows list streams" do
47 assert {:error, _} = Streamer.get_topic("list", nil, nil, %{"list" => 42})
48 end
49 end
50
51 describe "get_topic/_ (authenticated)" do
52 setup do: oauth_access(["read"])
53
54 test "allows public streams (regardless of OAuth token scopes)", %{
55 user: user,
56 token: read_oauth_token
57 } do
58 with oauth_token <- [nil, read_oauth_token] do
59 assert {:ok, "public"} = Streamer.get_topic("public", user, oauth_token)
60 assert {:ok, "public:local"} = Streamer.get_topic("public:local", user, oauth_token)
61 assert {:ok, "public:media"} = Streamer.get_topic("public:media", user, oauth_token)
62
63 assert {:ok, "public:local:media"} =
64 Streamer.get_topic("public:local:media", user, oauth_token)
65 end
66 end
67
68 test "allows user streams (with proper OAuth token scopes)", %{
69 user: user,
70 token: read_oauth_token
71 } do
72 %{token: read_notifications_token} = oauth_access(["read:notifications"], user: user)
73 %{token: read_statuses_token} = oauth_access(["read:statuses"], user: user)
74 %{token: badly_scoped_token} = oauth_access(["irrelevant:scope"], user: user)
75
76 expected_user_topic = "user:#{user.id}"
77 expected_notification_topic = "user:notification:#{user.id}"
78 expected_direct_topic = "direct:#{user.id}"
79
80 for valid_user_token <- [read_oauth_token, read_statuses_token] do
81 assert {:ok, ^expected_user_topic} = Streamer.get_topic("user", user, valid_user_token)
82
83 assert {:ok, ^expected_direct_topic} =
84 Streamer.get_topic("direct", user, valid_user_token)
85 end
86
87 for invalid_user_token <- [read_notifications_token, badly_scoped_token],
88 user_topic <- ["user", "direct"] do
89 assert {:error, :unauthorized} = Streamer.get_topic(user_topic, user, invalid_user_token)
90 end
91
92 for valid_notification_token <- [read_oauth_token, read_notifications_token] do
93 assert {:ok, ^expected_notification_topic} =
94 Streamer.get_topic("user:notification", user, valid_notification_token)
95 end
96
97 for invalid_notification_token <- [read_statuses_token, badly_scoped_token] do
98 assert {:error, :unauthorized} =
99 Streamer.get_topic("user:notification", user, invalid_notification_token)
100 end
101 end
102
103 test "allows hashtag streams (regardless of OAuth token scopes)", %{
104 user: user,
105 token: read_oauth_token
106 } do
107 for oauth_token <- [nil, read_oauth_token] do
108 assert {:ok, "hashtag:cofe"} =
109 Streamer.get_topic("hashtag", user, oauth_token, %{"tag" => "cofe"})
110 end
111 end
112
113 test "disallows registering to another user's stream", %{user: user, token: read_oauth_token} do
114 another_user = insert(:user)
115 assert {:error, _} = Streamer.get_topic("user:#{another_user.id}", user, read_oauth_token)
116
117 assert {:error, _} =
118 Streamer.get_topic("user:notification:#{another_user.id}", user, read_oauth_token)
119
120 assert {:error, _} = Streamer.get_topic("direct:#{another_user.id}", user, read_oauth_token)
121 end
122
123 test "allows list stream that are owned by the user (with `read` or `read:lists` scopes)", %{
124 user: user,
125 token: read_oauth_token
126 } do
127 %{token: read_lists_token} = oauth_access(["read:lists"], user: user)
128 %{token: invalid_token} = oauth_access(["irrelevant:scope"], user: user)
129 {:ok, list} = List.create("Test", user)
130
131 assert {:error, _} = Streamer.get_topic("list:#{list.id}", user, read_oauth_token)
132
133 for valid_token <- [read_oauth_token, read_lists_token] do
134 assert {:ok, _} = Streamer.get_topic("list", user, valid_token, %{"list" => list.id})
135 end
136
137 assert {:error, _} = Streamer.get_topic("list", user, invalid_token, %{"list" => list.id})
138 end
139
140 test "disallows list stream that are not owned by the user", %{user: user, token: oauth_token} do
141 another_user = insert(:user)
142 {:ok, list} = List.create("Test", another_user)
143
144 assert {:error, _} = Streamer.get_topic("list:#{list.id}", user, oauth_token)
145 assert {:error, _} = Streamer.get_topic("list", user, oauth_token, %{"list" => list.id})
146 end
147 end
148
149 describe "user streams" do
150 setup do
151 %{user: user, token: token} = oauth_access(["read"])
152 notify = insert(:notification, user: user, activity: build(:note_activity))
153 {:ok, %{user: user, notify: notify, token: token}}
154 end
155
156 test "it streams the user's post in the 'user' stream", %{user: user, token: oauth_token} do
157 Streamer.get_topic_and_add_socket("user", user, oauth_token)
158 {:ok, activity} = CommonAPI.post(user, %{status: "hey"})
159
160 stream_name = "user:#{user.id}"
161 assert_receive {:render_with_user, _, _, ^activity, ^stream_name}
162 refute Streamer.filtered_by_user?(user, activity)
163 end
164
165 test "it streams boosts of the user in the 'user' stream", %{user: user, token: oauth_token} do
166 Streamer.get_topic_and_add_socket("user", user, oauth_token)
167
168 other_user = insert(:user)
169 {:ok, activity} = CommonAPI.post(other_user, %{status: "hey"})
170 {:ok, announce} = CommonAPI.repeat(activity.id, user)
171
172 stream_name = "user:#{user.id}"
173
174 assert_receive {:render_with_user, Pleroma.Web.StreamerView, "update.json", ^announce,
175 ^stream_name}
176
177 refute Streamer.filtered_by_user?(user, announce)
178 end
179
180 test "it does not stream announces of the user's own posts in the 'user' stream", %{
181 user: user,
182 token: oauth_token
183 } do
184 Streamer.get_topic_and_add_socket("user", user, oauth_token)
185
186 other_user = insert(:user)
187 {:ok, activity} = CommonAPI.post(user, %{status: "hey"})
188 {:ok, announce} = CommonAPI.repeat(activity.id, other_user)
189
190 assert Streamer.filtered_by_user?(user, announce)
191 end
192
193 test "it does stream notifications announces of the user's own posts in the 'user' stream", %{
194 user: user,
195 token: oauth_token
196 } do
197 Streamer.get_topic_and_add_socket("user", user, oauth_token)
198
199 other_user = insert(:user)
200 {:ok, activity} = CommonAPI.post(user, %{status: "hey"})
201 {:ok, announce} = CommonAPI.repeat(activity.id, other_user)
202
203 notification =
204 Pleroma.Notification
205 |> Repo.get_by(%{user_id: user.id, activity_id: announce.id})
206 |> Repo.preload(:activity)
207
208 refute Streamer.filtered_by_user?(user, notification)
209 end
210
211 test "it streams boosts of mastodon user in the 'user' stream", %{
212 user: user,
213 token: oauth_token
214 } do
215 Streamer.get_topic_and_add_socket("user", user, oauth_token)
216
217 other_user = insert(:user)
218 {:ok, activity} = CommonAPI.post(other_user, %{status: "hey"})
219
220 data =
221 File.read!("test/fixtures/mastodon-announce.json")
222 |> Jason.decode!()
223 |> Map.put("object", activity.data["object"])
224 |> Map.put("actor", user.ap_id)
225
226 {:ok, %Pleroma.Activity{data: _data, local: false} = announce} =
227 Pleroma.Web.ActivityPub.Transmogrifier.handle_incoming(data)
228
229 stream_name = "user:#{user.id}"
230
231 assert_receive {:render_with_user, Pleroma.Web.StreamerView, "update.json", ^announce,
232 ^stream_name}
233
234 refute Streamer.filtered_by_user?(user, announce)
235 end
236
237 test "it sends notify to in the 'user' stream", %{
238 user: user,
239 token: oauth_token,
240 notify: notify
241 } do
242 Streamer.get_topic_and_add_socket("user", user, oauth_token)
243 Streamer.stream("user", notify)
244
245 assert_receive {:render_with_user, _, _, ^notify, "user"}
246 refute Streamer.filtered_by_user?(user, notify)
247 end
248
249 test "it sends notify to in the 'user:notification' stream", %{
250 user: user,
251 token: oauth_token,
252 notify: notify
253 } do
254 Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
255 Streamer.stream("user:notification", notify)
256
257 assert_receive {:render_with_user, _, _, ^notify, "user:notification"}
258 refute Streamer.filtered_by_user?(user, notify)
259 end
260
261 test "it doesn't send notify to the 'user:notification' stream when a user is blocked", %{
262 user: user,
263 token: oauth_token
264 } do
265 blocked = insert(:user)
266 {:ok, _user_relationship} = User.block(user, blocked)
267
268 Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
269
270 {:ok, activity} = CommonAPI.post(user, %{status: ":("})
271 {:ok, _} = CommonAPI.favorite(blocked, activity.id)
272
273 refute_receive _
274 end
275
276 test "it doesn't send notify to the 'user:notification' stream when a thread is muted", %{
277 user: user,
278 token: oauth_token
279 } do
280 user2 = insert(:user)
281
282 {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
283 {:ok, _} = CommonAPI.add_mute(user, activity)
284
285 Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
286
287 {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
288
289 refute_receive _
290 assert Streamer.filtered_by_user?(user, favorite_activity)
291 end
292
293 test "it sends favorite to 'user:notification' stream'", %{
294 user: user,
295 token: oauth_token
296 } do
297 user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
298
299 {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
300 Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
301 {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
302
303 assert_receive {:render_with_user, _, "notification.json", notif, "user:notification"}
304 assert notif.activity.id == favorite_activity.id
305 refute Streamer.filtered_by_user?(user, notif)
306 end
307
308 test "it doesn't send the 'user:notification' stream' when a domain is blocked", %{
309 user: user,
310 token: oauth_token
311 } do
312 user2 = insert(:user, %{ap_id: "https://hecking-lewd-place.com/user/meanie"})
313
314 {:ok, user} = User.block_domain(user, "hecking-lewd-place.com")
315 {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
316 Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
317 {:ok, favorite_activity} = CommonAPI.favorite(user2, activity.id)
318
319 refute_receive _
320 assert Streamer.filtered_by_user?(user, favorite_activity)
321 end
322
323 test "it sends follow activities to the 'user:notification' stream", %{
324 user: user,
325 token: oauth_token
326 } do
327 user2 = insert(:user)
328
329 Streamer.get_topic_and_add_socket("user:notification", user, oauth_token)
330 {:ok, _follower, _followed, follow_activity} = CommonAPI.follow(user2, user)
331
332 assert_receive {:render_with_user, _, "notification.json", notif, "user:notification"}
333 assert notif.activity.id == follow_activity.id
334 refute Streamer.filtered_by_user?(user, notif)
335 end
336
337 test "it sends follow relationships updates to the 'user' stream", %{
338 user: user,
339 token: oauth_token
340 } do
341 user_id = user.id
342 other_user = insert(:user)
343 other_user_id = other_user.id
344
345 Streamer.get_topic_and_add_socket("user", user, oauth_token)
346 {:ok, _follower, _followed, _follow_activity} = CommonAPI.follow(user, other_user)
347
348 assert_receive {:text, event}
349
350 assert %{"event" => "pleroma:follow_relationships_update", "payload" => payload} =
351 Jason.decode!(event)
352
353 assert %{
354 "follower" => %{
355 "follower_count" => 0,
356 "following_count" => 0,
357 "id" => ^user_id
358 },
359 "following" => %{
360 "follower_count" => 0,
361 "following_count" => 0,
362 "id" => ^other_user_id
363 },
364 "state" => "follow_pending"
365 } = Jason.decode!(payload)
366
367 assert_receive {:text, event}
368
369 assert %{"event" => "pleroma:follow_relationships_update", "payload" => payload} =
370 Jason.decode!(event)
371
372 assert %{
373 "follower" => %{
374 "follower_count" => 0,
375 "following_count" => 1,
376 "id" => ^user_id
377 },
378 "following" => %{
379 "follower_count" => 1,
380 "following_count" => 0,
381 "id" => ^other_user_id
382 },
383 "state" => "follow_accept"
384 } = Jason.decode!(payload)
385 end
386
387 test "it streams edits in the 'user' stream", %{user: user, token: oauth_token} do
388 sender = insert(:user)
389 {:ok, _, _, _} = CommonAPI.follow(user, sender)
390
391 {:ok, activity} = CommonAPI.post(sender, %{status: "hey"})
392
393 Streamer.get_topic_and_add_socket("user", user, oauth_token)
394 {:ok, edited} = CommonAPI.update(sender, activity, %{status: "mew mew"})
395 create = Pleroma.Activity.get_create_by_object_ap_id_with_object(activity.object.data["id"])
396
397 stream = "user:#{user.id}"
398 assert_receive {:render_with_user, _, "status_update.json", ^create, ^stream}
399 refute Streamer.filtered_by_user?(user, edited)
400 end
401
402 test "it streams own edits in the 'user' stream", %{user: user, token: oauth_token} do
403 {:ok, activity} = CommonAPI.post(user, %{status: "hey"})
404
405 Streamer.get_topic_and_add_socket("user", user, oauth_token)
406 {:ok, edited} = CommonAPI.update(user, activity, %{status: "mew mew"})
407 create = Pleroma.Activity.get_create_by_object_ap_id_with_object(activity.object.data["id"])
408
409 stream = "user:#{user.id}"
410 assert_receive {:render_with_user, _, "status_update.json", ^create, ^stream}
411 refute Streamer.filtered_by_user?(user, edited)
412 end
413
414 test "it streams posts containing followed hashtags on the 'user' stream", %{
415 user: user,
416 token: oauth_token
417 } do
418 hashtag = insert(:hashtag, %{name: "tenshi"})
419 other_user = insert(:user)
420 {:ok, user} = User.follow_hashtag(user, hashtag)
421
422 Streamer.get_topic_and_add_socket("user", user, oauth_token)
423 {:ok, activity} = CommonAPI.post(other_user, %{status: "hey #tenshi"})
424
425 assert_receive {:render_with_user, _, "update.json", ^activity, _}
426 end
427
428 test "should not stream private posts containing followed hashtags on the 'user' stream", %{
429 user: user,
430 token: oauth_token
431 } do
432 hashtag = insert(:hashtag, %{name: "tenshi"})
433 other_user = insert(:user)
434 {:ok, user} = User.follow_hashtag(user, hashtag)
435
436 Streamer.get_topic_and_add_socket("user", user, oauth_token)
437
438 {:ok, activity} =
439 CommonAPI.post(other_user, %{status: "hey #tenshi", visibility: "private"})
440
441 refute_receive {:render_with_user, _, "update.json", ^activity, _}
442 end
443 end
444
445 describe "public streams" do
446 test "it sends to public (authenticated)" do
447 %{user: user, token: oauth_token} = oauth_access(["read"])
448 other_user = insert(:user)
449
450 Streamer.get_topic_and_add_socket("public", user, oauth_token)
451
452 {:ok, activity} = CommonAPI.post(other_user, %{status: "Test"})
453 assert_receive {:render_with_user, _, _, ^activity, "public"}
454 refute Streamer.filtered_by_user?(other_user, activity)
455 end
456
457 test "it sends to public (unauthenticated)" do
458 user = insert(:user)
459
460 Streamer.get_topic_and_add_socket("public", nil, nil)
461
462 {:ok, activity} = CommonAPI.post(user, %{status: "Test"})
463 activity_id = activity.id
464 assert_receive {:text, event}
465 assert %{"event" => "update", "payload" => payload} = Jason.decode!(event)
466 assert %{"id" => ^activity_id} = Jason.decode!(payload)
467
468 {:ok, _} = CommonAPI.delete(activity.id, user)
469 assert_receive {:text, event}
470 assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
471 end
472
473 test "handles deletions" do
474 %{user: user, token: oauth_token} = oauth_access(["read"])
475 other_user = insert(:user)
476 {:ok, activity} = CommonAPI.post(other_user, %{status: "Test"})
477
478 Streamer.get_topic_and_add_socket("public", user, oauth_token)
479
480 {:ok, _} = CommonAPI.delete(activity.id, other_user)
481 activity_id = activity.id
482 assert_receive {:text, event}
483 assert %{"event" => "delete", "payload" => ^activity_id} = Jason.decode!(event)
484 end
485
486 test "it streams edits in the 'public' stream" do
487 sender = insert(:user)
488
489 Streamer.get_topic_and_add_socket("public", nil, nil)
490 {:ok, activity} = CommonAPI.post(sender, %{status: "hey"})
491 assert_receive {:text, _}
492
493 {:ok, edited} = CommonAPI.update(sender, activity, %{status: "mew mew"})
494
495 edited = Pleroma.Activity.normalize(edited)
496
497 %{id: activity_id} = Pleroma.Activity.get_create_by_object_ap_id(edited.object.data["id"])
498
499 assert_receive {:text, event}
500 assert %{"event" => "status.update", "payload" => payload} = Jason.decode!(event)
501 assert %{"id" => ^activity_id} = Jason.decode!(payload)
502 refute Streamer.filtered_by_user?(sender, edited)
503 end
504
505 test "it streams multiple edits in the 'public' stream correctly" do
506 sender = insert(:user)
507
508 Streamer.get_topic_and_add_socket("public", nil, nil)
509 {:ok, activity} = CommonAPI.post(sender, %{status: "hey"})
510 assert_receive {:text, _}
511
512 {:ok, edited} = CommonAPI.update(sender, activity, %{status: "mew mew"})
513
514 edited = Pleroma.Activity.normalize(edited)
515
516 %{id: activity_id} = Pleroma.Activity.get_create_by_object_ap_id(edited.object.data["id"])
517
518 assert_receive {:text, event}
519 assert %{"event" => "status.update", "payload" => payload} = Jason.decode!(event)
520 assert %{"id" => ^activity_id} = Jason.decode!(payload)
521 refute Streamer.filtered_by_user?(sender, edited)
522
523 {:ok, edited} = CommonAPI.update(sender, activity, %{status: "mew mew 2"})
524
525 edited = Pleroma.Activity.normalize(edited)
526
527 %{id: activity_id} = Pleroma.Activity.get_create_by_object_ap_id(edited.object.data["id"])
528 assert_receive {:text, event}
529 assert %{"event" => "status.update", "payload" => payload} = Jason.decode!(event)
530 assert %{"id" => ^activity_id, "content" => "mew mew 2"} = Jason.decode!(payload)
531 refute Streamer.filtered_by_user?(sender, edited)
532 end
533 end
534
535 describe "thread_containment/2" do
536 test "it filters to user if recipients invalid and thread containment is enabled" do
537 clear_config([:instance, :skip_thread_containment], false)
538 author = insert(:user)
539 %{user: user, token: oauth_token} = oauth_access(["read"])
540 User.follow(user, author, :follow_accept)
541
542 activity =
543 insert(:note_activity,
544 note:
545 insert(:note,
546 user: author,
547 data: %{"to" => ["TEST-FFF"]}
548 )
549 )
550
551 Streamer.get_topic_and_add_socket("public", user, oauth_token)
552 Streamer.stream("public", activity)
553 assert_receive {:render_with_user, _, _, ^activity, "public"}
554 assert Streamer.filtered_by_user?(user, activity)
555 end
556
557 test "it sends message if recipients invalid and thread containment is disabled" do
558 clear_config([:instance, :skip_thread_containment], true)
559 author = insert(:user)
560 %{user: user, token: oauth_token} = oauth_access(["read"])
561 User.follow(user, author, :follow_accept)
562
563 activity =
564 insert(:note_activity,
565 note:
566 insert(:note,
567 user: author,
568 data: %{"to" => ["TEST-FFF"]}
569 )
570 )
571
572 Streamer.get_topic_and_add_socket("public", user, oauth_token)
573 Streamer.stream("public", activity)
574
575 assert_receive {:render_with_user, _, _, ^activity, "public"}
576 refute Streamer.filtered_by_user?(user, activity)
577 end
578
579 test "it sends message if recipients invalid and thread containment is enabled but user's thread containment is disabled" do
580 clear_config([:instance, :skip_thread_containment], false)
581 author = insert(:user)
582 user = insert(:user, skip_thread_containment: true)
583 %{token: oauth_token} = oauth_access(["read"], user: user)
584 User.follow(user, author, :follow_accept)
585
586 activity =
587 insert(:note_activity,
588 note:
589 insert(:note,
590 user: author,
591 data: %{"to" => ["TEST-FFF"]}
592 )
593 )
594
595 Streamer.get_topic_and_add_socket("public", user, oauth_token)
596 Streamer.stream("public", activity)
597
598 assert_receive {:render_with_user, _, _, ^activity, "public"}
599 refute Streamer.filtered_by_user?(user, activity)
600 end
601 end
602
603 describe "blocks" do
604 setup do: oauth_access(["read"])
605
606 test "it filters messages involving blocked users", %{user: user, token: oauth_token} do
607 blocked_user = insert(:user)
608 {:ok, _user_relationship} = User.block(user, blocked_user)
609
610 Streamer.get_topic_and_add_socket("public", user, oauth_token)
611 {:ok, activity} = CommonAPI.post(blocked_user, %{status: "Test"})
612 assert_receive {:render_with_user, _, _, ^activity, "public"}
613 assert Streamer.filtered_by_user?(user, activity)
614 end
615
616 test "it filters messages transitively involving blocked users", %{
617 user: blocker,
618 token: blocker_token
619 } do
620 blockee = insert(:user)
621 friend = insert(:user)
622
623 Streamer.get_topic_and_add_socket("public", blocker, blocker_token)
624
625 {:ok, _user_relationship} = User.block(blocker, blockee)
626
627 {:ok, activity_one} = CommonAPI.post(friend, %{status: "hey! @#{blockee.nickname}"})
628
629 assert_receive {:render_with_user, _, _, ^activity_one, "public"}
630 assert Streamer.filtered_by_user?(blocker, activity_one)
631
632 {:ok, activity_two} = CommonAPI.post(blockee, %{status: "hey! @#{friend.nickname}"})
633
634 assert_receive {:render_with_user, _, _, ^activity_two, "public"}
635 assert Streamer.filtered_by_user?(blocker, activity_two)
636
637 {:ok, activity_three} = CommonAPI.post(blockee, %{status: "hey! @#{blocker.nickname}"})
638
639 assert_receive {:render_with_user, _, _, ^activity_three, "public"}
640 assert Streamer.filtered_by_user?(blocker, activity_three)
641 end
642 end
643
644 describe "lists" do
645 setup do: oauth_access(["read"])
646
647 test "it doesn't send unwanted DMs to list", %{user: user_a, token: user_a_token} do
648 user_b = insert(:user)
649 user_c = insert(:user)
650
651 {:ok, user_a, user_b} = User.follow(user_a, user_b)
652
653 {:ok, list} = List.create("Test", user_a)
654 {:ok, list} = List.follow(list, user_b)
655
656 Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})
657
658 {:ok, _activity} =
659 CommonAPI.post(user_b, %{
660 status: "@#{user_c.nickname} Test",
661 visibility: "direct"
662 })
663
664 refute_receive _
665 end
666
667 test "it doesn't send unwanted private posts to list", %{user: user_a, token: user_a_token} do
668 user_b = insert(:user)
669
670 {:ok, list} = List.create("Test", user_a)
671 {:ok, list} = List.follow(list, user_b)
672
673 Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})
674
675 {:ok, _activity} =
676 CommonAPI.post(user_b, %{
677 status: "Test",
678 visibility: "private"
679 })
680
681 refute_receive _
682 end
683
684 test "it sends wanted private posts to list", %{user: user_a, token: user_a_token} do
685 user_b = insert(:user)
686
687 {:ok, user_a, user_b} = User.follow(user_a, user_b)
688
689 {:ok, list} = List.create("Test", user_a)
690 {:ok, list} = List.follow(list, user_b)
691
692 Streamer.get_topic_and_add_socket("list", user_a, user_a_token, %{"list" => list.id})
693
694 {:ok, activity} =
695 CommonAPI.post(user_b, %{
696 status: "Test",
697 visibility: "private"
698 })
699
700 stream_name = "list:#{list.id}"
701 assert_receive {:render_with_user, _, _, ^activity, ^stream_name}
702 refute Streamer.filtered_by_user?(user_a, activity)
703 end
704 end
705
706 describe "muted reblogs" do
707 setup do: oauth_access(["read"])
708
709 test "it filters muted reblogs", %{user: user1, token: user1_token} do
710 user2 = insert(:user)
711 user3 = insert(:user)
712 CommonAPI.follow(user1, user2)
713 CommonAPI.hide_reblogs(user1, user2)
714
715 {:ok, create_activity} = CommonAPI.post(user3, %{status: "I'm kawen"})
716
717 Streamer.get_topic_and_add_socket("user", user1, user1_token)
718 {:ok, announce_activity} = CommonAPI.repeat(create_activity.id, user2)
719 stream_name = "user:#{user1.id}"
720 assert_receive {:render_with_user, _, _, ^announce_activity, ^stream_name}
721 assert Streamer.filtered_by_user?(user1, announce_activity)
722 end
723
724 test "it filters reblog notification for reblog-muted actors", %{
725 user: user1,
726 token: user1_token
727 } do
728 user2 = insert(:user)
729 CommonAPI.follow(user1, user2)
730 CommonAPI.hide_reblogs(user1, user2)
731
732 {:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
733 Streamer.get_topic_and_add_socket("user", user1, user1_token)
734 {:ok, _announce_activity} = CommonAPI.repeat(create_activity.id, user2)
735
736 assert_receive {:render_with_user, _, "notification.json", notif, "user"}
737 assert Streamer.filtered_by_user?(user1, notif)
738 end
739
740 test "it send non-reblog notification for reblog-muted actors", %{
741 user: user1,
742 token: user1_token
743 } do
744 user2 = insert(:user)
745 CommonAPI.follow(user1, user2)
746 CommonAPI.hide_reblogs(user1, user2)
747
748 {:ok, create_activity} = CommonAPI.post(user1, %{status: "I'm kawen"})
749 Streamer.get_topic_and_add_socket("user", user1, user1_token)
750 {:ok, _favorite_activity} = CommonAPI.favorite(user2, create_activity.id)
751
752 assert_receive {:render_with_user, _, "notification.json", notif, "user"}
753 refute Streamer.filtered_by_user?(user1, notif)
754 end
755 end
756
757 describe "muted threads" do
758 test "it filters posts from muted threads" do
759 user = insert(:user)
760 %{user: user2, token: user2_token} = oauth_access(["read"])
761 Streamer.get_topic_and_add_socket("user", user2, user2_token)
762
763 {:ok, user2, user, _activity} = CommonAPI.follow(user2, user)
764 {:ok, activity} = CommonAPI.post(user, %{status: "super hot take"})
765 {:ok, _} = CommonAPI.add_mute(user2, activity)
766
767 stream_name = "user:#{user2.id}"
768 assert_receive {:render_with_user, _, _, ^activity, ^stream_name}
769 assert Streamer.filtered_by_user?(user2, activity)
770 end
771 end
772
773 describe "direct streams" do
774 setup do: oauth_access(["read"])
775
776 test "it sends conversation update to the 'direct' stream", %{user: user, token: oauth_token} do
777 another_user = insert(:user)
778
779 Streamer.get_topic_and_add_socket("direct", user, oauth_token)
780
781 {:ok, _create_activity} =
782 CommonAPI.post(another_user, %{
783 status: "hey @#{user.nickname}",
784 visibility: "direct"
785 })
786
787 assert_receive {:text, received_event}
788
789 assert %{"event" => "conversation", "payload" => received_payload} =
790 Jason.decode!(received_event)
791
792 assert %{"last_status" => last_status} = Jason.decode!(received_payload)
793 [participation] = Participation.for_user(user)
794 assert last_status["pleroma"]["direct_conversation_id"] == participation.id
795 end
796
797 test "it doesn't send conversation update to the 'direct' stream when the last message in the conversation is deleted",
798 %{user: user, token: oauth_token} do
799 another_user = insert(:user)
800
801 Streamer.get_topic_and_add_socket("direct", user, oauth_token)
802
803 {:ok, create_activity} =
804 CommonAPI.post(another_user, %{
805 status: "hi @#{user.nickname}",
806 visibility: "direct"
807 })
808
809 create_activity_id = create_activity.id
810 stream_name = "direct:#{user.id}"
811 assert_receive {:render_with_user, _, _, ^create_activity, ^stream_name}
812 assert_receive {:text, received_conversation1}
813 assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
814
815 {:ok, _} = CommonAPI.delete(create_activity_id, another_user)
816
817 assert_receive {:text, received_event}
818
819 assert %{"event" => "delete", "payload" => ^create_activity_id} =
820 Jason.decode!(received_event)
821
822 refute_receive _
823 end
824
825 @tag :erratic
826 test "it sends conversation update to the 'direct' stream when a message is deleted", %{
827 user: user,
828 token: oauth_token
829 } do
830 another_user = insert(:user)
831 Streamer.get_topic_and_add_socket("direct", user, oauth_token)
832
833 {:ok, create_activity} =
834 CommonAPI.post(another_user, %{
835 status: "hi @#{user.nickname}",
836 visibility: "direct"
837 })
838
839 {:ok, create_activity2} =
840 CommonAPI.post(another_user, %{
841 status: "hi @#{user.nickname} 2",
842 in_reply_to_status_id: create_activity.id,
843 visibility: "direct"
844 })
845
846 stream_name = "direct:#{user.id}"
847 assert_receive {:render_with_user, _, _, ^create_activity, ^stream_name}
848 assert_receive {:render_with_user, _, _, ^create_activity2, ^stream_name}
849 assert_receive {:text, received_conversation1}
850 assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
851 assert_receive {:text, received_conversation1}
852 assert %{"event" => "conversation", "payload" => _} = Jason.decode!(received_conversation1)
853
854 {:ok, _} = CommonAPI.delete(create_activity2.id, another_user)
855
856 assert_receive {:text, received_event}
857 assert %{"event" => "delete", "payload" => _} = Jason.decode!(received_event)
858
859 assert_receive {:text, received_event}
860
861 assert %{"event" => "conversation", "payload" => received_payload} =
862 Jason.decode!(received_event)
863
864 assert %{"last_status" => last_status} = Jason.decode!(received_payload)
865 assert last_status["id"] == to_string(create_activity.id)
866 end
867 end
868
869 describe "stop streaming if token got revoked" do
870 setup do
871 child_proc = fn start, finalize ->
872 fn ->
873 start.()
874
875 receive do
876 {StreamerTest, :ready} ->
877 assert_receive {:render_with_user, _, "update.json", _, _}
878
879 receive do
880 {StreamerTest, :revoked} -> finalize.()
881 end
882 end
883 end
884 end
885
886 starter = fn user, token ->
887 fn -> Streamer.get_topic_and_add_socket("user", user, token) end
888 end
889
890 hit = fn -> assert_receive :close end
891 miss = fn -> refute_receive :close end
892
893 send_all = fn tasks, thing -> Enum.each(tasks, &send(&1.pid, thing)) end
894
895 %{
896 child_proc: child_proc,
897 starter: starter,
898 hit: hit,
899 miss: miss,
900 send_all: send_all
901 }
902 end
903
904 test "do not revoke other tokens", %{
905 child_proc: child_proc,
906 starter: starter,
907 hit: hit,
908 miss: miss,
909 send_all: send_all
910 } do
911 %{user: user, token: token} = oauth_access(["read"])
912 %{token: token2} = oauth_access(["read"], user: user)
913 %{user: user2, token: user2_token} = oauth_access(["read"])
914
915 post_user = insert(:user)
916 CommonAPI.follow(user, post_user)
917 CommonAPI.follow(user2, post_user)
918
919 tasks = [
920 Task.async(child_proc.(starter.(user, token), hit)),
921 Task.async(child_proc.(starter.(user, token2), miss)),
922 Task.async(child_proc.(starter.(user2, user2_token), miss))
923 ]
924
925 {:ok, _} =
926 CommonAPI.post(post_user, %{
927 status: "hi"
928 })
929
930 send_all.(tasks, {StreamerTest, :ready})
931
932 Pleroma.Web.OAuth.Token.Strategy.Revoke.revoke(token)
933
934 send_all.(tasks, {StreamerTest, :revoked})
935
936 Enum.each(tasks, &Task.await/1)
937 end
938
939 test "revoke all streams for this token", %{
940 child_proc: child_proc,
941 starter: starter,
942 hit: hit,
943 send_all: send_all
944 } do
945 %{user: user, token: token} = oauth_access(["read"])
946
947 post_user = insert(:user)
948 CommonAPI.follow(user, post_user)
949
950 tasks = [
951 Task.async(child_proc.(starter.(user, token), hit)),
952 Task.async(child_proc.(starter.(user, token), hit))
953 ]
954
955 {:ok, _} =
956 CommonAPI.post(post_user, %{
957 status: "hi"
958 })
959
960 send_all.(tasks, {StreamerTest, :ready})
961
962 Pleroma.Web.OAuth.Token.Strategy.Revoke.revoke(token)
963
964 send_all.(tasks, {StreamerTest, :revoked})
965
966 Enum.each(tasks, &Task.await/1)
967 end
968 end
969 end