mirror of
https://github.com/zulip/zulip.git
synced 2025-11-03 21:43:21 +00:00
466 lines
17 KiB
Python
466 lines
17 KiB
Python
import orjson
|
|
|
|
from zerver.lib.subscription_info import gather_subscriptions, gather_subscriptions_helper
|
|
from zerver.lib.test_classes import ZulipTestCase
|
|
from zerver.lib.test_helpers import get_subscription
|
|
from zerver.models import Recipient, Subscription
|
|
|
|
|
|
class SubscriptionPropertiesTest(ZulipTestCase):
|
|
def test_set_stream_color(self) -> None:
|
|
"""
|
|
A POST request to /api/v1/users/me/subscriptions/properties with stream_id and
|
|
color data sets the stream color, and for that stream only. Also, make sure that
|
|
any invalid hex color codes are bounced.
|
|
"""
|
|
test_user = self.example_user("hamlet")
|
|
self.login_user(test_user)
|
|
|
|
old_subs, _ = gather_subscriptions(test_user)
|
|
sub = old_subs[0]
|
|
stream_id = sub["stream_id"]
|
|
new_color = "#ffffff" # TODO: ensure that this is different from old_color
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": "color", "stream_id": stream_id, "value": "#ffffff"}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_success(result)
|
|
|
|
new_subs = gather_subscriptions(test_user)[0]
|
|
found_sub = None
|
|
for sub in new_subs:
|
|
if sub["stream_id"] == stream_id:
|
|
found_sub = sub
|
|
break
|
|
|
|
assert found_sub is not None
|
|
self.assertEqual(found_sub["color"], new_color)
|
|
|
|
new_subs.remove(found_sub)
|
|
for sub in old_subs:
|
|
if sub["stream_id"] == stream_id:
|
|
found_sub = sub
|
|
break
|
|
old_subs.remove(found_sub)
|
|
self.assertEqual(old_subs, new_subs)
|
|
|
|
invalid_color = "3ffrff"
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": "color", "stream_id": stream_id, "value": invalid_color}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_error(
|
|
result, "Invalid subscription_data[0]: Value error, color is not a valid hex color code"
|
|
)
|
|
|
|
def test_set_color_missing_stream_id(self) -> None:
|
|
"""
|
|
Updating the color property requires a `stream_id` key.
|
|
"""
|
|
test_user = self.example_user("hamlet")
|
|
self.login_user(test_user)
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": "color", "value": "#ffffff"}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_error(
|
|
result, 'subscription_data[0]["stream_id"] field is missing: Field required'
|
|
)
|
|
|
|
def test_set_color_unsubscribed_stream_id(self) -> None:
|
|
"""
|
|
Updating the color property requires a subscribed stream.
|
|
"""
|
|
test_user = self.example_user("hamlet")
|
|
self.login_user(test_user)
|
|
|
|
sub_info = gather_subscriptions_helper(test_user)
|
|
|
|
not_subbed = sub_info.never_subscribed
|
|
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[
|
|
{
|
|
"property": "color",
|
|
"stream_id": not_subbed[0]["stream_id"],
|
|
"value": "#ffffff",
|
|
}
|
|
]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_error(
|
|
result, "Not subscribed to channel ID {}".format(not_subbed[0]["stream_id"])
|
|
)
|
|
|
|
def test_set_color_missing_color(self) -> None:
|
|
"""
|
|
Updating the color property requires a color.
|
|
"""
|
|
test_user = self.example_user("hamlet")
|
|
self.login_user(test_user)
|
|
subs = gather_subscriptions(test_user)[0]
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": "color", "stream_id": subs[0]["stream_id"]}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_error(
|
|
result, 'subscription_data[0]["value"] field is missing: Field required'
|
|
)
|
|
|
|
def test_set_stream_wildcard_mentions_notify(self) -> None:
|
|
"""
|
|
A POST request to /api/v1/users/me/subscriptions/properties with wildcard_mentions_notify
|
|
sets the property.
|
|
"""
|
|
test_user = self.example_user("hamlet")
|
|
self.login_user(test_user)
|
|
|
|
subs = gather_subscriptions(test_user)[0]
|
|
sub = subs[0]
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[
|
|
{
|
|
"property": "wildcard_mentions_notify",
|
|
"stream_id": sub["stream_id"],
|
|
"value": True,
|
|
}
|
|
]
|
|
).decode()
|
|
},
|
|
)
|
|
|
|
self.assert_json_success(result)
|
|
|
|
updated_sub = get_subscription(sub["name"], test_user)
|
|
self.assertIsNotNone(updated_sub)
|
|
self.assertEqual(updated_sub.wildcard_mentions_notify, True)
|
|
|
|
def test_set_pin_to_top(self) -> None:
|
|
"""
|
|
A POST request to /api/v1/users/me/subscriptions/properties with stream_id and
|
|
pin_to_top data pins the stream.
|
|
"""
|
|
user = self.example_user("hamlet")
|
|
self.login_user(user)
|
|
|
|
old_subs, _ = gather_subscriptions(user)
|
|
sub = old_subs[0]
|
|
stream_id = sub["stream_id"]
|
|
new_pin_to_top = not sub["pin_to_top"]
|
|
result = self.api_post(
|
|
user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": "pin_to_top", "stream_id": stream_id, "value": new_pin_to_top}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_success(result)
|
|
|
|
updated_sub = get_subscription(sub["name"], user)
|
|
|
|
self.assertIsNotNone(updated_sub)
|
|
self.assertEqual(updated_sub.pin_to_top, new_pin_to_top)
|
|
|
|
def test_change_is_muted(self) -> None:
|
|
test_user = self.example_user("hamlet")
|
|
self.login_user(test_user)
|
|
subs = gather_subscriptions(test_user)[0]
|
|
|
|
sub = Subscription.objects.get(
|
|
recipient__type=Recipient.STREAM,
|
|
recipient__type_id=subs[0]["stream_id"],
|
|
user_profile=test_user,
|
|
)
|
|
self.assertEqual(sub.is_muted, False)
|
|
|
|
property_name = "is_muted"
|
|
with self.capture_send_event_calls(expected_num_events=2) as events:
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[
|
|
{
|
|
"property": property_name,
|
|
"value": True,
|
|
"stream_id": subs[0]["stream_id"],
|
|
}
|
|
]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_success(result)
|
|
self.assertEqual(events[0]["event"]["property"], "in_home_view")
|
|
self.assertEqual(events[0]["event"]["value"], False)
|
|
self.assertEqual(events[1]["event"]["property"], "is_muted")
|
|
self.assertEqual(events[1]["event"]["value"], True)
|
|
sub = Subscription.objects.get(
|
|
recipient__type=Recipient.STREAM,
|
|
recipient__type_id=subs[0]["stream_id"],
|
|
user_profile=test_user,
|
|
)
|
|
self.assertEqual(sub.is_muted, True)
|
|
|
|
legacy_property_name = "in_home_view"
|
|
with self.capture_send_event_calls(expected_num_events=2) as events:
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[
|
|
{
|
|
"property": legacy_property_name,
|
|
"value": True,
|
|
"stream_id": subs[0]["stream_id"],
|
|
}
|
|
]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_success(result)
|
|
self.assertEqual(events[0]["event"]["property"], "in_home_view")
|
|
self.assertEqual(events[0]["event"]["value"], True)
|
|
self.assertEqual(events[1]["event"]["property"], "is_muted")
|
|
self.assertEqual(events[1]["event"]["value"], False)
|
|
self.assert_json_success(result)
|
|
sub = Subscription.objects.get(
|
|
recipient__type=Recipient.STREAM,
|
|
recipient__type_id=subs[0]["stream_id"],
|
|
user_profile=test_user,
|
|
)
|
|
self.assertEqual(sub.is_muted, False)
|
|
|
|
with self.capture_send_event_calls(expected_num_events=2) as events:
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[
|
|
{
|
|
"property": legacy_property_name,
|
|
"value": False,
|
|
"stream_id": subs[0]["stream_id"],
|
|
}
|
|
]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_success(result)
|
|
self.assertEqual(events[0]["event"]["property"], "in_home_view")
|
|
self.assertEqual(events[0]["event"]["value"], False)
|
|
self.assertEqual(events[1]["event"]["property"], "is_muted")
|
|
self.assertEqual(events[1]["event"]["value"], True)
|
|
|
|
sub = Subscription.objects.get(
|
|
recipient__type=Recipient.STREAM,
|
|
recipient__type_id=subs[0]["stream_id"],
|
|
user_profile=test_user,
|
|
)
|
|
self.assertEqual(sub.is_muted, True)
|
|
|
|
def test_set_subscription_property_incorrect(self) -> None:
|
|
"""
|
|
Trying to set a property incorrectly returns a JSON error.
|
|
"""
|
|
test_user = self.example_user("hamlet")
|
|
self.login_user(test_user)
|
|
subs = gather_subscriptions(test_user)[0]
|
|
|
|
property_name = "is_muted"
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": property_name, "value": "bad", "stream_id": subs[0]["stream_id"]}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_error(result, f"{property_name} is not a boolean")
|
|
|
|
property_name = "in_home_view"
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": property_name, "value": "bad", "stream_id": subs[0]["stream_id"]}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_error(result, f"{property_name} is not a boolean")
|
|
|
|
property_name = "desktop_notifications"
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": property_name, "value": "bad", "stream_id": subs[0]["stream_id"]}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_error(result, f"{property_name} is not a boolean")
|
|
|
|
property_name = "audible_notifications"
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": property_name, "value": "bad", "stream_id": subs[0]["stream_id"]}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_error(result, f"{property_name} is not a boolean")
|
|
|
|
property_name = "push_notifications"
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": property_name, "value": "bad", "stream_id": subs[0]["stream_id"]}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_error(result, f"{property_name} is not a boolean")
|
|
|
|
property_name = "email_notifications"
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": property_name, "value": "bad", "stream_id": subs[0]["stream_id"]}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_error(result, f"{property_name} is not a boolean")
|
|
|
|
property_name = "wildcard_mentions_notify"
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": property_name, "value": "bad", "stream_id": subs[0]["stream_id"]}]
|
|
).decode()
|
|
},
|
|
)
|
|
|
|
self.assert_json_error(result, f"{property_name} is not a boolean")
|
|
|
|
property_name = "color"
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": property_name, "value": False, "stream_id": subs[0]["stream_id"]}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_error(
|
|
result, "Invalid subscription_data[0]: Value error, color is not a valid hex color code"
|
|
)
|
|
|
|
def test_json_subscription_property_invalid_stream(self) -> None:
|
|
test_user = self.example_user("hamlet")
|
|
self.login_user(test_user)
|
|
|
|
stream_id = 1000
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": "is_muted", "stream_id": stream_id, "value": False}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_error(result, "Invalid channel ID")
|
|
|
|
def test_set_invalid_property(self) -> None:
|
|
"""
|
|
Trying to set an invalid property returns a JSON error.
|
|
"""
|
|
test_user = self.example_user("hamlet")
|
|
self.login_user(test_user)
|
|
subs = gather_subscriptions(test_user)[0]
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[{"property": "bad", "value": "bad", "stream_id": subs[0]["stream_id"]}]
|
|
).decode()
|
|
},
|
|
)
|
|
self.assert_json_error(result, "Unknown subscription property: bad")
|
|
|
|
def test_ignored_parameters_in_subscriptions_properties_endpoint(self) -> None:
|
|
"""
|
|
Sending an invalid parameter with a valid parameter returns
|
|
an `ignored_parameters_unsupported` array.
|
|
"""
|
|
test_user = self.example_user("hamlet")
|
|
self.login_user(test_user)
|
|
|
|
subs = gather_subscriptions(test_user)[0]
|
|
sub = subs[0]
|
|
result = self.api_post(
|
|
test_user,
|
|
"/api/v1/users/me/subscriptions/properties",
|
|
{
|
|
"subscription_data": orjson.dumps(
|
|
[
|
|
{
|
|
"property": "wildcard_mentions_notify",
|
|
"stream_id": sub["stream_id"],
|
|
"value": True,
|
|
}
|
|
]
|
|
).decode(),
|
|
"invalid_parameter": orjson.dumps(
|
|
[{"property": "pin_to_top", "stream_id": sub["stream_id"], "value": False}]
|
|
).decode(),
|
|
},
|
|
)
|
|
|
|
self.assert_json_success(result, ignored_parameters=["invalid_parameter"])
|