paulmataruso/zulip
1
0
Fork 0
mirror of https://github.com/zulip/zulip.git synced 2025-11-04 22:13:26 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
2795f11e3ffe910780cce3cc4026da555a3e0c0d
zulip/zerver/tests/test_user_topics.py
Prakhar Pratyush 58568a60d6 settings: Add automatically follow and unmute topics policy settings. 
This commit adds two user settings, named
* `automatically_follow_topics_policy`
* `automatically_unmute_topics_in_muted_streams_policy`

The settings control the user's preference on which topics they
will automatically 'follow' or 'unmute in muted streams'.

The policies offer four options:
1. Topics I participate in
2. Topics I send a message to
3. Topics I start
4. Never (default)

There is no support for configuring the settings through the UI yet.
2023-10-04 13:04:29 -07:00

1537 lines
62 KiB
Python
Raw Blame History

from datetime import datetime, timezone
from typing import Any, Dict, List
import orjson
import time_machine
from django.utils.timezone import now as timezone_now
from zerver.actions.reactions import check_add_reaction
from zerver.actions.user_settings import do_change_user_setting
from zerver.actions.user_topics import do_set_user_topic_visibility_policy
from zerver.lib.stream_topic import StreamTopicTarget
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import get_subscription
from zerver.lib.user_topics import get_topic_mutes, topic_has_visibility_policy
from zerver.models import UserProfile, UserTopic, get_stream
class MutedTopicsTestsDeprecated(ZulipTestCase):
# Tests the deprecated URL: "/api/v1/users/me/subscriptions/muted_topics".
# It exists for backward compatibility and should be removed once
# we remove the deprecated URL.
def test_get_deactivated_muted_topic(self) -> None:
user = self.example_user("hamlet")
self.login_user(user)
stream = get_stream("Verona", user.realm)
mock_date_muted = datetime(2020, 1, 1, tzinfo=timezone.utc).timestamp()
url = "/api/v1/users/me/subscriptions/muted_topics"
data = {"stream_id": stream.id, "topic": "Verona3", "op": "add"}
with time_machine.travel(datetime(2020, 1, 1, tzinfo=timezone.utc), tick=False):
result = self.api_patch(user, url, data)
self.assert_json_success(result)
stream.deactivated = True
stream.save()
self.assertNotIn((stream.name, "Verona3", mock_date_muted), get_topic_mutes(user))
self.assertIn((stream.name, "Verona3", mock_date_muted), get_topic_mutes(user, True))
def test_user_ids_muting_topic(self) -> None:
hamlet = self.example_user("hamlet")
cordelia = self.example_user("cordelia")
realm = hamlet.realm
stream = get_stream("Verona", realm)
topic_name = "teST topic"
date_muted = datetime(2020, 1, 1, tzinfo=timezone.utc)
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.MUTED
)
self.assertEqual(user_ids, set())
url = "/api/v1/users/me/subscriptions/muted_topics"
data = {"stream_id": stream.id, "topic": "test TOPIC", "op": "add"}
def mute_topic_for_user(user: UserProfile) -> None:
with time_machine.travel(datetime(2020, 1, 1, tzinfo=timezone.utc), tick=False):
result = self.api_patch(user, url, data)
self.assert_json_success(result)
mute_topic_for_user(hamlet)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.MUTED
)
self.assertEqual(user_ids, {hamlet.id})
hamlet_date_muted = UserTopic.objects.filter(
user_profile=hamlet, visibility_policy=UserTopic.VisibilityPolicy.MUTED
)[0].last_updated
self.assertEqual(hamlet_date_muted, date_muted)
mute_topic_for_user(cordelia)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.MUTED
)
self.assertEqual(user_ids, {hamlet.id, cordelia.id})
cordelia_date_muted = UserTopic.objects.filter(
user_profile=cordelia, visibility_policy=UserTopic.VisibilityPolicy.MUTED
)[0].last_updated
self.assertEqual(cordelia_date_muted, date_muted)
def test_add_muted_topic(self) -> None:
user = self.example_user("hamlet")
self.login_user(user)
stream = get_stream("Verona", user.realm)
url = "/api/v1/users/me/subscriptions/muted_topics"
payloads: List[Dict[str, object]] = [
{"stream": stream.name, "topic": "Verona3", "op": "add"},
{"stream_id": stream.id, "topic": "Verona3", "op": "add"},
]
mock_date_muted = datetime(2020, 1, 1, tzinfo=timezone.utc).timestamp()
for data in payloads:
with time_machine.travel(datetime(2020, 1, 1, tzinfo=timezone.utc), tick=False):
result = self.api_patch(user, url, data)
self.assert_json_success(result)
self.assertIn((stream.name, "Verona3", mock_date_muted), get_topic_mutes(user))
self.assertTrue(
topic_has_visibility_policy(
user, stream.id, "verona3", UserTopic.VisibilityPolicy.MUTED
)
)
do_set_user_topic_visibility_policy(
user,
stream,
"Verona3",
visibility_policy=UserTopic.VisibilityPolicy.INHERIT,
)
assert stream.recipient is not None
result = self.api_patch(user, url, data)
# Now check that no error is raised when attempted to mute
# an already muted topic. This should be case-insensitive.
user_topic_count = UserTopic.objects.count()
data["topic"] = "VERONA3"
with self.assertLogs(level="INFO") as info_logs:
result = self.api_patch(user, url, data)
self.assert_json_success(result)
self.assertEqual(
info_logs.output[0],
f"INFO:root:User {user.id} tried to set visibility_policy to its current value of {UserTopic.VisibilityPolicy.MUTED}",
)
# Verify that we didn't end up with duplicate UserTopic rows
# with the two different cases after the previous API call.
self.assertEqual(UserTopic.objects.count() - user_topic_count, 0)
def test_remove_muted_topic(self) -> None:
user = self.example_user("hamlet")
realm = user.realm
self.login_user(user)
stream = get_stream("Verona", realm)
url = "/api/v1/users/me/subscriptions/muted_topics"
payloads: List[Dict[str, object]] = [
{"stream": stream.name, "topic": "vERONA3", "op": "remove"},
{"stream_id": stream.id, "topic": "vEroNA3", "op": "remove"},
]
mock_date_muted = datetime(2020, 1, 1, tzinfo=timezone.utc).timestamp()
for data in payloads:
do_set_user_topic_visibility_policy(
user,
stream,
"Verona3",
visibility_policy=UserTopic.VisibilityPolicy.MUTED,
last_updated=datetime(2020, 1, 1, tzinfo=timezone.utc),
)
self.assertIn((stream.name, "Verona3", mock_date_muted), get_topic_mutes(user))
result = self.api_patch(user, url, data)
self.assert_json_success(result)
self.assertNotIn((stream.name, "Verona3", mock_date_muted), get_topic_mutes(user))
self.assertFalse(
topic_has_visibility_policy(
user, stream.id, "verona3", UserTopic.VisibilityPolicy.MUTED
)
)
def test_muted_topic_add_invalid(self) -> None:
user = self.example_user("hamlet")
realm = user.realm
self.login_user(user)
stream = get_stream("Verona", realm)
do_set_user_topic_visibility_policy(
user,
stream,
"Verona3",
visibility_policy=UserTopic.VisibilityPolicy.MUTED,
last_updated=timezone_now(),
)
url = "/api/v1/users/me/subscriptions/muted_topics"
data = {"stream_id": 999999999, "topic": "Verona3", "op": "add"}
result = self.api_patch(user, url, data)
self.assert_json_error(result, "Invalid stream ID")
data = {"topic": "Verona3", "op": "add"}
result = self.api_patch(user, url, data)
self.assert_json_error(result, "Please supply 'stream'.")
data = {"stream": stream.name, "stream_id": stream.id, "topic": "Verona3", "op": "add"}
result = self.api_patch(user, url, data)
self.assert_json_error(result, "Please choose one: 'stream' or 'stream_id'.")
def test_muted_topic_remove_invalid(self) -> None:
user = self.example_user("hamlet")
realm = user.realm
self.login_user(user)
stream = get_stream("Verona", realm)
url = "/api/v1/users/me/subscriptions/muted_topics"
data: Dict[str, Any] = {"stream": "BOGUS", "topic": "Verona3", "op": "remove"}
result = self.api_patch(user, url, data)
self.assert_json_error(result, "Topic is not muted")
# Check that removing mute from a topic for which the user
# doesn't already have a visibility_policy doesn't cause an error.
data = {"stream": stream.name, "topic": "BOGUS", "op": "remove"}
with self.assertLogs(level="INFO") as info_logs:
result = self.api_patch(user, url, data)
self.assert_json_success(result)
self.assertEqual(
info_logs.output[0],
f"INFO:root:User {user.id} tried to remove visibility_policy, which actually doesn't exist",
)
data = {"stream_id": 999999999, "topic": "BOGUS", "op": "remove"}
result = self.api_patch(user, url, data)
self.assert_json_error(result, "Topic is not muted")
data = {"topic": "Verona3", "op": "remove"}
result = self.api_patch(user, url, data)
self.assert_json_error(result, "Please supply 'stream'.")
data = {"stream": stream.name, "stream_id": stream.id, "topic": "Verona3", "op": "remove"}
result = self.api_patch(user, url, data)
self.assert_json_error(result, "Please choose one: 'stream' or 'stream_id'.")
class MutedTopicsTests(ZulipTestCase):
def test_get_deactivated_muted_topic(self) -> None:
user = self.example_user("hamlet")
self.login_user(user)
stream = get_stream("Verona", user.realm)
url = "/api/v1/user_topics"
data = {
"stream_id": stream.id,
"topic": "Verona3",
"visibility_policy": UserTopic.VisibilityPolicy.MUTED,
}
mock_date_muted = datetime(2020, 1, 1, tzinfo=timezone.utc).timestamp()
with time_machine.travel(datetime(2020, 1, 1, tzinfo=timezone.utc), tick=False):
result = self.api_post(user, url, data)
self.assert_json_success(result)
stream.deactivated = True
stream.save()
self.assertNotIn((stream.name, "Verona3", mock_date_muted), get_topic_mutes(user))
self.assertIn((stream.name, "Verona3", mock_date_muted), get_topic_mutes(user, True))
def test_user_ids_muting_topic(self) -> None:
hamlet = self.example_user("hamlet")
cordelia = self.example_user("cordelia")
realm = hamlet.realm
stream = get_stream("Verona", realm)
topic_name = "teST topic"
date_muted = datetime(2020, 1, 1, tzinfo=timezone.utc)
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.MUTED
)
self.assertEqual(user_ids, set())
url = "/api/v1/user_topics"
def set_topic_visibility_for_user(user: UserProfile, visibility_policy: int) -> None:
data = {
"stream_id": stream.id,
"topic": "test TOPIC",
"visibility_policy": visibility_policy,
}
with time_machine.travel(date_muted, tick=False):
result = self.api_post(user, url, data)
self.assert_json_success(result)
set_topic_visibility_for_user(hamlet, UserTopic.VisibilityPolicy.MUTED)
set_topic_visibility_for_user(cordelia, UserTopic.VisibilityPolicy.UNMUTED)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.MUTED
)
self.assertEqual(user_ids, {hamlet.id})
hamlet_date_muted = UserTopic.objects.filter(
user_profile=hamlet, visibility_policy=UserTopic.VisibilityPolicy.MUTED
)[0].last_updated
self.assertEqual(hamlet_date_muted, date_muted)
set_topic_visibility_for_user(cordelia, UserTopic.VisibilityPolicy.MUTED)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.MUTED
)
self.assertEqual(user_ids, {hamlet.id, cordelia.id})
cordelia_date_muted = UserTopic.objects.filter(
user_profile=cordelia, visibility_policy=UserTopic.VisibilityPolicy.MUTED
)[0].last_updated
self.assertEqual(cordelia_date_muted, date_muted)
def test_add_muted_topic(self) -> None:
user = self.example_user("hamlet")
self.login_user(user)
stream = get_stream("Verona", user.realm)
url = "/api/v1/user_topics"
data = {
"stream_id": stream.id,
"topic": "Verona3",
"visibility_policy": UserTopic.VisibilityPolicy.MUTED,
}
mock_date_muted = datetime(2020, 1, 1, tzinfo=timezone.utc).timestamp()
with self.capture_send_event_calls(expected_num_events=2) as events:
with time_machine.travel(datetime(2020, 1, 1, tzinfo=timezone.utc), tick=False):
result = self.api_post(user, url, data)
self.assert_json_success(result)
self.assertTrue(
topic_has_visibility_policy(
user, stream.id, "verona3", UserTopic.VisibilityPolicy.MUTED
)
)
# Verify if events are sent properly
user_topic_event: Dict[str, Any] = {
"type": "user_topic",
"stream_id": stream.id,
"topic_name": "Verona3",
"last_updated": mock_date_muted,
"visibility_policy": UserTopic.VisibilityPolicy.MUTED,
}
muted_topics_event = dict(type="muted_topics", muted_topics=get_topic_mutes(user))
self.assertEqual(events[0]["event"], muted_topics_event)
self.assertEqual(events[1]["event"], user_topic_event)
# Now check that no error is raised when attempted to mute
# an already muted topic. This should be case-insensitive.
user_topic_count = UserTopic.objects.count()
data["topic"] = "VERONA3"
with self.assertLogs(level="INFO") as info_logs:
result = self.api_post(user, url, data)
self.assert_json_success(result)
self.assertEqual(
info_logs.output[0],
f"INFO:root:User {user.id} tried to set visibility_policy to its current value of {UserTopic.VisibilityPolicy.MUTED}",
)
# Verify that we didn't end up with duplicate UserTopic rows
# with the two different cases after the previous API call.
self.assertEqual(UserTopic.objects.count() - user_topic_count, 0)
def test_remove_muted_topic(self) -> None:
user = self.example_user("hamlet")
realm = user.realm
self.login_user(user)
stream = get_stream("Verona", realm)
do_set_user_topic_visibility_policy(
user,
stream,
"Verona3",
visibility_policy=UserTopic.VisibilityPolicy.MUTED,
last_updated=datetime(2020, 1, 1, tzinfo=timezone.utc),
)
self.assertTrue(
topic_has_visibility_policy(
user, stream.id, "verona3", UserTopic.VisibilityPolicy.MUTED
)
)
url = "/api/v1/user_topics"
data = {
"stream_id": stream.id,
"topic": "Verona3",
"visibility_policy": UserTopic.VisibilityPolicy.INHERIT,
}
mock_date_mute_removed = datetime(2020, 1, 1, tzinfo=timezone.utc).timestamp()
with self.capture_send_event_calls(expected_num_events=2) as events:
with time_machine.travel(datetime(2020, 1, 1, tzinfo=timezone.utc), tick=False):
result = self.api_post(user, url, data)
self.assert_json_success(result)
self.assertFalse(
topic_has_visibility_policy(
user, stream.id, "verona3", UserTopic.VisibilityPolicy.MUTED
)
)
# Verify if events are sent properly
user_topic_event: Dict[str, Any] = {
"type": "user_topic",
"stream_id": stream.id,
"topic_name": data["topic"],
"last_updated": mock_date_mute_removed,
"visibility_policy": UserTopic.VisibilityPolicy.INHERIT,
}
muted_topics_event = dict(type="muted_topics", muted_topics=get_topic_mutes(user))
self.assertEqual(events[0]["event"], muted_topics_event)
self.assertEqual(events[1]["event"], user_topic_event)
# Check that removing mute from a topic for which the user
# doesn't already have a visibility_policy doesn't cause an error.
with self.assertLogs(level="INFO") as info_logs:
result = self.api_post(user, url, data)
self.assert_json_success(result)
self.assertEqual(
info_logs.output[0],
f"INFO:root:User {user.id} tried to remove visibility_policy, which actually doesn't exist",
)
def test_muted_topic_add_invalid(self) -> None:
user = self.example_user("hamlet")
self.login_user(user)
url = "/api/v1/user_topics"
data = {
"stream_id": 999999999,
"topic": "Verona3",
"visibility_policy": UserTopic.VisibilityPolicy.MUTED,
}
result = self.api_post(user, url, data)
self.assert_json_error(result, "Invalid stream ID")
def test_muted_topic_remove_invalid(self) -> None:
user = self.example_user("hamlet")
self.login_user(user)
url = "/api/v1/user_topics"
data = {
"stream_id": 999999999,
"topic": "Verona3",
"visibility_policy": UserTopic.VisibilityPolicy.INHERIT,
}
result = self.api_post(user, url, data)
self.assert_json_error(result, "Invalid stream ID")
class UnmutedTopicsTests(ZulipTestCase):
def test_user_ids_unmuting_topic(self) -> None:
hamlet = self.example_user("hamlet")
cordelia = self.example_user("cordelia")
realm = hamlet.realm
stream = get_stream("Verona", realm)
topic_name = "teST topic"
date_unmuted = datetime(2020, 1, 1, tzinfo=timezone.utc)
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
url = "/api/v1/user_topics"
def set_topic_visibility_for_user(user: UserProfile, visibility_policy: int) -> None:
data = {
"stream_id": stream.id,
"topic": "test TOPIC",
"visibility_policy": visibility_policy,
}
with time_machine.travel(datetime(2020, 1, 1, tzinfo=timezone.utc), tick=False):
result = self.api_post(user, url, data)
self.assert_json_success(result)
set_topic_visibility_for_user(hamlet, UserTopic.VisibilityPolicy.UNMUTED)
set_topic_visibility_for_user(cordelia, UserTopic.VisibilityPolicy.MUTED)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, {hamlet.id})
hamlet_date_unmuted = UserTopic.objects.filter(
user_profile=hamlet, visibility_policy=UserTopic.VisibilityPolicy.UNMUTED
)[0].last_updated
self.assertEqual(hamlet_date_unmuted, date_unmuted)
set_topic_visibility_for_user(cordelia, UserTopic.VisibilityPolicy.UNMUTED)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, {hamlet.id, cordelia.id})
cordelia_date_unmuted = UserTopic.objects.filter(
user_profile=cordelia, visibility_policy=UserTopic.VisibilityPolicy.UNMUTED
)[0].last_updated
self.assertEqual(cordelia_date_unmuted, date_unmuted)
def test_add_unmuted_topic(self) -> None:
user = self.example_user("hamlet")
self.login_user(user)
stream = get_stream("Verona", user.realm)
url = "/api/v1/user_topics"
data = {
"stream_id": stream.id,
"topic": "Verona3",
"visibility_policy": UserTopic.VisibilityPolicy.UNMUTED,
}
mock_date_unmuted = datetime(2020, 1, 1, tzinfo=timezone.utc).timestamp()
with self.capture_send_event_calls(expected_num_events=2) as events:
with time_machine.travel(datetime(2020, 1, 1, tzinfo=timezone.utc), tick=False):
result = self.api_post(user, url, data)
self.assert_json_success(result)
self.assertTrue(
topic_has_visibility_policy(
user, stream.id, "verona3", UserTopic.VisibilityPolicy.UNMUTED
)
)
# Verify if events are sent properly
user_topic_event: Dict[str, Any] = {
"type": "user_topic",
"stream_id": stream.id,
"topic_name": "Verona3",
"last_updated": mock_date_unmuted,
"visibility_policy": UserTopic.VisibilityPolicy.UNMUTED,
}
muted_topics_event = dict(type="muted_topics", muted_topics=get_topic_mutes(user))
self.assertEqual(events[0]["event"], muted_topics_event)
self.assertEqual(events[1]["event"], user_topic_event)
# Now check that no error is raised when attempted to UNMUTE
# an already UNMUTED topic. This should be case-insensitive.
user_topic_count = UserTopic.objects.count()
data["topic"] = "VERONA3"
with self.assertLogs(level="INFO") as info_logs:
result = self.api_post(user, url, data)
self.assert_json_success(result)
self.assertEqual(
info_logs.output[0],
f"INFO:root:User {user.id} tried to set visibility_policy to its current value of {UserTopic.VisibilityPolicy.UNMUTED}",
)
# Verify that we didn't end up with duplicate UserTopic rows
# with the two different cases after the previous API call.
self.assertEqual(UserTopic.objects.count() - user_topic_count, 0)
def test_remove_unmuted_topic(self) -> None:
user = self.example_user("hamlet")
realm = user.realm
self.login_user(user)
stream = get_stream("Verona", realm)
do_set_user_topic_visibility_policy(
user,
stream,
"Verona3",
visibility_policy=UserTopic.VisibilityPolicy.UNMUTED,
last_updated=datetime(2020, 1, 1, tzinfo=timezone.utc),
)
self.assertTrue(
topic_has_visibility_policy(
user, stream.id, "verona3", UserTopic.VisibilityPolicy.UNMUTED
)
)
url = "/api/v1/user_topics"
data = {
"stream_id": stream.id,
"topic": "vEroNA3",
"visibility_policy": UserTopic.VisibilityPolicy.INHERIT,
}
mock_date_unmute_removed = datetime(2020, 1, 1, tzinfo=timezone.utc).timestamp()
with self.capture_send_event_calls(expected_num_events=2) as events:
with time_machine.travel(datetime(2020, 1, 1, tzinfo=timezone.utc), tick=False):
result = self.api_post(user, url, data)
self.assert_json_success(result)
self.assertFalse(
topic_has_visibility_policy(
user, stream.id, "verona3", UserTopic.VisibilityPolicy.UNMUTED
)
)
# Verify if events are sent properly
user_topic_event: Dict[str, Any] = {
"type": "user_topic",
"stream_id": stream.id,
"topic_name": data["topic"],
"last_updated": mock_date_unmute_removed,
"visibility_policy": UserTopic.VisibilityPolicy.INHERIT,
}
muted_topics_event = dict(type="muted_topics", muted_topics=get_topic_mutes(user))
self.assertEqual(events[0]["event"], muted_topics_event)
self.assertEqual(events[1]["event"], user_topic_event)
# Check that removing UNMUTE from a topic for which the user
# doesn't already have a visibility_policy doesn't cause an error.
with self.assertLogs(level="INFO") as info_logs:
result = self.api_post(user, url, data)
self.assert_json_success(result)
self.assertEqual(
info_logs.output[0],
f"INFO:root:User {user.id} tried to remove visibility_policy, which actually doesn't exist",
)
def test_unmuted_topic_add_invalid(self) -> None:
user = self.example_user("hamlet")
self.login_user(user)
url = "/api/v1/user_topics"
data = {
"stream_id": 999999999,
"topic": "Verona3",
"visibility_policy": UserTopic.VisibilityPolicy.UNMUTED,
}
result = self.api_post(user, url, data)
self.assert_json_error(result, "Invalid stream ID")
def test_unmuted_topic_remove_invalid(self) -> None:
user = self.example_user("hamlet")
self.login_user(user)
url = "/api/v1/user_topics"
data = {
"stream_id": 999999999,
"topic": "Verona3",
"visibility_policy": UserTopic.VisibilityPolicy.INHERIT,
}
result = self.api_post(user, url, data)
self.assert_json_error(result, "Invalid stream ID")
class AutomaticallyFollowTopicsTests(ZulipTestCase):
def test_automatically_follow_topic_on_initiation(self) -> None:
hamlet = self.example_user("hamlet")
cordelia = self.example_user("cordelia")
iago = self.example_user("iago")
stream = get_stream("Verona", hamlet.realm)
topic_name = "teST topic"
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, set())
# For hamlet & cordelia,
# 'automatically_follow_topics_policy' set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION'.
for user in [hamlet, cordelia]:
do_change_user_setting(
user,
"automatically_follow_topics_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
acting_user=None,
)
# Hamlet starts a topic. DO automatically follow the topic.
self.send_stream_message(hamlet, stream_name=stream.name, topic_name=topic_name)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, {hamlet.id})
# Cordelia sends a message to the topic which hamlet started. DON'T automatically follow the topic.
self.send_stream_message(cordelia, stream_name=stream.name, topic_name=topic_name)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, {hamlet.id})
# Iago has 'automatically_follow_topics_policy' set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_NEVER'.
# DON'T automatically follow the topic, even if he starts the topic.
do_change_user_setting(
iago,
"automatically_follow_topics_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_NEVER,
acting_user=None,
)
self.send_stream_message(iago, stream_name=stream.name, topic_name="New Topic")
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name="New Topic",
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, set())
# When a user sends the first message to a topic with protected history,
# the user starts that topic from their perspective. So, the user
# should follow the topic if 'automatically_follow_topics_policy' is set
# to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION', even if the message
# is not the first message in the topic.
private_stream = self.make_stream(stream_name="private stream", invite_only=True)
self.subscribe(iago, private_stream.name)
self.send_stream_message(iago, private_stream.name)
# Hamlet should automatically follow the topic, even if it already has messages.
self.subscribe(hamlet, private_stream.name)
self.send_stream_message(hamlet, private_stream.name)
stream_topic_target = StreamTopicTarget(
stream_id=private_stream.id,
topic_name="test",
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, {hamlet.id})
def test_automatically_follow_topic_on_send(self) -> None:
hamlet = self.example_user("hamlet")
aaron = self.example_user("aaron")
stream = get_stream("Verona", hamlet.realm)
topic_name = "teST topic"
self.send_stream_message(aaron, stream.name, "hello", topic_name)
# For hamlet, 'automatically_follow_topics_policy' set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_SEND'.
do_change_user_setting(
hamlet,
"automatically_follow_topics_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_SEND,
acting_user=None,
)
# For aaron, 'automatically_follow_topics_policy' NOT set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_SEND'.
do_change_user_setting(
aaron,
"automatically_follow_topics_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
acting_user=None,
)
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, set())
# Hamlet sends a message. DO automatically follow the topic.
# Aaron sends a message. DON'T automatically follow the topic.
self.send_stream_message(hamlet, stream_name=stream.name, topic_name=topic_name)
self.send_stream_message(aaron, stream_name=stream.name, topic_name=topic_name)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, {hamlet.id})
def test_automatically_follow_topic_on_participation_send_message(self) -> None:
hamlet = self.example_user("hamlet")
aaron = self.example_user("aaron")
stream = get_stream("Verona", hamlet.realm)
topic_name = "teST topic"
# For hamlet, 'automatically_follow_topics_policy' set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
hamlet,
"automatically_follow_topics_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
acting_user=None,
)
# For aaron, 'automatically_follow_topics_policy' NOT set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
aaron,
"automatically_follow_topics_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
acting_user=None,
)
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, set())
# Hamlet sends a message. DO automatically follow the topic.
# Aaron sends a message. DON'T automatically follow the topic.
self.send_stream_message(hamlet, stream_name=stream.name, topic_name=topic_name)
self.send_stream_message(aaron, stream_name=stream.name, topic_name=topic_name)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, {hamlet.id})
def test_automatically_follow_topic_on_participation_add_reaction(self) -> None:
cordelia = self.example_user("cordelia")
hamlet = self.example_user("hamlet")
aaron = self.example_user("aaron")
stream = get_stream("Verona", aaron.realm)
topic_name = "teST topic"
# For cordelia, 'automatically_follow_topics_policy' set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
cordelia,
"automatically_follow_topics_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
acting_user=None,
)
# For aaron, 'automatically_follow_topics_policy' NOT set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
aaron,
"automatically_follow_topics_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
acting_user=None,
)
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, set())
message_id = self.send_stream_message(
hamlet, stream_name=stream.name, topic_name=topic_name
)
# Cordelia reacts to a message. DO automatically follow the topic.
# Aaron reacts to a message. DON'T automatically follow the topic.
check_add_reaction(
user_profile=cordelia,
message_id=message_id,
emoji_name="smile",
emoji_code=None,
reaction_type=None,
)
check_add_reaction(
user_profile=aaron,
message_id=message_id,
emoji_name="smile",
emoji_code=None,
reaction_type=None,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, {cordelia.id})
# We don't decrease visibility policy
sub = get_subscription(stream.name, cordelia)
sub.is_muted = True
sub.save()
do_change_user_setting(
cordelia,
"automatically_follow_topics_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_NEVER,
acting_user=None,
)
do_change_user_setting(
cordelia,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
acting_user=None,
)
check_add_reaction(
user_profile=cordelia,
message_id=message_id,
emoji_name="plus",
emoji_code=None,
reaction_type=None,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, {cordelia.id})
# increase visibility policy
do_set_user_topic_visibility_policy(
cordelia,
stream,
topic_name,
visibility_policy=UserTopic.VisibilityPolicy.MUTED,
)
check_add_reaction(
user_profile=cordelia,
message_id=message_id,
emoji_name="heart",
emoji_code=None,
reaction_type=None,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, {cordelia.id})
# Add test coverage for 'should_change_visibility_policy' when
# new_visibility_policy == current_visibility_policy
check_add_reaction(
user_profile=cordelia,
message_id=message_id,
emoji_name="tada",
emoji_code=None,
reaction_type=None,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, {cordelia.id})
def test_automatically_follow_topic_on_participation_participate_in_poll(self) -> None:
iago = self.example_user("iago")
hamlet = self.example_user("hamlet")
aaron = self.example_user("aaron")
stream = get_stream("Verona", aaron.realm)
topic_name = "teST topic"
# For iago, 'automatically_follow_topics_policy' set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
iago,
"automatically_follow_topics_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
acting_user=None,
)
# For aaron, 'automatically_follow_topics_policy' NOT set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
aaron,
"automatically_follow_topics_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
acting_user=None,
)
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, set())
# Hamlet creates a poll.
payload = dict(
type="stream",
to=orjson.dumps(stream.name).decode(),
topic=topic_name,
content="/poll Preference?\n\nyes\nno",
)
result = self.api_post(hamlet, "/api/v1/messages", payload)
self.assert_json_success(result)
# Iago participates in the poll. DO automatically follow the topic.
# Aaron participates in the poll. DON'T automatically follow the topic.
message = self.get_last_message()
def participate_in_poll(user: UserProfile, data: Dict[str, object]) -> None:
content = orjson.dumps(data).decode()
payload = dict(
message_id=message.id,
msg_type="widget",
content=content,
)
result = self.api_post(user, "/api/v1/submessage", payload)
self.assert_json_success(result)
participate_in_poll(iago, dict(type="vote", key="1,1", vote=1))
participate_in_poll(aaron, dict(type="new_option", idx=7, option="maybe"))
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, {iago.id})
def test_automatically_follow_topic_on_participation_edit_todo_list(self) -> None:
othello = self.example_user("othello")
hamlet = self.example_user("hamlet")
aaron = self.example_user("aaron")
stream = get_stream("Verona", aaron.realm)
topic_name = "teST topic"
# For othello, 'automatically_follow_topics_policy' set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
othello,
"automatically_follow_topics_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
acting_user=None,
)
# For aaron, 'automatically_follow_topics_policy' NOT set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
aaron,
"automatically_follow_topics_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
acting_user=None,
)
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, set())
# Hamlet creates a todo list.
payload = dict(
type="stream",
to=orjson.dumps(stream.name).decode(),
topic=topic_name,
content="/todo",
)
result = self.api_post(hamlet, "/api/v1/messages", payload)
self.assert_json_success(result)
# Othello edits the todo list. DO automatically follow the topic.
# Aaron edits the todo list. DON'T automatically follow the topic.
message = self.get_last_message()
def edit_todo_list(user: UserProfile, data: Dict[str, object]) -> None:
content = orjson.dumps(data).decode()
payload = dict(
message_id=message.id,
msg_type="widget",
content=content,
)
result = self.api_post(user, "/api/v1/submessage", payload)
self.assert_json_success(result)
edit_todo_list(othello, dict(type="new_task", key=7, task="eat", desc="", completed=False))
edit_todo_list(aaron, dict(type="strike", key="5,9"))
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.FOLLOWED
)
self.assertEqual(user_ids, {othello.id})
class AutomaticallyUnmuteTopicsTests(ZulipTestCase):
def test_automatically_unmute_topic_on_initiation(self) -> None:
hamlet = self.example_user("hamlet")
cordelia = self.example_user("cordelia")
iago = self.example_user("iago")
stream = get_stream("Verona", hamlet.realm)
topic_name = "teST topic"
for user in [hamlet, cordelia, iago]:
sub = get_subscription(stream.name, user)
sub.is_muted = True
sub.save()
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
# For hamlet & cordelia, 'automatically_unmute_topics_in_muted_streams_policy'
# set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION'.
for user in [hamlet, cordelia]:
do_change_user_setting(
user,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
acting_user=None,
)
# Hamlet starts a topic. DO automatically unmute the topic.
self.send_stream_message(hamlet, stream_name=stream.name, topic_name=topic_name)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, {hamlet.id})
# Cordelia sends a message to the topic which hamlet started. DON'T automatically unmute the topic.
self.send_stream_message(cordelia, stream_name=stream.name, topic_name=topic_name)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, {hamlet.id})
# Iago has 'automatically_unmute_topics_in_muted_streams_policy' set to
# 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_NEVER'.
# DON'T automatically unmute the topic, even if he starts the topic.
do_change_user_setting(
iago,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_NEVER,
acting_user=None,
)
self.send_stream_message(iago, stream_name=stream.name, topic_name="New Topic")
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name="New Topic",
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
# When a user sends the first message to a topic with protected history,
# the user starts that topic from their perspective. So, the user
# should unmute the topic if 'automatically_unmute_topics_in_muted_streams_policy'
# is set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION', even if
# the message is not the first message in the topic.
private_stream = self.make_stream(stream_name="private stream", invite_only=True)
self.subscribe(iago, private_stream.name)
self.send_stream_message(iago, private_stream.name)
# Hamlet should automatically unmute the topic, even if it already has messages.
self.subscribe(hamlet, private_stream.name)
sub = get_subscription(private_stream.name, hamlet)
sub.is_muted = True
sub.save()
self.send_stream_message(hamlet, private_stream.name)
stream_topic_target = StreamTopicTarget(
stream_id=private_stream.id,
topic_name="test",
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, {hamlet.id})
def test_automatically_unmute_topic_on_send(self) -> None:
hamlet = self.example_user("hamlet")
aaron = self.example_user("aaron")
stream = get_stream("Verona", hamlet.realm)
topic_name = "teST topic"
self.send_stream_message(aaron, stream.name, "hello", topic_name)
for user in [hamlet, aaron]:
sub = get_subscription(stream.name, user)
sub.is_muted = True
sub.save()
# For hamlet, 'automatically_unmute_topics_in_muted_streams_policy'
# set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_SEND'.
do_change_user_setting(
hamlet,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_SEND,
acting_user=None,
)
# For aaron, 'automatically_unmute_topics_in_muted_streams_policy' NOT
# set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_SEND'.
do_change_user_setting(
aaron,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
acting_user=None,
)
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
# Hamlet sends a message. DO automatically unmute the topic.
# Aaron sends a message. DON'T automatically unmute the topic.
self.send_stream_message(hamlet, stream_name=stream.name, topic_name=topic_name)
self.send_stream_message(aaron, stream_name=stream.name, topic_name=topic_name)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, {hamlet.id})
def test_automatically_unmute_topic_on_participation_send_message(self) -> None:
hamlet = self.example_user("hamlet")
aaron = self.example_user("aaron")
stream = get_stream("Verona", hamlet.realm)
topic_name = "teST topic"
for user in [hamlet, aaron]:
sub = get_subscription(stream.name, user)
sub.is_muted = True
sub.save()
# For hamlet, 'automatically_unmute_topics_in_muted_streams_policy'
# set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
hamlet,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
acting_user=None,
)
# For aaron, 'automatically_unmute_topics_in_muted_streams_policy' NOT
# set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
aaron,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
acting_user=None,
)
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
# Hamlet sends a message. DO automatically unmute the topic.
# Aaron sends a message. DON'T automatically unmute the topic.
self.send_stream_message(hamlet, stream_name=stream.name, topic_name=topic_name)
self.send_stream_message(aaron, stream_name=stream.name, topic_name=topic_name)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, {hamlet.id})
def test_automatically_unmute_topic_on_participation_add_reaction(self) -> None:
cordelia = self.example_user("cordelia")
hamlet = self.example_user("hamlet")
aaron = self.example_user("aaron")
stream = get_stream("Verona", aaron.realm)
topic_name = "teST topic"
for user in [cordelia, hamlet, aaron]:
sub = get_subscription(stream.name, user)
sub.is_muted = True
sub.save()
# For cordelia, 'automatically_unmute_topics_in_muted_streams_policy'
# set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
cordelia,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
acting_user=None,
)
# For aaron, 'automatically_unmute_topics_in_muted_streams_policy' NOT
# set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
aaron,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
acting_user=None,
)
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
message_id = self.send_stream_message(
hamlet, stream_name=stream.name, topic_name=topic_name
)
# Cordelia reacts to a message. DO automatically unmute the topic.
# Aaron reacts to a message. DON'T automatically unmute the topic.
check_add_reaction(
user_profile=cordelia,
message_id=message_id,
emoji_name="smile",
emoji_code=None,
reaction_type=None,
)
check_add_reaction(
user_profile=aaron,
message_id=message_id,
emoji_name="smile",
emoji_code=None,
reaction_type=None,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, {cordelia.id})
def test_automatically_unmute_topic_on_participation_participate_in_poll(self) -> None:
iago = self.example_user("iago")
hamlet = self.example_user("hamlet")
aaron = self.example_user("aaron")
stream = get_stream("Verona", aaron.realm)
topic_name = "teST topic"
for user in [iago, hamlet, aaron]:
sub = get_subscription(stream.name, user)
sub.is_muted = True
sub.save()
# For iago, 'automatically_unmute_topics_in_muted_streams_policy'
# set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
iago,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
acting_user=None,
)
# For aaron, 'automatically_unmute_topics_in_muted_streams_policy' NOT
# set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
aaron,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
acting_user=None,
)
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
# Hamlet creates a poll.
payload = dict(
type="stream",
to=orjson.dumps(stream.name).decode(),
topic=topic_name,
content="/poll Preference?\n\nyes\nno",
)
result = self.api_post(hamlet, "/api/v1/messages", payload)
self.assert_json_success(result)
# Iago participates in the poll. DO automatically unmute the topic.
# Aaron participates in the poll. DON'T automatically unmute the topic.
message = self.get_last_message()
def participate_in_poll(user: UserProfile, data: Dict[str, object]) -> None:
content = orjson.dumps(data).decode()
payload = dict(
message_id=message.id,
msg_type="widget",
content=content,
)
result = self.api_post(user, "/api/v1/submessage", payload)
self.assert_json_success(result)
participate_in_poll(iago, dict(type="vote", key="1,1", vote=1))
participate_in_poll(aaron, dict(type="new_option", idx=7, option="maybe"))
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, {iago.id})
def test_automatically_unmute_topic_on_participation_edit_todo_list(self) -> None:
othello = self.example_user("othello")
hamlet = self.example_user("hamlet")
aaron = self.example_user("aaron")
stream = get_stream("Verona", aaron.realm)
topic_name = "teST topic"
for user in [othello, hamlet, aaron]:
sub = get_subscription(stream.name, user)
sub.is_muted = True
sub.save()
# For othello, 'automatically_unmute_topics_in_muted_streams_policy'
# set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
othello,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
acting_user=None,
)
# For aaron, 'automatically_unmute_topics_in_muted_streams_policy' NOT
# set to 'AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION'.
do_change_user_setting(
aaron,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
acting_user=None,
)
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
# Hamlet creates a todo list.
payload = dict(
type="stream",
to=orjson.dumps(stream.name).decode(),
topic=topic_name,
content="/todo",
)
result = self.api_post(hamlet, "/api/v1/messages", payload)
self.assert_json_success(result)
# Othello edits the todo list. DO automatically unmute the topic.
# Aaron edits the todo list. DON'T automatically unmute the topic.
message = self.get_last_message()
def edit_todo_list(user: UserProfile, data: Dict[str, object]) -> None:
content = orjson.dumps(data).decode()
payload = dict(
message_id=message.id,
msg_type="widget",
content=content,
)
result = self.api_post(user, "/api/v1/submessage", payload)
self.assert_json_success(result)
edit_todo_list(othello, dict(type="new_task", key=7, task="eat", desc="", completed=False))
edit_todo_list(aaron, dict(type="strike", key="5,9"))
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, {othello.id})
def test_only_automatically_increase_visibility_policy(self) -> None:
aaron = self.example_user("aaron")
hamlet = self.example_user("hamlet")
stream = get_stream("Verona", aaron.realm)
topic_name = "teST topic"
for user in [hamlet, aaron]:
sub = get_subscription(stream.name, user)
sub.is_muted = True
sub.save()
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
# If a topic is already FOLLOWED, we don't change the state to UNMUTED as the
# intent of these "automatically follow or unmute" policies is that they can only
# increase the user's visibility policy for the topic.
do_set_user_topic_visibility_policy(
aaron,
stream,
topic_name,
visibility_policy=UserTopic.VisibilityPolicy.FOLLOWED,
)
do_change_user_setting(
aaron,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
acting_user=None,
)
self.send_stream_message(aaron, stream_name=stream.name, topic_name=topic_name)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
# increase visibility from MUTED to UNMUTED
topic_name = "new Topic"
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
do_set_user_topic_visibility_policy(
hamlet,
stream,
topic_name,
visibility_policy=UserTopic.VisibilityPolicy.MUTED,
)
do_change_user_setting(
hamlet,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
acting_user=None,
)
self.send_stream_message(hamlet, stream_name=stream.name, topic_name=topic_name)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, {hamlet.id})
def test_automatically_unmute_policy_unmuted_stream(self) -> None:
aaron = self.example_user("aaron")
cordelia = self.example_user("cordelia")
stream = get_stream("Verona", aaron.realm)
topic_name = "teST topic"
stream_topic_target = StreamTopicTarget(
stream_id=stream.id,
topic_name=topic_name,
)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
# The 'automatically_unmute_topics_in_muted_streams_policy' setting has
# NO effect in unmuted streams.
do_change_user_setting(
aaron,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_INITIATION,
acting_user=None,
)
self.send_stream_message(aaron, stream_name=stream.name, topic_name=topic_name)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
do_set_user_topic_visibility_policy(
cordelia,
stream,
topic_name,
visibility_policy=UserTopic.VisibilityPolicy.MUTED,
)
do_change_user_setting(
cordelia,
"automatically_unmute_topics_in_muted_streams_policy",
UserProfile.AUTOMATICALLY_CHANGE_VISIBILITY_POLICY_ON_PARTICIPATION,
acting_user=None,
)
self.send_stream_message(cordelia, stream_name=stream.name, topic_name=topic_name)
user_ids = stream_topic_target.user_ids_with_visibility_policy(
UserTopic.VisibilityPolicy.UNMUTED
)
self.assertEqual(user_ids, set())
Reference in New Issue View Git Blame Copy Permalink