user_groups: Create NamedUserGroup objects when creating new groups.

This commit is contained in:
Sahil Batra
2024-04-16 19:35:43 +05:30
committed by Tim Abbott
parent 71b601cf5a
commit 86f73fcb3d
7 changed files with 166 additions and 86 deletions

View File

@@ -2,17 +2,19 @@ from contextlib import contextmanager
from dataclasses import dataclass
from typing import Collection, Dict, Iterable, Iterator, List, Mapping, TypedDict
from django.db import transaction
from django.db import connection, transaction
from django.db.models import F, QuerySet
from django.utils.timezone import now as timezone_now
from django.utils.translation import gettext as _
from django_cte import With
from django_stubs_ext import ValuesQuerySet
from psycopg2.sql import SQL, Literal
from zerver.lib.exceptions import JsonableError
from zerver.lib.types import GroupPermissionSetting, ServerSupportedPermissionSettings
from zerver.models import (
GroupGroupMembership,
NamedUserGroup,
Realm,
RealmAuditLog,
Stream,
@@ -394,8 +396,8 @@ def get_recursive_subgroups_for_groups(
return recursive_subgroups
def get_role_based_system_groups_dict(realm: Realm) -> Dict[str, UserGroup]:
system_groups = UserGroup.objects.filter(realm=realm, is_system_group=True)
def get_role_based_system_groups_dict(realm: Realm) -> Dict[str, NamedUserGroup]:
system_groups = NamedUserGroup.objects.filter(realm=realm, is_system_group=True)
system_groups_name_dict = {}
for group in system_groups:
system_groups_name_dict[group.name] = group
@@ -404,10 +406,10 @@ def get_role_based_system_groups_dict(realm: Realm) -> Dict[str, UserGroup]:
def set_defaults_for_group_settings(
user_group: UserGroup,
user_group: NamedUserGroup,
group_settings_map: Mapping[str, UserGroup],
system_groups_name_dict: Dict[str, UserGroup],
) -> UserGroup:
system_groups_name_dict: Dict[str, NamedUserGroup],
) -> NamedUserGroup:
for setting_name, permission_config in UserGroup.GROUP_PERMISSION_SETTINGS.items():
if setting_name in group_settings_map:
# We skip the settings for which a value is passed
@@ -419,73 +421,121 @@ def set_defaults_for_group_settings(
else:
default_group_name = permission_config.default_group_name
default_group = system_groups_name_dict[default_group_name]
default_group = system_groups_name_dict[default_group_name].usergroup_ptr
setattr(user_group, setting_name, default_group)
setting_name_for_named_object = "named_group_" + setting_name
setattr(user_group, setting_name_for_named_object, default_group)
return user_group
@transaction.atomic(savepoint=False)
def create_system_user_groups_for_realm(realm: Realm) -> Dict[int, UserGroup]:
"""Any changes to this function likely require a migration to adjust
existing realms. See e.g. migration 0382_create_role_based_system_groups.py,
which is a copy of this function from when we introduced system groups.
"""
role_system_groups_dict: Dict[int, UserGroup] = {}
def bulk_create_system_user_groups(groups: List[Dict[str, str]], realm: Realm) -> None:
# This value will be used to set the temporary initial value for different
# settings since we can only set them to the correct values after the groups
# are created.
initial_group_setting_value = -1
for role in UserGroup.SYSTEM_USER_GROUP_ROLE_MAP:
user_group_params = UserGroup.SYSTEM_USER_GROUP_ROLE_MAP[role]
user_group = UserGroup(
name=user_group_params["name"],
description=user_group_params["description"],
realm=realm,
is_system_group=True,
can_mention_group_id=initial_group_setting_value,
rows = [
SQL("({},{},{},{},{})").format(
Literal(realm.id),
Literal(group["name"]),
Literal(group["description"]),
Literal(True),
Literal(initial_group_setting_value),
)
role_system_groups_dict[role] = user_group
for group in groups
]
query = SQL(
"""
INSERT INTO zerver_usergroup (realm_id, name, description, is_system_group, can_mention_group_id)
VALUES {rows}
RETURNING id
"""
).format(rows=SQL(", ").join(rows))
with connection.cursor() as cursor:
cursor.execute(query)
user_group_ids = [id for (id,) in cursor.fetchall()]
rows = [
SQL("({},{},{},{},{},{})").format(
Literal(user_group_ids[idx]),
Literal(realm.id),
Literal(group["name"]),
Literal(group["description"]),
Literal(True),
Literal(initial_group_setting_value),
)
for idx, group in enumerate(groups)
]
query = SQL(
"""
INSERT INTO zerver_namedusergroup (usergroup_ptr_id, realm_id, name, description, is_system_group, can_mention_group_id)
VALUES {rows}
"""
).format(rows=SQL(", ").join(rows))
with connection.cursor() as cursor:
cursor.execute(query)
@transaction.atomic(savepoint=False)
def create_system_user_groups_for_realm(realm: Realm) -> Dict[int, NamedUserGroup]:
"""Any changes to this function likely require a migration to adjust
existing realms. See e.g. migration 0382_create_role_based_system_groups.py,
which is a copy of this function from when we introduced system groups.
"""
role_system_groups_dict: Dict[int, NamedUserGroup] = {}
system_groups_info_list: List[Dict[str, str]] = []
nobody_group_info = {
"name": SystemGroups.NOBODY,
"description": "Nobody",
}
full_members_group_info = {
"name": SystemGroups.FULL_MEMBERS,
"description": "Members of this organization, not including new accounts and guests",
}
everyone_on_internet_group_info = {
"name": SystemGroups.EVERYONE_ON_INTERNET,
"description": "Everyone on the Internet",
}
system_groups_info_list = [
nobody_group_info,
UserGroup.SYSTEM_USER_GROUP_ROLE_MAP[UserProfile.ROLE_REALM_OWNER],
UserGroup.SYSTEM_USER_GROUP_ROLE_MAP[UserProfile.ROLE_REALM_ADMINISTRATOR],
UserGroup.SYSTEM_USER_GROUP_ROLE_MAP[UserProfile.ROLE_MODERATOR],
full_members_group_info,
UserGroup.SYSTEM_USER_GROUP_ROLE_MAP[UserProfile.ROLE_MEMBER],
UserGroup.SYSTEM_USER_GROUP_ROLE_MAP[UserProfile.ROLE_GUEST],
everyone_on_internet_group_info,
]
bulk_create_system_user_groups(system_groups_info_list, realm)
system_groups_name_dict: Dict[str, NamedUserGroup] = get_role_based_system_groups_dict(realm)
for role in UserGroup.SYSTEM_USER_GROUP_ROLE_MAP:
group_name = UserGroup.SYSTEM_USER_GROUP_ROLE_MAP[role]["name"]
role_system_groups_dict[role] = system_groups_name_dict[group_name]
full_members_system_group = UserGroup(
name=SystemGroups.FULL_MEMBERS,
description="Members of this organization, not including new accounts and guests",
realm=realm,
is_system_group=True,
can_mention_group_id=initial_group_setting_value,
)
everyone_on_internet_system_group = UserGroup(
name=SystemGroups.EVERYONE_ON_INTERNET,
description="Everyone on the Internet",
realm=realm,
is_system_group=True,
can_mention_group_id=initial_group_setting_value,
)
nobody_system_group = UserGroup(
name=SystemGroups.NOBODY,
description="Nobody",
realm=realm,
is_system_group=True,
can_mention_group_id=initial_group_setting_value,
)
# Order of this list here is important to create correct GroupGroupMembership objects
# Note that because we do not create user memberships here, no audit log entries for
# user memberships are populated either.
system_user_groups_list = [
nobody_system_group,
role_system_groups_dict[UserProfile.ROLE_REALM_OWNER],
role_system_groups_dict[UserProfile.ROLE_REALM_ADMINISTRATOR],
role_system_groups_dict[UserProfile.ROLE_MODERATOR],
full_members_system_group,
role_system_groups_dict[UserProfile.ROLE_MEMBER],
role_system_groups_dict[UserProfile.ROLE_GUEST],
everyone_on_internet_system_group,
system_groups_name_dict[SystemGroups.NOBODY],
system_groups_name_dict[SystemGroups.OWNERS],
system_groups_name_dict[SystemGroups.ADMINISTRATORS],
system_groups_name_dict[SystemGroups.MODERATORS],
system_groups_name_dict[SystemGroups.FULL_MEMBERS],
system_groups_name_dict[SystemGroups.MEMBERS],
system_groups_name_dict[SystemGroups.EVERYONE],
system_groups_name_dict[SystemGroups.EVERYONE_ON_INTERNET],
]
creation_time = timezone_now()
UserGroup.objects.bulk_create(system_user_groups_list)
realmauditlog_objects = [
RealmAuditLog(
realm=realm,
@@ -498,11 +548,12 @@ def create_system_user_groups_for_realm(realm: Realm) -> Dict[int, UserGroup]:
]
groups_with_updated_settings = []
system_groups_name_dict = get_role_based_system_groups_dict(realm)
for group in system_user_groups_list:
user_group = set_defaults_for_group_settings(group, {}, system_groups_name_dict)
groups_with_updated_settings.append(user_group)
UserGroup.objects.bulk_update(groups_with_updated_settings, ["can_mention_group"])
NamedUserGroup.objects.bulk_update(
groups_with_updated_settings, ["can_mention_group", "named_group_can_mention_group"]
)
subgroup_objects: List[GroupGroupMembership] = []
# "Nobody" system group is not a subgroup of any user group, since it is already empty.