paulmataruso/zulip
1
0
Fork 0
mirror of https://github.com/zulip/zulip.git synced 2025-11-03 13:33:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

streams: Do not send update event to users gaining metadata access.

Browse Source
This commit is contained in:
Shubham Padia
2025-03-18 11:29:14 +00:00
committed by Tim Abbott
parent 0c7c13fc30
commit 25f91a51ec
3 changed files with 214 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
zerver/tests/test_subs.py
View File

@@ -84,7 +84,6 @@ from zerver.lib.streams import (
list_to_streams,
public_stream_user_ids,
user_has_content_access,
user_has_metadata_access,
)
from zerver.lib.subscription_info import (
bulk_get_subscriber_user_ids,
@@ -2989,160 +2988,6 @@ class StreamAdminTest(ZulipTestCase):
for setting_name in Stream.stream_permission_group_settings:
self.do_test_change_stream_permission_setting(setting_name)
def do_test_events_on_changing_private_stream_permission_settings_granting_metadata_access(
self, setting_name: str
) -> None:
iago = self.example_user("iago")
aaron = self.example_user("aaron")
private_stream = get_stream("private_stream", iago.realm)
self.login_user(iago)
params = {}
self.assertFalse(
user_has_metadata_access(
aaron,
private_stream,
UserGroupMembershipDetails(user_recursive_group_ids=None),
is_subscribed=False,
)
)
params[setting_name] = orjson.dumps(
{
"new": {
"direct_members": [aaron.id],
"direct_subgroups": [],
},
}
).decode()
with self.capture_send_event_calls(expected_num_events=2) as events:
result = self.client_patch(
f"/json/streams/{private_stream.id}",
params,
)
self.assert_json_success(result)
event = events[0]["event"]
self.assertEqual(event["type"], "stream")
self.assertEqual(event["op"], "update")
self.assertEqual(event["stream_id"], private_stream.id)
event = events[1]["event"]
self.assertEqual(event["type"], "stream")
self.assertEqual(event["op"], "create")
self.assertEqual(event["streams"][0]["stream_id"], private_stream.id)
nobody_group = NamedUserGroup.objects.get(
name=SystemGroups.NOBODY, realm=iago.realm, is_system_group=True
)
private_stream = get_stream("private_stream", iago.realm)
self.assertTrue(
user_has_metadata_access(
aaron,
private_stream,
UserGroupMembershipDetails(user_recursive_group_ids=None),
is_subscribed=False,
)
)
params[setting_name] = orjson.dumps(
{
"new": nobody_group.id,
}
).decode()
with self.capture_send_event_calls(expected_num_events=2) as events:
result = self.client_patch(
f"/json/streams/{private_stream.id}",
params,
)
self.assert_json_success(result)
event = events[0]["event"]
self.assertEqual(event["type"], "stream")
self.assertEqual(event["op"], "update")
self.assertEqual(event["stream_id"], private_stream.id)
event = events[1]["event"]
self.assertEqual(event["type"], "stream")
self.assertEqual(event["op"], "delete")
self.assertEqual(event["streams"][0]["stream_id"], private_stream.id)
def do_test_events_on_changing_private_stream_permission_settings_not_granting_metadata_access(
self, setting_name: str
) -> None:
iago = self.example_user("iago")
aaron = self.example_user("aaron")
private_stream = get_stream("private_stream", iago.realm)
params = {}
self.login_user(iago)
expected_num_events = 1
if setting_name == "can_send_message_group":
expected_num_events = 2
self.assertFalse(
user_has_metadata_access(
aaron,
private_stream,
UserGroupMembershipDetails(user_recursive_group_ids=None),
is_subscribed=False,
)
)
params[setting_name] = orjson.dumps(
{
"new": {
"direct_members": [aaron.id],
"direct_subgroups": [],
},
}
).decode()
with self.capture_send_event_calls(expected_num_events=expected_num_events) as events:
result = self.client_patch(
f"/json/streams/{private_stream.id}",
params,
)
self.assert_json_success(result)
event = events[0]["event"]
self.assertEqual(event["type"], "stream")
self.assertEqual(event["op"], "update")
self.assertEqual(event["stream_id"], private_stream.id)
nobody_group = NamedUserGroup.objects.get(
name=SystemGroups.NOBODY, realm=iago.realm, is_system_group=True
)
private_stream = get_stream("private_stream", iago.realm)
self.assertFalse(
user_has_metadata_access(
aaron,
private_stream,
UserGroupMembershipDetails(user_recursive_group_ids=None),
is_subscribed=False,
)
)
params[setting_name] = orjson.dumps(
{
"new": nobody_group.id,
}
).decode()
with self.capture_send_event_calls(expected_num_events=expected_num_events) as events:
result = self.client_patch(
f"/json/streams/{private_stream.id}",
params,
)
self.assert_json_success(result)
event = events[0]["event"]
self.assertEqual(event["type"], "stream")
self.assertEqual(event["op"], "update")
self.assertEqual(event["stream_id"], private_stream.id)
def test_events_on_changing_private_stream_permission_settings(self) -> None:
self.make_stream("private_stream", invite_only=True)
self.subscribe(self.example_user("iago"), "private_stream")
for setting_name in Stream.stream_permission_group_settings:
if setting_name in Stream.stream_permission_group_settings_granting_metadata_access:
self.do_test_events_on_changing_private_stream_permission_settings_granting_metadata_access(
setting_name
)
else:
self.do_test_events_on_changing_private_stream_permission_settings_not_granting_metadata_access(
setting_name
)
def test_notification_on_changing_stream_posting_permission(self) -> None:
desdemona = self.example_user("desdemona")
realm = desdemona.realm