diff --git a/zerver/tests/test_channel_permissions.py b/zerver/tests/test_channel_permissions.py index fd2e963c65..98e18d7e55 100644 --- a/zerver/tests/test_channel_permissions.py +++ b/zerver/tests/test_channel_permissions.py @@ -1,16 +1,25 @@ +from typing import TypedDict + import orjson from typing_extensions import override +from zerver.actions.channel_folders import check_add_channel_folder from zerver.actions.realm_settings import ( do_change_realm_permission_group_setting, do_set_realm_property, ) -from zerver.actions.streams import do_change_stream_group_based_setting +from zerver.actions.streams import ( + do_change_stream_group_based_setting, + do_change_stream_permission, + do_deactivate_stream, +) from zerver.actions.user_groups import add_subgroups_to_user_group, check_add_user_group from zerver.actions.users import do_change_user_role +from zerver.lib.streams import subscribed_to_stream from zerver.lib.test_classes import ZulipTestCase from zerver.lib.test_helpers import get_subscription from zerver.lib.types import UserGroupMembersData +from zerver.lib.user_groups import get_group_setting_value_for_api from zerver.models import NamedUserGroup, Recipient, Stream, Subscription, UserProfile from zerver.models.groups import SystemGroups from zerver.models.realms import get_realm @@ -746,3 +755,506 @@ class ChannelSubscriptionPermissionTest(ZulipTestCase): self.assert_json_success(result) stream = get_stream("stream_name1", realm) self.assertEqual(stream.message_retention_days, 2) + + +class PermissionCheckConfigDict(TypedDict): + setting_group: NamedUserGroup | UserGroupMembersData + users_with_permission: list[UserProfile] + users_without_permission: list[UserProfile] + + +class ChannelAdministerPermissionTest(ZulipTestCase): + @override + def setUp(self) -> None: + super().setUp() + self.realm = get_realm("zulip") + self.admin = self.example_user("iago") + self.moderator = self.example_user("shiva") + self.guest = self.example_user("polonius") + + self.hamletcharacters_group = NamedUserGroup.objects.get( + name="hamletcharacters", realm=self.realm + ) + self.moderators_group = NamedUserGroup.objects.get( + name=SystemGroups.MODERATORS, realm=self.realm, is_system_group=True + ) + self.nobody_group = NamedUserGroup.objects.get( + name=SystemGroups.NOBODY, realm=self.realm, is_system_group=True + ) + self.members_group = NamedUserGroup.objects.get( + name=SystemGroups.MEMBERS, realm=self.realm, is_system_group=True + ) + + def do_test_updating_channel( + self, stream: Stream, property_name: str, new_value: str | int | bool + ) -> None: + hamlet = self.example_user("hamlet") + prospero = self.example_user("prospero") + + # For some properties, name of the field in Stream model + # is different from parameter name used in API request. + api_parameter_name_dict = dict( + name="new_name", + deactivated="is_archived", + ) + api_parameter_name = property_name + if property_name in api_parameter_name_dict: + api_parameter_name = api_parameter_name_dict[property_name] + + data = {} + if not isinstance(new_value, str): + data[api_parameter_name] = orjson.dumps(new_value).decode() + else: + data[api_parameter_name] = new_value + + default_error_msg = "You do not have permission to administer this channel." + + def check_channel_property_update( + user: UserProfile, allow_fail: bool = False, error_msg: str = default_error_msg + ) -> None: + old_value = getattr(stream, property_name) + + result = self.api_patch(user, f"/api/v1/streams/{stream.id}", info=data) + + if allow_fail: + self.assert_json_error(result, error_msg) + return + + self.assert_json_success(result) + stream.refresh_from_db() + self.assertEqual(getattr(stream, property_name), new_value) + + # Reset to original value. + setattr(stream, property_name, old_value) + stream.save(update_fields=[property_name]) + + anonymous_group_dict = UserGroupMembersData( + direct_members=[prospero.id, self.guest.id], direct_subgroups=[] + ) + + group_permission_checks: list[PermissionCheckConfigDict] = [ + # Check admin can always administer channel. + PermissionCheckConfigDict( + setting_group=self.nobody_group, + users_without_permission=[self.moderator], + users_with_permission=[self.admin], + ), + # Check case when can_administer_channel_group is set to a system group. + PermissionCheckConfigDict( + setting_group=self.moderators_group, + users_without_permission=[hamlet], + users_with_permission=[self.moderator], + ), + # Check case when can_administer_channel_group is set to a user-defined group. + PermissionCheckConfigDict( + setting_group=self.hamletcharacters_group, + users_without_permission=[self.moderator], + users_with_permission=[hamlet], + ), + # Check case when can_administer_channel_group is set to an anonymous group. + PermissionCheckConfigDict( + setting_group=anonymous_group_dict, + users_without_permission=[self.moderator, hamlet], + users_with_permission=[prospero], + ), + ] + + for check_config in group_permission_checks: + do_change_stream_group_based_setting( + stream, + "can_administer_channel_group", + check_config["setting_group"], + acting_user=self.admin, + ) + + for user in check_config["users_without_permission"]: + error_msg = default_error_msg + if stream.invite_only and not subscribed_to_stream(user, stream.id): + # In private streams, users that are not subscribed cannot access + # the stream. + error_msg = "Invalid channel ID" + check_channel_property_update(user, allow_fail=True, error_msg=error_msg) + + for user in check_config["users_with_permission"]: + check_channel_property_update(user) + + # Check guests cannot update property even when they belong + # to "can_administer_channel_group". + check_channel_property_update(self.guest, allow_fail=True, error_msg="Invalid channel ID") + self.subscribe(self.guest, stream.name) + check_channel_property_update(self.guest, allow_fail=True) + self.unsubscribe(self.guest, stream.name) + + def test_administering_permission_for_updating_channel(self) -> None: + """ + This test is only for checking permission to update basic channel + properties like name, description, folder and permission to unarchive + the channel. Other things like permission to update group settings and + channel privacy are tested separately. + """ + public_stream = self.make_stream("test stream") + + private_stream = self.make_stream("private_stream", invite_only=True) + self.subscribe(self.admin, private_stream.name) + self.subscribe(self.moderator, private_stream.name) + self.subscribe(self.example_user("hamlet"), private_stream.name) + self.subscribe(self.example_user("prospero"), private_stream.name) + + unsubscribed_private_stream = self.make_stream( + "unsubscribed_private_stream", invite_only=True + ) + # Subscribing a user that is not used in this test, as we do not allow + # unarchiving vacant private streams. + self.subscribe(self.example_user("desdemona"), unsubscribed_private_stream.name) + + channel_folder = check_add_channel_folder( + self.realm, "Frontend", "", acting_user=self.admin + ) + + for stream in [public_stream, private_stream, unsubscribed_private_stream]: + self.do_test_updating_channel(stream, "name", "Renamed stream") + self.do_test_updating_channel(stream, "description", "Edited stream description") + self.do_test_updating_channel(stream, "folder_id", channel_folder.id) + + do_deactivate_stream(stream, acting_user=None) + self.do_test_updating_channel(stream, "deactivated", False) + + def check_channel_privacy_update( + self, user: UserProfile, property_name: str, new_value: bool, error_msg: str | None = None + ) -> None: + stream = get_stream("test_stream", user.realm) + data = {} + if property_name == "invite_only": + data["is_private"] = orjson.dumps(new_value).decode() + else: + data[property_name] = orjson.dumps(new_value).decode() + + old_value = getattr(stream, property_name) + result = self.api_patch(user, f"/api/v1/streams/{stream.id}", info=data) + + if error_msg is not None: + self.assert_json_error(result, error_msg) + return + + self.assert_json_success(result) + stream.refresh_from_db() + self.assertEqual(getattr(stream, property_name), new_value) + + # Reset to original value. + setattr(stream, property_name, old_value) + stream.save(update_fields=[property_name]) + + # Reset history_public_to_subscribers field when stream + # is changed from private to public. + if not stream.invite_only and not stream.history_public_to_subscribers: + stream.history_public_to_subscribers = True + stream.save(update_fields=["history_public_to_subscribers"]) + + def do_test_updating_channel_privacy(self, property_name: str, new_value: bool) -> None: + hamlet = self.example_user("hamlet") + prospero = self.example_user("prospero") + + stream = get_stream("test_stream", self.realm) + + data = {} + if property_name == "invite_only": + data["is_private"] = orjson.dumps(new_value).decode() + else: + data[property_name] = orjson.dumps(new_value).decode() + + default_error_msg = "You do not have permission to administer this channel." + + anonymous_group_dict = UserGroupMembersData( + direct_members=[prospero.id, self.guest.id], direct_subgroups=[] + ) + + group_permission_checks: list[PermissionCheckConfigDict] = [ + # Check admin can always administer channel. + PermissionCheckConfigDict( + setting_group=self.nobody_group, + users_without_permission=[self.moderator], + users_with_permission=[self.admin], + ), + # Check case when can_administer_channel_group is set to a system group. + PermissionCheckConfigDict( + setting_group=self.moderators_group, + users_without_permission=[hamlet], + users_with_permission=[self.moderator], + ), + # Check case when can_administer_channel_group is set to a user-defined group. + PermissionCheckConfigDict( + setting_group=self.hamletcharacters_group, + users_without_permission=[self.moderator], + users_with_permission=[hamlet], + ), + # Check case when can_administer_channel_group is set to an anonymous group. + PermissionCheckConfigDict( + setting_group=anonymous_group_dict, + users_without_permission=[self.moderator, hamlet], + users_with_permission=[prospero], + ), + ] + + for check_config in group_permission_checks: + do_change_stream_group_based_setting( + stream, + "can_administer_channel_group", + check_config["setting_group"], + acting_user=self.admin, + ) + + for user in check_config["users_without_permission"]: + self.check_channel_privacy_update(user, property_name, new_value, default_error_msg) + + for user in check_config["users_with_permission"]: + self.check_channel_privacy_update(user, property_name, new_value) + + # Check guests cannot update property even when they belong + # to "can_administer_channel_group". + self.check_channel_privacy_update( + self.guest, property_name, new_value, error_msg="Invalid channel ID" + ) + self.subscribe(self.guest, stream.name) + self.check_channel_privacy_update(self.guest, property_name, new_value, default_error_msg) + self.unsubscribe(self.guest, stream.name) + + def test_administering_permission_for_updating_channel_privacy(self) -> None: + stream = self.make_stream("test_stream") + + # Give permission to create web-public channels to everyone, so + # that we can check administering permissions easily, although + # we do not do not allow members to create web-public channels + # in production. + do_change_realm_permission_group_setting( + self.realm, "can_create_web_public_channel_group", self.members_group, acting_user=None + ) + + # Test making a public stream private with protected history. + self.do_test_updating_channel_privacy("invite_only", True) + + # Test making a public stream web-public. + self.do_test_updating_channel_privacy("is_web_public", True) + + do_change_stream_permission( + stream, + invite_only=True, + history_public_to_subscribers=False, + is_web_public=False, + acting_user=self.admin, + ) + self.subscribe(self.admin, stream.name) + self.subscribe(self.moderator, stream.name) + self.subscribe(self.example_user("hamlet"), stream.name) + self.subscribe(self.example_user("prospero"), stream.name) + + # Test making a private stream with protected history public. + self.do_test_updating_channel_privacy("invite_only", False) + + def test_permission_for_updating_privacy_of_unsubscribed_private_channel(self) -> None: + hamlet = self.example_user("hamlet") + + stream = self.make_stream("test_stream", invite_only=True) + + do_change_stream_group_based_setting( + stream, "can_administer_channel_group", self.members_group, acting_user=self.admin + ) + + error_msg = "Channel content access is required." + self.check_channel_privacy_update(self.admin, "invite_only", False, error_msg) + self.check_channel_privacy_update(self.moderator, "invite_only", False, error_msg) + self.check_channel_privacy_update(hamlet, "invite_only", False, error_msg) + + do_change_stream_group_based_setting( + stream, "can_add_subscribers_group", self.moderators_group, acting_user=self.admin + ) + self.check_channel_privacy_update(hamlet, "invite_only", False, error_msg=error_msg) + self.check_channel_privacy_update(self.admin, "invite_only", False) + self.check_channel_privacy_update(self.moderator, "invite_only", False) + + # Users who are part of can_subscribe_group get content access + # to a private stream even if they are not subscribed to it. + do_change_stream_group_based_setting( + stream, "can_subscribe_group", self.hamletcharacters_group, acting_user=self.admin + ) + self.check_channel_privacy_update(hamlet, "invite_only", False) + + def check_channel_group_setting_update( + self, user: UserProfile, property_name: str, error_msg: str | None = None + ) -> None: + stream = get_stream("test_stream", user.realm) + new_value = self.hamletcharacters_group.id + + data = {} + data[property_name] = orjson.dumps({"new": new_value}).decode() + + old_value = getattr(stream, property_name) + # old_value is stored as UserGroupMembersData dict if the + # setting is set to an anonymous group and a NamedUserGroup + # object otherwise, so that we can pass it directly to + # do_change_stream_group_based_setting. + if not hasattr(old_value, "named_user_group"): + old_value = get_group_setting_value_for_api(old_value) + else: + old_value = old_value.named_user_group + + result = self.api_patch(user, f"/api/v1/streams/{stream.id}", info=data) + + if error_msg is not None: + self.assert_json_error(result, error_msg) + return + + self.assert_json_success(result) + stream.refresh_from_db() + self.assertEqual(getattr(stream, property_name + "_id"), new_value) + + # Reset to original value. + do_change_stream_group_based_setting( + stream, property_name, old_value, acting_user=self.example_user("iago") + ) + + def do_test_updating_channel_group_settings(self, property_name: str) -> None: + hamlet = self.example_user("hamlet") + prospero = self.example_user("prospero") + + stream = get_stream("test_stream", self.realm) + + default_error_msg = "You do not have permission to administer this channel." + + anonymous_group_dict = UserGroupMembersData( + direct_members=[prospero.id, self.guest.id], direct_subgroups=[] + ) + + group_permission_checks: list[PermissionCheckConfigDict] = [ + # Check admin can always administer channel. + PermissionCheckConfigDict( + setting_group=self.nobody_group, + users_without_permission=[self.moderator], + users_with_permission=[self.admin], + ), + # Check case when can_administer_channel_group is set to a system group. + PermissionCheckConfigDict( + setting_group=self.moderators_group, + users_without_permission=[hamlet], + users_with_permission=[self.moderator], + ), + # Check case when can_administer_channel_group is set to a user-defined group. + PermissionCheckConfigDict( + setting_group=self.hamletcharacters_group, + users_without_permission=[self.moderator], + users_with_permission=[hamlet], + ), + # Check case when can_administer_channel_group is set to an anonymous group. + PermissionCheckConfigDict( + setting_group=anonymous_group_dict, + users_without_permission=[self.moderator, hamlet], + users_with_permission=[prospero], + ), + ] + + for check_config in group_permission_checks: + do_change_stream_group_based_setting( + stream, + "can_administer_channel_group", + check_config["setting_group"], + acting_user=self.admin, + ) + + for user in check_config["users_without_permission"]: + self.check_channel_group_setting_update(user, property_name, default_error_msg) + + for user in check_config["users_with_permission"]: + self.check_channel_group_setting_update(user, property_name) + + # Check guests cannot update property even when they belong + # to "can_administer_channel_group". + self.check_channel_group_setting_update( + self.guest, property_name, error_msg="Invalid channel ID" + ) + self.subscribe(self.guest, stream.name) + self.check_channel_group_setting_update(self.guest, property_name, default_error_msg) + self.unsubscribe(self.guest, stream.name) + + def do_test_updating_group_settings_for_unsubscribed_private_channels( + self, property_name: str + ) -> None: + # If stream is private, test which permissions require having + # content access to the channel. + hamlet = self.example_user("hamlet") + + stream = get_stream("test_stream", self.realm) + self.assertTrue(stream.invite_only) + + do_change_stream_group_based_setting( + stream, "can_administer_channel_group", self.members_group, acting_user=self.admin + ) + + if property_name not in Stream.stream_permission_group_settings_requiring_content_access: + # Users without content access can modify properties not in + # stream_permission_group_settings_requiring_content_access. + self.check_channel_group_setting_update(self.admin, property_name) + self.check_channel_group_setting_update(self.moderator, property_name) + self.check_channel_group_setting_update(hamlet, property_name) + return + + error_msg = "Channel content access is required." + # Even realm and channel admins need content access to + # a private channel to update the permissions in + # stream_permission_group_settings_requiring_content_access. + self.check_channel_group_setting_update(self.admin, property_name, error_msg=error_msg) + self.check_channel_group_setting_update(self.moderator, property_name, error_msg=error_msg) + self.check_channel_group_setting_update(hamlet, property_name, error_msg=error_msg) + + # Users who are part of can_add_subscribers_group get content access + # to a private stream even if they are not subscribed to it. + do_change_stream_group_based_setting( + stream, "can_add_subscribers_group", self.moderators_group, acting_user=self.admin + ) + self.check_channel_group_setting_update(hamlet, property_name, error_msg=error_msg) + self.check_channel_group_setting_update(self.admin, property_name) + self.check_channel_group_setting_update(self.moderator, property_name) + + # Users who are part of can_subscribe_group get content access + # to a private stream even if they are not subscribed to it. + do_change_stream_group_based_setting( + stream, "can_subscribe_group", self.hamletcharacters_group, acting_user=self.admin + ) + self.check_channel_group_setting_update(hamlet, property_name) + + # Reset the setting values to "Nobody" group. + do_change_stream_group_based_setting( + stream, "can_add_subscribers_group", self.nobody_group, acting_user=self.admin + ) + do_change_stream_group_based_setting( + stream, "can_subscribe_group", self.nobody_group, acting_user=self.admin + ) + + def test_administering_permission_for_updating_channel_group_settings(self) -> None: + stream = self.make_stream("test_stream") + hamlet = self.example_user("hamlet") + prospero = self.example_user("prospero") + + for setting_name in Stream.stream_permission_group_settings: + self.do_test_updating_channel_group_settings(setting_name) + + # Test changing group settings for a private stream when user is + # subscribed to the stream. + do_change_stream_permission( + stream, + invite_only=True, + history_public_to_subscribers=False, + is_web_public=False, + acting_user=self.admin, + ) + for user in [self.admin, self.moderator, hamlet, prospero]: + self.subscribe(user, stream.name) + + for setting_name in Stream.stream_permission_group_settings: + self.do_test_updating_channel_group_settings(setting_name) + + # Unsubscribe user from private stream to test gaining + # content access from group settings. + for user in [self.admin, self.moderator, hamlet, prospero]: + self.unsubscribe(user, stream.name) + + for setting_name in Stream.stream_permission_group_settings: + self.do_test_updating_group_settings_for_unsubscribed_private_channels(setting_name) diff --git a/zerver/tests/test_subs.py b/zerver/tests/test_subs.py index 4958a0d14c..d69aa3dba7 100644 --- a/zerver/tests/test_subs.py +++ b/zerver/tests/test_subs.py @@ -293,29 +293,6 @@ class StreamAdminTest(ZulipTestCase): self.assert_json_error(result, "Moderation request channel must be private.") self.assertTrue(private_stream.invite_only) - do_change_user_role(user_profile, UserProfile.ROLE_MEMBER, acting_user=None) - params = { - "is_private": orjson.dumps(False).decode(), - } - stream = self.subscribe(user_profile, "private_stream_2") - self.assertFalse(is_user_in_group(stream.can_administer_channel_group_id, user_profile)) - result = self.client_patch(f"/json/streams/{stream.id}", params) - self.assertTrue(stream.invite_only) - self.assert_json_error(result, "You do not have permission to administer this channel.") - - user_profile_group = check_add_user_group( - realm, "user_profile_group", [user_profile], acting_user=user_profile - ) - do_change_stream_group_based_setting( - stream, - "can_administer_channel_group", - user_profile_group, - acting_user=user_profile, - ) - result = self.client_patch(f"/json/streams/{stream.id}", params) - self.assertTrue(stream.invite_only) - self.assert_json_success(result) - stream = self.subscribe(user_profile, "private_stream_3", invite_only=True) do_change_user_role(user_profile, UserProfile.ROLE_REALM_OWNER, acting_user=None) nobody_group = NamedUserGroup.objects.get( @@ -330,6 +307,9 @@ class StreamAdminTest(ZulipTestCase): result = self.client_patch(f"/json/streams/{stream.id}", params) self.assert_json_error(result, "Insufficient permission") + user_profile_group = check_add_user_group( + realm, "user_profile_group", [user_profile], acting_user=user_profile + ) do_change_realm_permission_group_setting( realm, "can_create_public_channel_group", @@ -402,29 +382,6 @@ class StreamAdminTest(ZulipTestCase): self.assert_json_error(result, "A default channel cannot be private.") self.assertFalse(default_stream.invite_only) - do_change_user_role(user_profile, UserProfile.ROLE_MEMBER, acting_user=None) - params = { - "is_private": orjson.dumps(True).decode(), - } - stream = self.subscribe(user_profile, "public_stream_2") - self.assertFalse(is_user_in_group(stream.can_administer_channel_group_id, user_profile)) - result = self.client_patch(f"/json/streams/{stream.id}", params) - self.assertFalse(stream.invite_only) - self.assert_json_error(result, "You do not have permission to administer this channel.") - - user_profile_group = check_add_user_group( - realm, "user_profile_group", [user_profile], acting_user=user_profile - ) - do_change_stream_group_based_setting( - stream, - "can_administer_channel_group", - user_profile_group, - acting_user=user_profile, - ) - result = self.client_patch(f"/json/streams/{stream.id}", params) - self.assertFalse(stream.invite_only) - self.assert_json_success(result) - stream = self.subscribe(user_profile, "public_stream_3") do_change_user_role(user_profile, UserProfile.ROLE_REALM_OWNER, acting_user=None) nobody_group = NamedUserGroup.objects.get( @@ -439,6 +396,9 @@ class StreamAdminTest(ZulipTestCase): result = self.client_patch(f"/json/streams/{stream.id}", params) self.assert_json_error(result, "Insufficient permission") + user_profile_group = check_add_user_group( + realm, "user_profile_group", [user_profile], acting_user=user_profile + ) do_change_realm_permission_group_setting( realm, "can_create_private_channel_group", @@ -624,14 +584,20 @@ class StreamAdminTest(ZulipTestCase): stream = self.subscribe(user_profile, "test_stream") stream_id = stream.id + user_profile_group = check_add_user_group( + realm, "user_profile_group", [user_profile], acting_user=user_profile + ) + do_change_stream_group_based_setting( + stream, + "can_administer_channel_group", + user_profile_group, + acting_user=user_profile, + ) + params = { "is_web_public": orjson.dumps(True).decode(), "history_public_to_subscribers": orjson.dumps(True).decode(), } - self.assertFalse(is_user_in_group(stream.can_administer_channel_group_id, user_profile)) - result = self.client_patch(f"/json/streams/{stream_id}", params) - self.assert_json_error(result, "You do not have permission to administer this channel.") - owners_group = NamedUserGroup.objects.get( name=SystemGroups.OWNERS, realm=realm, is_system_group=True ) @@ -716,37 +682,6 @@ class StreamAdminTest(ZulipTestCase): } self.assertEqual(realm_audit_log.extra_data, expected_extra_data) - # Test non-admin belonging to can_administer_channel_group - # can also make the stream public. - do_change_user_role(user_profile, UserProfile.ROLE_MEMBER, acting_user=None) - stream = self.make_stream("test_stream_1", realm=realm) - stream_id = self.subscribe(user_profile, "test_stream_1").id - user_profile_group = check_add_user_group( - realm, "user_profile_group", [user_profile], acting_user=user_profile - ) - do_change_realm_permission_group_setting( - realm, - "can_create_web_public_channel_group", - user_profile_group, - acting_user=None, - ) - params = { - "is_web_public": orjson.dumps(True).decode(), - "history_public_to_subscribers": orjson.dumps(True).decode(), - } - self.assertFalse(is_user_in_group(stream.can_administer_channel_group_id, user_profile)) - result = self.client_patch(f"/json/streams/{stream_id}", params) - self.assert_json_error(result, "You do not have permission to administer this channel.") - - do_change_stream_group_based_setting( - stream, - "can_administer_channel_group", - user_profile_group, - acting_user=user_profile, - ) - result = self.client_patch(f"/json/streams/{stream_id}", params) - self.assert_json_success(result) - def test_change_history_access_for_private_streams(self) -> None: user_profile = self.example_user("iago") self.login_user(user_profile) @@ -1411,52 +1346,6 @@ class StreamAdminTest(ZulipTestCase): }, ) - def test_permissions_and_archiving_behavior(self) -> None: - """ - Test permissions for archiving and unarchiving streams, and ensure users without - the necessary permissions cannot archive or unarchive a stream. - """ - desdemona = self.example_user("desdemona") - iago = self.example_user("iago") - - stream = self.make_stream("test_stream", invite_only=False) - self.subscribe(iago, stream.name) - - do_deactivate_stream(stream, acting_user=None) - stream.refresh_from_db() - self.assertTrue(stream.deactivated) - - data = {} - data["is_archived"] = "false" - result = self.api_patch(desdemona, f"/api/v1/streams/{stream.id}", info=data) - self.assert_json_success(result) - stream.refresh_from_db() - self.assertFalse(stream.deactivated) - - cordelia = self.example_user("cordelia") - stream1 = self.make_stream("test_stream_1", invite_only=False) - self.subscribe(iago, stream1.name) - - do_deactivate_stream(stream1, acting_user=None) - stream1.refresh_from_db() - self.assertTrue(stream1.deactivated) - - result = self.api_patch(cordelia, f"/api/v1/streams/{stream1.id}", info=data) - self.assert_json_error(result, "You do not have permission to administer this channel.") - - do_change_stream_group_based_setting( - stream1, - "can_administer_channel_group", - UserGroupMembersData(direct_members=[cordelia.id], direct_subgroups=[]), - acting_user=desdemona, - ) - - result = self.api_patch(cordelia, f"/api/v1/streams/{stream1.id}", info=data) - self.assert_json_success(result) - - stream1.refresh_from_db() - self.assertFalse(stream1.deactivated) - def test_is_archived_true_does_not_archive_stream(self) -> None: """ Ensure that passing `is_archived` as True does not archive the stream. @@ -1543,20 +1432,10 @@ class StreamAdminTest(ZulipTestCase): realm = user_profile.realm stream = self.subscribe(user_profile, "stream_name") - user_profile_group = check_add_user_group( - realm, "user_profile_group", [user_profile], acting_user=user_profile - ) - do_change_stream_group_based_setting( - stream, - "can_administer_channel_group", - user_profile_group, - acting_user=user_profile, - ) + do_change_user_role(user_profile, UserProfile.ROLE_REALM_ADMINISTRATOR, acting_user=None) result = self.client_patch(f"/json/streams/{stream.id}", {"new_name": "stream_name1"}) self.assert_json_success(result) - do_change_user_role(user_profile, UserProfile.ROLE_REALM_ADMINISTRATOR, acting_user=None) - result = self.client_patch(f"/json/streams/{stream.id}", {"new_name": "stream_name1"}) self.assert_json_error(result, "Channel already has that name.") result = self.client_patch(f"/json/streams/{stream.id}", {"new_name": "Denmark"}) @@ -1735,17 +1614,6 @@ class StreamAdminTest(ZulipTestCase): # User belonging to `can_subscribe_group` should be notified. self.assertIn(self.example_user("ZOE").id, notified_user_ids) - def test_rename_stream_requires_admin(self) -> None: - user_profile = self.example_user("hamlet") - self.login_user(user_profile) - stream = self.make_stream("stream_name1") - self.subscribe(user_profile, "stream_name1") - - stream_id = get_stream("stream_name1", user_profile.realm).id - self.assertFalse(is_user_in_group(stream.can_administer_channel_group_id, user_profile)) - result = self.client_patch(f"/json/streams/{stream_id}", {"new_name": "stream_name2"}) - self.assert_json_error(result, "You do not have permission to administer this channel.") - def test_notify_on_stream_rename(self) -> None: user_profile = self.example_user("hamlet") self.login_user(user_profile) @@ -2015,21 +1883,6 @@ class StreamAdminTest(ZulipTestCase): f'

#{core_stream.name} > testing

', ) - def test_change_stream_description_requires_administer_channel_permissions(self) -> None: - user_profile = self.example_user("hamlet") - self.login_user(user_profile) - - self.make_stream("stream_name1") - self.subscribe(user_profile, "stream_name1") - do_change_user_role(user_profile, UserProfile.ROLE_MEMBER, acting_user=None) - - stream = get_stream("stream_name1", user_profile.realm) - self.assertFalse(is_user_in_group(stream.can_administer_channel_group_id, user_profile)) - result = self.client_patch( - f"/json/streams/{stream.id}", {"description": "Test description"} - ) - self.assert_json_error(result, "You do not have permission to administer this channel.") - def test_change_stream_message_retention_days_notifications(self) -> None: user_profile = self.example_user("desdemona") self.login_user(user_profile) @@ -2213,43 +2066,10 @@ class StreamAdminTest(ZulipTestCase): nobody_group = NamedUserGroup.objects.get( name="role:nobody", is_system_group=True, realm=realm ) - - do_change_stream_group_based_setting( - stream, - "can_administer_channel_group", - nobody_group, - acting_user=user_profile, - ) - moderators_system_group = NamedUserGroup.objects.get( name="role:moderators", realm=realm, is_system_group=True ) - shiva = self.example_user("shiva") - self.login_user(shiva) - self.assertFalse(is_user_in_group(stream.can_administer_channel_group_id, shiva)) - params = {} - params[setting_name] = orjson.dumps({"new": moderators_system_group.id}).decode() - result = self.client_patch(f"/json/streams/{stream.id}", params) - self.assert_json_error(result, "You do not have permission to administer this channel.") - - do_change_stream_group_based_setting( - stream, - "can_administer_channel_group", - moderators_system_group, - acting_user=user_profile, - ) - members_system_group = NamedUserGroup.objects.get( - name="role:members", realm=realm, is_system_group=True - ) - params[setting_name] = orjson.dumps({"new": members_system_group.id}).decode() - result = self.client_patch( - f"/json/streams/{stream.id}", - params, - ) - self.assert_json_success(result) - stream = get_stream("stream_name1", realm) - self.assertEqual(getattr(stream, setting_name).id, members_system_group.id) self.login("iago") params[setting_name] = orjson.dumps({"new": moderators_system_group.id}).decode() @@ -2360,90 +2180,8 @@ class StreamAdminTest(ZulipTestCase): f"'{setting_name}' setting cannot be set to 'role:internet' group.", ) - # For private streams, realm admins need not be subscribed to - # the stream to change the setting as they can administer the - # channel by default. - stream = get_stream("stream_name2", realm) - params[setting_name] = orjson.dumps({"new": moderators_system_group.id}).decode() - result = self.client_patch( - f"/json/streams/{stream.id}", - params, - ) - if setting_name in Stream.stream_permission_group_settings_requiring_content_access: - self.assert_json_error(result, "Channel content access is required.") - else: - self.assert_json_success(result) - stream = get_stream("stream_name2", realm) - self.assertEqual(getattr(stream, setting_name).id, moderators_system_group.id) - - # For private streams, channel admins need not be subscribed to - # the stream to change the setting as they can administer the - # channel by default. - shiva_group_member_dict = UserGroupMembersData( - direct_members=[shiva.id], direct_subgroups=[] - ) - do_change_stream_group_based_setting( - stream, - "can_administer_channel_group", - shiva_group_member_dict, - acting_user=shiva, - ) - self.assertTrue(is_user_in_group(stream.can_administer_channel_group_id, shiva)) - params[setting_name] = orjson.dumps({"new": owners_group.id}).decode() - self.login_user(shiva) - result = self.client_patch( - f"/json/streams/{stream.id}", - params, - ) - if setting_name in Stream.stream_permission_group_settings_requiring_content_access: - self.assert_json_error(result, "Channel content access is required.") - do_change_stream_group_based_setting( - stream, - "can_add_subscribers_group", - shiva_group_member_dict, - acting_user=shiva, - ) - result = self.client_patch( - f"/json/streams/{stream.id}", - params, - ) - self.assert_json_success(result) - stream = get_stream("stream_name2", realm) - self.assertEqual(getattr(stream, setting_name).id, owners_group.id) - else: - self.assert_json_success(result) - stream = get_stream("stream_name2", realm) - self.assertEqual(getattr(stream, setting_name).id, owners_group.id) - - # Guest user cannot be a channel admin for a public channel. - # `user_has_permission_for_group_setting` will not allow a guest - # to be a part of `can_administer_channel_group` since that - # group has `allow_everyone_group` set to false. - stream = get_stream("stream_name1", realm) - polonius = self.example_user("polonius") - polonius_group_member_dict = UserGroupMembersData( - direct_members=[polonius.id], direct_subgroups=[] - ) - do_change_stream_group_based_setting( - stream, - "can_administer_channel_group", - polonius_group_member_dict, - acting_user=polonius, - ) - subbed_users = self.users_subscribed_to_stream(stream.name, polonius.realm) - self.assertNotIn(polonius, subbed_users) - self.login_user(polonius) - result = self.client_patch( - f"/json/streams/{stream.id}", - params, - ) - self.assert_json_error(result, "Invalid channel ID") - def test_changing_stream_permission_settings(self) -> None: self.make_stream("stream_name1") - self.make_stream("stream_name2", invite_only=True) - # Subscribe at least one user to the private stream. - self.subscribe(self.example_user("hamlet"), "stream_name2") for setting_name in Stream.stream_permission_group_settings: self.do_test_change_stream_permission_setting(setting_name) @@ -2916,56 +2654,6 @@ class StreamAdminTest(ZulipTestCase): ) self.assert_json_error(result, "Invalid channel folder ID") - def test_permission_to_change_stream_folder(self) -> None: - iago = self.example_user("iago") - hamlet = self.example_user("hamlet") - realm = iago.realm - channel_folder = check_add_channel_folder(realm, "Frontend", "", acting_user=iago) - stream = self.make_stream("test_stream") - - self.assertIsNone(stream.folder_id) - - nobody_group = NamedUserGroup.objects.get( - name=SystemGroups.NOBODY, realm=realm, is_system_group=True - ) - do_change_stream_group_based_setting( - stream, - "can_administer_channel_group", - nobody_group, - acting_user=iago, - ) - - result = self.api_patch( - iago, - f"/api/v1/streams/{stream.id}", - {"folder_id": orjson.dumps(channel_folder.id).decode()}, - ) - self.assert_json_success(result) - stream = get_stream("test_stream", realm) - self.assertEqual(stream.folder_id, channel_folder.id) - - result = self.api_patch( - hamlet, - f"/api/v1/streams/{stream.id}", - {"folder_id": orjson.dumps(None).decode()}, - ) - self.assert_json_error(result, "You do not have permission to administer this channel.") - - do_change_stream_group_based_setting( - stream, - "can_administer_channel_group", - UserGroupMembersData(direct_members=[hamlet.id], direct_subgroups=[]), - acting_user=iago, - ) - result = self.api_patch( - hamlet, - f"/api/v1/streams/{stream.id}", - {"folder_id": orjson.dumps(None).decode()}, - ) - self.assert_json_success(result) - stream = get_stream("test_stream", realm) - self.assertIsNone(stream.folder_id) - def attempt_unsubscribe_of_principal( self, target_users: list[UserProfile],