mirror of
https://github.com/zulip/zulip.git
synced 2025-10-30 03:23:50 +00:00
042bbf2936
This commit updates 'set_user_topic_visibility_policy_in_database' to not raise an error when deleting a UserTopic row and the user doesn't have a visibility_policy for the topic yet, or when setting the visibility_policy to its current value. Also, it includes the changes to not send unnecessary events in such cases.
278 lines
11 KiB
Python
278 lines
11 KiB
Python
from datetime import datetime, timezone
|
|
from typing import Any, Dict, List
|
|
from unittest import mock
|
|
|
|
from django.utils.timezone import now as timezone_now
|
|
|
|
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.user_topics import (
|
|
get_topic_mutes,
|
|
topic_is_muted,
|
|
)
|
|
from zerver.models import UserProfile, UserTopic, get_stream
|
|
|
|
|
|
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)
|
|
|
|
mock_date_muted = datetime(2020, 1, 1, tzinfo=timezone.utc).timestamp()
|
|
|
|
do_set_user_topic_visibility_policy(
|
|
user,
|
|
stream,
|
|
"Verona3",
|
|
visibility_policy=UserTopic.VisibilityPolicy.MUTED,
|
|
last_updated=datetime(2020, 1, 1, tzinfo=timezone.utc),
|
|
)
|
|
|
|
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())
|
|
|
|
def mute_topic_for_user(user: UserProfile) -> None:
|
|
do_set_user_topic_visibility_policy(
|
|
user,
|
|
stream,
|
|
"test TOPIC",
|
|
visibility_policy=UserTopic.VisibilityPolicy.MUTED,
|
|
last_updated=date_muted,
|
|
)
|
|
|
|
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 mock.patch(
|
|
"zerver.views.user_topics.timezone_now",
|
|
return_value=datetime(2020, 1, 1, tzinfo=timezone.utc),
|
|
):
|
|
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_is_muted(user, stream.id, "verona3"))
|
|
|
|
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_is_muted(user, stream.id, "verona3"))
|
|
|
|
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 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())
|
|
|
|
def set_topic_visibility_for_user(user: UserProfile, visibility_policy: int) -> None:
|
|
do_set_user_topic_visibility_policy(
|
|
user,
|
|
stream,
|
|
"test TOPIC",
|
|
visibility_policy=visibility_policy,
|
|
last_updated=date_unmuted,
|
|
)
|
|
|
|
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)
|