ref(notifications): remove more unused notification methods (#60726)

Now that we don't use the object manager for `NotificationSetting`, we
can remove the custom manager and all related code. This also removes
other unused methods from the helper file.

src/sentry/models/notificationsetting.py

@@ -1,5 +1,3 @@
-from typing import ClassVar
-
 import sentry_sdk
 from django.conf import settings
 from django.db import models
@@ -14,7 +12,6 @@ from sentry.db.models import (
     sane_repr,
 )
 from sentry.db.models.fields.hybrid_cloud_foreign_key import HybridCloudForeignKey
-from sentry.notifications.manager import NotificationsManager
 from sentry.notifications.types import (
     NotificationScopeType,
     NotificationSettingOptionValues,
@@ -107,8 +104,6 @@ class NotificationSetting(Model):
         null=False,
     )
 
-    objects: ClassVar[NotificationsManager] = NotificationsManager()
-
     class Meta:
         app_label = "sentry"
         db_table = "sentry_notificationsetting"

src/sentry/notifications/helpers.py

@@ -4,7 +4,6 @@ import logging
 from collections import defaultdict
 from typing import TYPE_CHECKING, Any, Iterable, Mapping, MutableMapping
 
-from django.contrib.auth.models import AnonymousUser
 from django.db.models import Subquery
 
 from sentry.hybridcloud.models.externalactorreplica import ExternalActorReplica
@@ -16,69 +15,33 @@ from sentry.notifications.defaults import (
     NOTIFICATION_SETTINGS_ALL_SOMETIMES_V2,
 )
 from sentry.notifications.types import (
-    NOTIFICATION_SCOPE_TYPE,
     NOTIFICATION_SETTING_OPTION_VALUES,
     NOTIFICATION_SETTING_TYPES,
     SUBSCRIPTION_REASON_MAP,
     VALID_VALUES_FOR_KEY,
     VALID_VALUES_FOR_KEY_V2,
     GroupSubscriptionReason,
-    NotificationScopeType,
     NotificationSettingEnum,
     NotificationSettingOptionValues,
     NotificationSettingsOptionEnum,
     NotificationSettingTypes,
 )
-from sentry.services.hybrid_cloud import extract_id_from
 from sentry.services.hybrid_cloud.actor import ActorType, RpcActor
 from sentry.services.hybrid_cloud.user.model import RpcUser
 from sentry.types.integrations import (
     EXTERNAL_PROVIDERS,
     PERSONAL_NOTIFICATION_PROVIDERS_AS_INT,
     ExternalProviderEnum,
-    ExternalProviders,
-    get_provider_enum_from_string,
-    get_provider_name,
 )
 
 if TYPE_CHECKING:
     from sentry.models.group import Group
-    from sentry.models.groupsubscription import GroupSubscription
-    from sentry.models.organization import Organization
-    from sentry.models.project import Project
     from sentry.models.team import Team
     from sentry.models.user import User
 
 logger = logging.getLogger(__name__)
 
 
-def _get_notification_setting_default(
-    provider: ExternalProviders,
-    type: NotificationSettingTypes,
-    recipient: RpcActor | None = None,  # not needed right now
-) -> NotificationSettingOptionValues:
-    """
-    In order to increase engagement, we automatically opt users into receiving
-    Slack notifications if they install Slack and link their identity.
-    Approval notifications always default to Slack being on.
-    """
-
-    # every team default is off
-    if recipient is not None and recipient.actor_type == ActorType.TEAM:
-        return NotificationSettingOptionValues.NEVER
-    return NOTIFICATION_SETTING_DEFAULTS[provider][type]
-
-
-def _get_default_value_by_provider(
-    type: NotificationSettingTypes,
-    recipient: RpcActor | None = None,
-) -> Mapping[ExternalProviders, NotificationSettingOptionValues]:
-    return {
-        provider: _get_notification_setting_default(provider, type, recipient)
-        for provider in NOTIFICATION_SETTING_DEFAULTS.keys()
-    }
-
-
 def get_provider_defaults() -> list[ExternalProviderEnum]:
     # create the data structure outside the endpoint
     provider_defaults = []
@@ -132,128 +95,6 @@ def get_type_defaults() -> Mapping[NotificationSettingEnum, NotificationSettings
     return type_defaults
 
 
-def _get_setting_mapping_from_mapping(
-    notification_settings_by_recipient: Mapping[
-        RpcActor,
-        Mapping[NotificationScopeType, Mapping[ExternalProviders, NotificationSettingOptionValues]],
-    ],
-    recipient: RpcActor,
-    type: NotificationSettingTypes,
-) -> Mapping[ExternalProviders, NotificationSettingOptionValues]:
-    """
-    XXX(CEO): may not respect granularity of a setting for Slack a setting for
-     email but we'll worry about that later since we don't have a FE for it yet.
-    """
-    return merge_notification_settings_up(
-        _get_default_value_by_provider(type, recipient),
-        *(
-            notification_settings_by_recipient.get(recipient, {}).get(scope, {})
-            for scope in (
-                NotificationScopeType.USER,
-                NotificationScopeType.TEAM,
-                get_scope_type(type),
-            )
-        ),
-    )
-
-
-def where_should_recipient_be_notified(
-    notification_settings_by_recipient: Mapping[
-        RpcActor,
-        Mapping[NotificationScopeType, Mapping[ExternalProviders, NotificationSettingOptionValues]],
-    ],
-    recipient: RpcActor,
-    type: NotificationSettingTypes = NotificationSettingTypes.ISSUE_ALERTS,
-) -> list[ExternalProviders]:
-    """
-    Given a mapping of default and specific notification settings by user,
-    return the list of providers after verifying the user has opted into this notification.
-    """
-    mapping = _get_setting_mapping_from_mapping(
-        notification_settings_by_recipient,
-        recipient,
-        type,
-    )
-    return [
-        provider
-        for provider, value in mapping.items()
-        if value == NotificationSettingOptionValues.ALWAYS
-    ]
-
-
-def should_be_participating(
-    subscription: Any | None,
-    value: NotificationSettingOptionValues,
-) -> bool:
-    """
-    Give an Actor's subscription (on, off, or null) to a group and their
-    notification setting value(on, off, or sometimes), decide whether or not to
-    send the Actor a notification.
-    """
-    return (
-        subscription and subscription.is_active and value != NotificationSettingOptionValues.NEVER
-    ) or (not subscription and value == NotificationSettingOptionValues.ALWAYS)
-
-
-def where_should_be_participating(
-    recipient: RpcActor,
-    subscription: GroupSubscription | None,
-    notification_settings_by_recipient: Mapping[
-        RpcActor,
-        Mapping[NotificationScopeType, Mapping[ExternalProviders, NotificationSettingOptionValues]],
-    ],
-) -> list[ExternalProviders]:
-    """
-    Given a mapping of users to subscriptions and a mapping of default and
-    specific notification settings by user, determine where a user should receive
-    a WORKFLOW notification. Unfortunately, this algorithm does not respect
-    NotificationSettingOptionValues.ALWAYS. If the user is unsubscribed from
-    the group, that overrides their notification preferences.
-    """
-    mapping = _get_setting_mapping_from_mapping(
-        notification_settings_by_recipient,
-        recipient,
-        NotificationSettingTypes.WORKFLOW,
-    )
-    return [
-        provider
-        for provider, value in mapping.items()
-        if should_be_participating(subscription, value)
-    ]
-
-
-def get_values_by_provider_by_type(
-    notification_settings_by_scope: Mapping[
-        NotificationScopeType, Mapping[ExternalProviders, NotificationSettingOptionValues]
-    ],
-    all_providers: Iterable[ExternalProviders],
-    type: NotificationSettingTypes,
-    recipient: RpcActor | None = None,
-) -> Mapping[ExternalProviders, NotificationSettingOptionValues]:
-    """
-    Given a mapping of scopes to a mapping of default and specific notification
-    settings by provider, determine the notification setting by provider for
-    the given notification type.
-    """
-    parent_scope = get_scope_type(type)
-
-    parent_specific_mapping = notification_settings_by_scope.get(parent_scope, {})
-    organization_independent_mapping = (
-        notification_settings_by_scope.get(NotificationScopeType.USER)
-        or notification_settings_by_scope.get(NotificationScopeType.TEAM)
-        or {}
-    )
-
-    return {
-        provider: (
-            parent_specific_mapping.get(provider)
-            or organization_independent_mapping.get(provider)
-            or _get_notification_setting_default(provider, type, recipient)
-        )
-        for provider in all_providers
-    }
-
-
 def validate(type: NotificationSettingTypes, value: NotificationSettingOptionValues) -> bool:
     """:returns boolean. True if the "value" is valid for the "type"."""
     return value in VALID_VALUES_FOR_KEY.get(type, {})
@@ -264,57 +105,6 @@ def validate_v2(type: NotificationSettingTypes, value: NotificationSettingOption
     return value in VALID_VALUES_FOR_KEY_V2.get(type, {})
 
 
-def get_scope_type(type: NotificationSettingTypes) -> NotificationScopeType:
-    """In which scope (proj or org) can a user set more specific settings?"""
-    if type in [
-        NotificationSettingTypes.DEPLOY,
-        NotificationSettingTypes.APPROVAL,
-        NotificationSettingTypes.QUOTA,
-        NotificationSettingTypes.QUOTA_ERRORS,
-        NotificationSettingTypes.QUOTA_TRANSACTIONS,
-        NotificationSettingTypes.QUOTA_ATTACHMENTS,
-        NotificationSettingTypes.QUOTA_REPLAYS,
-        NotificationSettingTypes.QUOTA_WARNINGS,
-        NotificationSettingTypes.QUOTA_SPEND_ALLOCATIONS,
-    ]:
-        return NotificationScopeType.ORGANIZATION
-
-    if type in [
-        NotificationSettingTypes.WORKFLOW,
-        NotificationSettingTypes.ISSUE_ALERTS,
-        NotificationSettingTypes.SPIKE_PROTECTION,
-    ]:
-        return NotificationScopeType.PROJECT
-
-    raise Exception(
-        f"type {type}, must be alerts, deploy, workflow, approval, quota, quotaErrors, quotaTransactions, quotaAttachments, quotaReplays, quotaWarnings, quotaSpendAllocations, spikeProtection"
-    )
-
-
-def get_scope(
-    user: User | int | None = None,
-    team: Team | int | None = None,
-    project: Project | int | None = None,
-    organization: Organization | int | None = None,
-) -> tuple[NotificationScopeType, int]:
-    """
-    Figure out the scope from parameters and return it as a tuple.
-    TODO(mgaeta): Make sure the user/team is in the project/organization.
-    """
-    if project:
-        return NotificationScopeType.PROJECT, extract_id_from(project)
-
-    if organization:
-        return NotificationScopeType.ORGANIZATION, extract_id_from(organization)
-
-    if user is not None:
-        return NotificationScopeType.USER, extract_id_from(user)
-    if team is not None:
-        return NotificationScopeType.TEAM, extract_id_from(team)
-
-    raise Exception("scope must be either user, team, organization, or project")
-
-
 def get_subscription_from_attributes(
     attrs: Mapping[str, Any]
 ) -> tuple[bool, Mapping[str, str | bool] | None]:
@@ -330,38 +120,6 @@ def get_subscription_from_attributes(
     return is_subscribed, subscription_details
 
 
-# TODO(snigdha): cleanup after v2
-def get_groups_for_query(
-    groups_by_project: Mapping[int, set[Group]],
-    notification_settings_by_scope: Mapping[
-        NotificationScopeType,
-        Mapping[int, Mapping[ExternalProviders, NotificationSettingOptionValues]],
-    ],
-    user: User,
-) -> set[Group]:
-    """
-    If there is a subscription record associated with the group, we can just use
-    that to know if a user is subscribed or not, as long as notifications aren't
-    disabled for the project.
-    """
-
-    # Avoid n queries for actors.
-    actor = RpcActor.from_object(user)
-
-    # Although this can be done with a comprehension, looping for clarity.
-    output = set()
-    for project_id, groups in groups_by_project.items():
-        value = get_most_specific_notification_setting_value(
-            notification_settings_by_scope,
-            recipient=actor,
-            parent_id=project_id,
-            type=NotificationSettingTypes.WORKFLOW,
-        )
-        if value != NotificationSettingOptionValues.NEVER:
-            output |= groups
-    return output
-
-
 def collect_groups_by_project(groups: Iterable[Group]) -> Mapping[int, set[Group]]:
     """
     Collect all of the projects to look up, and keep a set of groups that are
@@ -374,124 +132,6 @@ def collect_groups_by_project(groups: Iterable[Group]) -> Mapping[int, set[Group
     return projects
 
 
-def get_user_subscriptions_for_groups(
-    groups_by_project: Mapping[int, set[Group]],
-    notification_settings_by_scope: Mapping[
-        NotificationScopeType,
-        Mapping[int, Mapping[ExternalProviders, NotificationSettingOptionValues]],
-    ],
-    subscriptions_by_group_id: Mapping[int, GroupSubscription],
-    user: User,
-) -> Mapping[int, tuple[bool, bool, GroupSubscription | None]]:
-    """
-    For each group, use the combination of GroupSubscription and
-    NotificationSetting rows to determine if the user is explicitly or
-    implicitly subscribed (or if they can subscribe at all.)
-    """
-    results = {}
-    actor = RpcActor.from_orm_user(user)
-    for project_id, groups in groups_by_project.items():
-        notification_settings_by_provider = get_values_by_provider(
-            notification_settings_by_scope,
-            recipient=actor,
-            parent_id=project_id,
-            type=NotificationSettingTypes.WORKFLOW,
-        )
-        for group in groups:
-            results[group.id] = _get_subscription_values(
-                group,
-                subscriptions_by_group_id,
-                notification_settings_by_provider,
-            )
-    return results
-
-
-def _get_subscription_values(
-    group: Group,
-    subscriptions_by_group_id: Mapping[int, GroupSubscription],
-    notification_settings_by_provider: Mapping[ExternalProviders, NotificationSettingOptionValues],
-) -> tuple[bool, bool, GroupSubscription | None]:
-    is_disabled = False
-    subscription = subscriptions_by_group_id.get(group.id)
-    if subscription:
-        # Having a GroupSubscription overrides NotificationSettings.
-        is_active = subscription.is_active
-    else:
-        value = get_highest_notification_setting_value(notification_settings_by_provider)
-        if value == NotificationSettingOptionValues.NEVER:
-            # The user has disabled notifications in all cases.
-            is_disabled = True
-            is_active = False
-        else:
-            # Since there is no subscription, it is only active if the value is ALWAYS.
-            is_active = value == NotificationSettingOptionValues.ALWAYS
-
-    return is_disabled, is_active, subscription
-
-
-def get_settings_by_provider(
-    settings: Mapping[
-        NotificationScopeType, Mapping[ExternalProviders, NotificationSettingOptionValues]
-    ]
-) -> MutableMapping[
-    ExternalProviders, MutableMapping[NotificationScopeType, NotificationSettingOptionValues]
-]:
-    output: MutableMapping[
-        ExternalProviders, MutableMapping[NotificationScopeType, NotificationSettingOptionValues]
-    ] = defaultdict(dict)
-
-    for scope_type in settings:
-        for provider, value in settings[scope_type].items():
-            output[provider][scope_type] = value
-
-    return output
-
-
-def get_fallback_settings(
-    types_to_serialize: Iterable[NotificationSettingTypes],
-    project_ids: Iterable[int],
-    organization_ids: Iterable[int],
-    recipient: RpcActor | None = None,
-) -> MutableMapping[str, MutableMapping[str, MutableMapping[int, MutableMapping[str, str]]]]:
-    """
-    The API is responsible for calculating the implied setting values when a
-    user or team does not have explicit notification settings. This function
-    creates a "dummy" version of the nested object of notification settings that
-    can be overridden by explicit settings.
-    """
-    data: MutableMapping[
-        str, MutableMapping[str, MutableMapping[int, MutableMapping[str, str]]]
-    ] = defaultdict(lambda: defaultdict(lambda: defaultdict(dict)))
-
-    parent_independent_value_str = NOTIFICATION_SETTING_OPTION_VALUES[
-        NotificationSettingOptionValues.DEFAULT
-    ]
-
-    # Set the application-wide defaults in case they aren't set.
-    for type_enum in types_to_serialize:
-        scope_type = get_scope_type(type_enum)
-        scope_str = NOTIFICATION_SCOPE_TYPE[scope_type]
-        type_str = NOTIFICATION_SETTING_TYPES[type_enum]
-
-        for provider in NOTIFICATION_SETTING_DEFAULTS.keys():
-            provider_str = EXTERNAL_PROVIDERS[provider]
-
-            parent_ids = (
-                project_ids if scope_type == NotificationScopeType.PROJECT else organization_ids
-            )
-            for parent_id in parent_ids:
-                data[type_str][scope_str][parent_id][provider_str] = parent_independent_value_str
-
-            if recipient:
-                # Each provider has it's own defaults by type.
-                value = _get_notification_setting_default(provider, type_enum, recipient)
-                value_str = NOTIFICATION_SETTING_OPTION_VALUES[value]
-                user_scope_str = NOTIFICATION_SCOPE_TYPE[NotificationScopeType.USER]
-
-                data[type_str][user_scope_str][recipient.id][provider_str] = value_str
-    return data
-
-
 def get_reason_context(extra_context: Mapping[str, Any]) -> MutableMapping[str, str]:
     """Get user-specific context. Do not call get_context() here."""
     reason = extra_context.get("reason", 0)
@@ -500,134 +140,6 @@ def get_reason_context(extra_context: Mapping[str, Any]) -> MutableMapping[str,
     }
 
 
-# TODO(snigdha): cleanup after v2
-def get_highest_notification_setting_value(
-    notification_settings_by_provider: Mapping[ExternalProviders, NotificationSettingOptionValues],
-) -> NotificationSettingOptionValues | None:
-    """
-    Find the "most specific" notification setting value. Currently non-NEVER
-    values are locked together (for example, you cannot have
-    `{"email": "always", "slack": "subscribe_only"}` but you can have
-    `{"email": "always", "slack": "never"}` and
-    `{"email": "always", "slack": "always"}`), but this might change. This is a
-    HACK but if we put an explicit ordering here It'd match the implicit ordering.
-    """
-    if not notification_settings_by_provider:
-        return None
-    return max(notification_settings_by_provider.values(), key=lambda v: v.value)
-
-
-def get_value_for_parent(
-    notification_settings_by_scope: Mapping[
-        NotificationScopeType,
-        Mapping[int, Mapping[ExternalProviders, NotificationSettingOptionValues]],
-    ],
-    parent_id: int,
-    type: NotificationSettingTypes,
-) -> Mapping[ExternalProviders, NotificationSettingOptionValues]:
-    """
-    Given notification settings by scope, an organization or project, and a
-    notification type, get the notification settings by provider.
-    """
-    return notification_settings_by_scope.get(get_scope_type(type), {}).get(parent_id, {})
-
-
-def _get_value_for_actor(
-    notification_settings_by_scope: Mapping[
-        NotificationScopeType,
-        Mapping[int, Mapping[ExternalProviders, NotificationSettingOptionValues]],
-    ],
-    recipient: RpcActor,
-) -> Mapping[ExternalProviders, NotificationSettingOptionValues]:
-    """
-    Instead of checking the DB to see if `recipient` is a Team or User, just
-    `get()` both since only one of them can have a value.
-    """
-    return (
-        notification_settings_by_scope.get(NotificationScopeType.USER)
-        or notification_settings_by_scope.get(NotificationScopeType.TEAM)
-        or {}
-    ).get(recipient.id, {})
-
-
-# TODO(snigdha): cleanup after v2
-def get_most_specific_notification_setting_value(
-    notification_settings_by_scope: Mapping[
-        NotificationScopeType,
-        Mapping[int, Mapping[ExternalProviders, NotificationSettingOptionValues]],
-    ],
-    recipient: RpcActor | AnonymousUser,
-    parent_id: int,
-    type: NotificationSettingTypes,
-) -> NotificationSettingOptionValues:
-    """
-    Get the "most specific" notification setting value for a given user and
-    project. If there are no settings, default to the default setting for EMAIL.
-    """
-    if isinstance(recipient, AnonymousUser):
-        return _get_notification_setting_default(ExternalProviders.EMAIL, type, None)
-
-    recipient_actor = RpcActor.from_object(recipient)
-    return (
-        get_highest_notification_setting_value(
-            get_value_for_parent(notification_settings_by_scope, parent_id, type)
-        )
-        or get_highest_notification_setting_value(
-            _get_value_for_actor(notification_settings_by_scope, recipient_actor)
-        )
-        or _get_notification_setting_default(ExternalProviders.EMAIL, type, recipient_actor)
-    )
-
-
-def merge_notification_settings_up(
-    *settings_mappings: Mapping[ExternalProviders, NotificationSettingOptionValues],
-) -> Mapping[ExternalProviders, NotificationSettingOptionValues]:
-    """
-    Given a list of notification settings by provider ordered by increasing
-    specificity, get the most specific value by provider.
-    """
-    value_by_provider: MutableMapping[ExternalProviders, NotificationSettingOptionValues] = {}
-    for notification_settings_by_provider in settings_mappings:
-        value_by_provider.update(notification_settings_by_provider)
-    return value_by_provider
-
-
-def get_values_by_provider(
-    notification_settings_by_scope: Mapping[
-        NotificationScopeType,
-        Mapping[int, Mapping[ExternalProviders, NotificationSettingOptionValues]],
-    ],
-    recipient: RpcActor,
-    parent_id: int,
-    type: NotificationSettingTypes,
-) -> Mapping[ExternalProviders, NotificationSettingOptionValues]:
-    """
-    Given notification settings by scope, an organization or project, a
-    recipient, and a notification type, what is the non-never notification
-    setting by provider?
-    """
-    return merge_notification_settings_up(
-        _get_default_value_by_provider(type, recipient),
-        _get_value_for_actor(notification_settings_by_scope, recipient),
-        get_value_for_parent(notification_settings_by_scope, parent_id, type),
-    )
-
-
-def get_providers_for_recipient(
-    recipient: User,
-) -> Iterable[ExternalProviders]:
-    from sentry.models.identity import Identity
-
-    possible_providers = NOTIFICATION_SETTING_DEFAULTS.keys()
-    provider_names = [get_provider_name(provider.value) for provider in possible_providers]
-    idp_types = Identity.objects.filter(
-        user__id=recipient.id, idp__type__in=provider_names
-    ).values_list("idp__type", flat=True)
-    user_providers = [get_provider_enum_from_string(idp_type) for idp_type in idp_types]
-    user_providers.append(ExternalProviders.EMAIL)  # always add in email as an option
-    return user_providers
-
-
 def recipient_is_user(recipient: RpcActor | Team | RpcUser | User) -> bool:
     from sentry.models.user import User
 
@@ -644,16 +156,6 @@ def recipient_is_team(recipient: RpcActor | Team | RpcUser | User) -> bool:
     return isinstance(recipient, Team)
 
 
-def get_recipient_from_team_or_user(user_id: int | None, team_id: int | None) -> RpcUser | Team:
-    if user_id is not None:
-        recipient = RpcUser(id=user_id)
-    elif team_id is not None:
-        recipient = Team.objects.get(id=team_id)
-    if not recipient:
-        raise Exception("Unable to find user or team")
-    return recipient
-
-
 def team_is_valid_recipient(team: Team | RpcActor) -> bool:
     """
     A team is a valid recipient if it has a linked integration (ie. linked Slack channel)

src/sentry/notifications/manager.py

@@ -1,486 +0,0 @@
-from __future__ import annotations
-
-import logging
-from collections import defaultdict
-from typing import TYPE_CHECKING, Iterable, Mapping, MutableSet, Optional, Set, Union
-
-from django.db import router, transaction
-from django.db.models import Q, QuerySet
-
-from sentry import analytics
-from sentry.db.models.manager import BaseManager
-from sentry.models.notificationsettingoption import NotificationSettingOption
-from sentry.models.notificationsettingprovider import NotificationSettingProvider
-from sentry.models.project import Project
-from sentry.models.team import Team
-from sentry.models.user import User
-from sentry.notifications.helpers import get_scope, get_scope_type, validate
-from sentry.notifications.notificationcontroller import NotificationController
-from sentry.notifications.types import (
-    NOTIFICATION_SCOPE_TYPE,
-    NOTIFICATION_SETTING_OPTION_VALUES,
-    NOTIFICATION_SETTING_TYPES,
-    VALID_VALUES_FOR_KEY_V2,
-    NotificationScopeEnum,
-    NotificationScopeType,
-    NotificationSettingEnum,
-    NotificationSettingOptionValues,
-    NotificationSettingsOptionEnum,
-    NotificationSettingTypes,
-)
-from sentry.services.hybrid_cloud.actor import ActorType, RpcActor
-from sentry.services.hybrid_cloud.user.model import RpcUser
-from sentry.types.integrations import (
-    EXTERNAL_PROVIDERS,
-    PERSONAL_NOTIFICATION_PROVIDERS,
-    ExternalProviders,
-)
-from sentry.utils.sdk import configure_scope
-
-if TYPE_CHECKING:
-    from sentry.models.notificationsetting import NotificationSetting  # noqa: F401
-    from sentry.models.organization import Organization
-
-REMOVE_SETTING_BATCH_SIZE = 1000
-logger = logging.getLogger(__name__)
-
-
-class NotificationsManager(BaseManager["NotificationSetting"]):
-    """
-    TODO(mgaeta): Add a caching layer for notification settings
-    """
-
-    def get_settings(
-        self,
-        provider: ExternalProviders,
-        type: NotificationSettingTypes,
-        user_id: int | None = None,
-        team_id: int | None = None,
-        project: Project | None = None,
-        organization: Organization | None = None,
-    ) -> NotificationSettingOptionValues:
-        """
-        One and only one of (user, team, project, or organization)
-        must not be null. This function automatically translates a missing DB
-        row to NotificationSettingOptionValues.DEFAULT.
-        """
-        # The `unique_together` constraint should guarantee 0 or 1 rows, but
-        # using `list()` rather than `.first()` to prevent Django from adding an
-        # ordering that could make the query slow.
-
-        settings = list(
-            self.find_settings(
-                provider,
-                type,
-                team_id=team_id,
-                user_id=user_id,
-                project=project,
-                organization=organization,
-            )
-        )[:1]
-        return (
-            NotificationSettingOptionValues(settings[0].value)
-            if settings
-            else NotificationSettingOptionValues.DEFAULT
-        )
-
-    def _update_settings(
-        self,
-        provider: ExternalProviders,
-        type: NotificationSettingTypes,
-        value: NotificationSettingOptionValues,
-        scope_type: NotificationScopeType,
-        scope_identifier: int,
-        user_id: Optional[int] = None,
-        team_id: Optional[int] = None,
-    ) -> None:
-        """Save a NotificationSettings row."""
-        from sentry.models.notificationsetting import NotificationSetting  # noqa: F811
-
-        defaults = {"value": value.value}
-        with configure_scope() as scope:
-            with transaction.atomic(router.db_for_write(NotificationSetting)):
-                setting, created = self.get_or_create(
-                    provider=provider.value,
-                    type=type.value,
-                    scope_type=scope_type.value,
-                    scope_identifier=scope_identifier,
-                    user_id=user_id,
-                    team_id=team_id,
-                    defaults=defaults,
-                )
-                if not created and setting.value != value.value:
-                    scope.set_tag("notif_setting_type", setting.type_str)
-                    scope.set_tag("notif_setting_value", setting.value_str)
-                    scope.set_tag("notif_setting_provider", setting.provider_str)
-                    scope.set_tag("notif_setting_scope", setting.scope_str)
-                    setting.update(value=value.value)
-
-    def update_settings(
-        self,
-        provider: ExternalProviders,
-        type: NotificationSettingTypes,
-        value: NotificationSettingOptionValues,
-        user: User | None = None,
-        user_id: int | None = None,
-        team_id: int | None = None,
-        project: Project | int | None = None,
-        organization: Organization | int | None = None,
-        actor: RpcActor | None = None,
-        skip_provider_updates: bool = False,
-    ) -> None:
-        """
-        Save a target's notification preferences.
-        Examples:
-          * Updating a user's org-independent preferences
-          * Updating a user's per-project preferences
-          * Updating a user's per-organization preferences
-        """
-        if user:
-            user_id = user.id
-        elif actor:
-            if actor.actor_type == ActorType.USER:
-                user_id = actor.id
-            else:
-                team_id = actor.id
-
-        if user_id is not None:
-            actor_type = ActorType.USER
-            actor_id = user_id
-
-        if team_id is not None:
-            actor_type = ActorType.TEAM
-            actor_id = team_id
-        assert actor_type, "None actor cannot have settings updated"
-
-        analytics.record(
-            "notifications.settings_updated",
-            target_type="user" if actor_type == ActorType.USER else "team",
-            actor_id=None,
-            id=actor_id,
-        )
-
-        scope_type, scope_identifier = get_scope(
-            team=team_id, user=user_id, project=project, organization=organization
-        )
-        # A missing DB row is equivalent to DEFAULT.
-        if value == NotificationSettingOptionValues.DEFAULT:
-            self.remove_settings(
-                provider,
-                type,
-                user_id=user_id,
-                team_id=team_id,
-                project=project,
-                organization=organization,
-            )
-        else:
-            if not validate(type, value):
-                raise Exception(f"value '{value}' is not valid for type '{type}'")
-
-            id_key = "user_id" if actor_type == ActorType.USER else "team_id"
-            self._update_settings(
-                provider=provider,
-                type=type,
-                value=value,
-                scope_type=scope_type,
-                scope_identifier=scope_identifier,
-                **{id_key: actor_id},
-            )
-
-        # implement the double write now
-        query_args = {
-            "type": NOTIFICATION_SETTING_TYPES[type],
-            "scope_type": NOTIFICATION_SCOPE_TYPE[scope_type],
-            "scope_identifier": scope_identifier,
-            "user_id": user_id,
-            "team_id": team_id,
-        }
-        # if default, delete the row
-        if value == NotificationSettingOptionValues.DEFAULT:
-            NotificationSettingOption.objects.filter(**query_args).delete()
-
-        else:
-            NotificationSettingOption.objects.create_or_update(
-                **query_args,
-                values={"value": NOTIFICATION_SETTING_OPTION_VALUES[value]},
-            )
-        if not skip_provider_updates:
-            self.update_provider_settings(user_id, team_id)
-
-    def remove_settings(
-        self,
-        provider: ExternalProviders,
-        type: NotificationSettingTypes,
-        user: User | None = None,
-        user_id: int | None = None,
-        team_id: int | None = None,
-        project: Project | int | None = None,
-        organization: Organization | int | None = None,
-        organization_id_for_team: Optional[int] = None,
-    ) -> None:
-        """
-        We don't anticipate this function will be used by the API but is useful
-        for tests. This can also be called by `update_settings` when attempting
-        to set a notification preference to DEFAULT.
-        """
-        if user:
-            user_id = user.id
-
-        # get the actor type and actor id
-        scope_type, scope_identifier = get_scope(
-            team=team_id, user=user_id, project=project, organization=organization
-        )
-        scope_type_str = NOTIFICATION_SCOPE_TYPE[scope_type]
-        # remove the option setting
-        NotificationSettingOption.objects.filter(
-            scope_type=scope_type_str,
-            scope_identifier=scope_identifier,
-            team_id=team_id,
-            user_id=user_id,
-            type=type,
-        ).delete()
-        # the provider setting is updated elsewhere
-
-        self.find_settings(
-            provider,
-            type,
-            team_id=team_id,
-            user_id=user_id,
-            project=project,
-            organization=organization,
-        ).delete()
-
-    def _filter(
-        self,
-        provider: ExternalProviders | None = None,
-        type: NotificationSettingTypes | None = None,
-        scope_type: NotificationScopeType | None = None,
-        scope_identifier: int | None = None,
-        user_ids: Iterable[int] | None = None,
-        team_ids: Iterable[int] | None = None,
-    ) -> QuerySet:
-        """Wrapper for .filter that translates types to actual attributes to column types."""
-        query = Q()
-        if provider:
-            query = query & Q(provider=provider.value)
-
-        if type:
-            query = query & Q(type=type.value)
-
-        if scope_type:
-            query = query & Q(scope_type=scope_type.value)
-
-        if scope_identifier:
-            query = query & Q(scope_identifier=scope_identifier)
-
-        if team_ids or user_ids:
-            query = query & (
-                Q(team_id__in=team_ids if team_ids else [])
-                | Q(user_id__in=user_ids if user_ids else [])
-            )
-
-        return self.filter(query)
-
-    # only used in tests
-    def remove_for_user(self, user: User, type: NotificationSettingTypes | None = None) -> None:
-        """Bulk delete all Notification Settings for a USER, optionally by type."""
-        self._filter(user_ids=[user.id], type=type).delete()
-
-    def remove_for_project(
-        self, project_id: int, type: NotificationSettingTypes | None = None
-    ) -> None:
-        """Bulk delete all Notification Settings for a PROJECT, optionally by type."""
-        self._filter(
-            scope_type=NotificationScopeType.PROJECT,
-            scope_identifier=project_id,
-            type=type,
-        ).delete()
-
-    def remove_for_organization(
-        self, organization_id: int, type: NotificationSettingTypes | None = None
-    ) -> None:
-        """Bulk delete all Notification Settings for an ENTIRE ORGANIZATION, optionally by type."""
-        self._filter(
-            scope_type=NotificationScopeType.ORGANIZATION,
-            scope_identifier=organization_id,
-            type=type,
-        ).delete()
-
-    def find_settings(
-        self,
-        provider: ExternalProviders,
-        type: NotificationSettingTypes,
-        user_id: int | None = None,
-        team_id: int | None = None,
-        project: Project | int | None = None,
-        organization: Organization | int | None = None,
-    ) -> QuerySet:
-        """Wrapper for .filter that translates object parameters to scopes and targets."""
-
-        team_ids = set()
-        user_ids = set()
-
-        if team_id:
-            team_ids.add(team_id)
-        if user_id:
-            user_ids.add(user_id)
-
-        assert (team_ids and not user_ids) or (
-            user_ids and not team_ids
-        ), "Can only get settings for team or user"
-
-        scope_type, scope_identifier = get_scope(
-            team=team_id, user=user_id, project=project, organization=organization
-        )
-        assert (len(team_ids) == 1 and len(user_ids) == 0) or (
-            len(team_ids) == 0 and len(user_ids) == 1
-        ), "Cannot find settings for None actor_id"
-        return self._filter(
-            provider, type, scope_type, scope_identifier, team_ids=team_ids, user_ids=user_ids
-        )
-
-    def get_for_recipient_by_parent(
-        self,
-        type_: NotificationSettingTypes,
-        parent: Organization | Project,
-        recipients: Iterable[RpcActor | Team | User | RpcUser],
-    ) -> QuerySet:
-        """
-        Find all of a project/organization's notification settings for a list of
-        users or teams. Note that this WILL work with a mixed list. This will
-        include each user or team's project/organization-independent settings.
-        """
-        user_ids: MutableSet[int] = set()
-        team_ids: MutableSet[int] = set()
-
-        for raw_recipient in recipients:
-            recipient = RpcActor.from_object(raw_recipient, fetch_actor=False)
-            if recipient.actor_type == ActorType.TEAM:
-                team_ids.add(recipient.id)
-            if recipient.actor_type == ActorType.USER:
-                user_ids.add(recipient.id)
-
-        # If the list would be empty, don't bother querying.
-        if not (team_ids or user_ids):
-            return self.none()
-
-        parent_specific_scope_type = get_scope_type(type_)
-        return self.filter(
-            Q(
-                scope_type=parent_specific_scope_type.value,
-                scope_identifier=parent.id,
-            )
-            | Q(
-                scope_type=NotificationScopeType.USER.value,
-                scope_identifier__in=user_ids,
-            )
-            | Q(
-                scope_type=NotificationScopeType.TEAM.value,
-                scope_identifier__in=team_ids,
-            ),
-            (Q(team_id__in=team_ids) | Q(user_id__in=user_ids)),
-            type=type_.value,
-        )
-
-    def filter_to_accepting_recipients(
-        self,
-        parent: Union[Organization, Project],
-        recipients: Iterable[RpcActor | Team | RpcUser | User],
-        type: NotificationSettingTypes = NotificationSettingTypes.ISSUE_ALERTS,
-    ) -> Mapping[ExternalProviders, Iterable[RpcActor]]:
-        """
-        Filters a list of teams or users down to the recipients by provider who
-        are subscribed to alerts. We check both the project level settings and
-        global default settings.
-        """
-        recipient_actors = RpcActor.many_from_object(recipients)
-
-        if isinstance(parent, Project):
-            organization = parent.organization
-            project_ids = [parent.id]
-        else:
-            organization = parent
-            project_ids = None
-
-        setting_type = (
-            NotificationSettingEnum(NOTIFICATION_SETTING_TYPES[type])
-            if type
-            else NotificationSettingEnum.ISSUE_ALERTS
-        )
-        controller = NotificationController(
-            recipients=recipient_actors,
-            project_ids=project_ids,
-            organization_id=organization.id,
-            type=setting_type,
-        )
-
-        return controller.get_notification_recipients(type=setting_type)
-
-    def update_provider_settings(self, user_id: int | None, team_id: int | None):
-        """
-        Find all settings for a team/user, determine the provider settings and map it to the parent scope
-        """
-        assert team_id or user_id, "Cannot update settings if user or team is not passed"
-
-        parent_scope_type = (
-            NotificationScopeEnum.TEAM if team_id is not None else NotificationScopeEnum.USER
-        )
-        parent_scope_identifier = team_id if team_id is not None else user_id
-
-        # Get all the NotificationSetting rows for the user or team at the top level scope
-        notification_settings = self.filter(
-            user_id=user_id,
-            team_id=team_id,
-            scope_type__in=[
-                NotificationScopeType.USER.value,
-                NotificationScopeType.TEAM.value,
-            ],
-        )
-        # Initialize a dictionary to store the new NotificationSettingProvider values
-        enabled_providers_by_type: Mapping[str, Set[str]] = defaultdict(set)
-        disabled_providers_by_type: Mapping[str, Set[str]] = defaultdict(set)
-
-        # Iterate through all the stored NotificationSetting rows
-        for setting in notification_settings:
-            provider = EXTERNAL_PROVIDERS[setting.provider]
-            type = setting.type
-            if setting.value == NotificationSettingOptionValues.NEVER.value:
-                disabled_providers_by_type[type].add(provider)
-            else:
-                # if the value is not never, it's explicitly enabled
-                enabled_providers_by_type[type].add(provider)
-
-        # This is a bad N+1 query but we don't have a better way to do this
-        for provider in PERSONAL_NOTIFICATION_PROVIDERS:
-            types_to_delete = []
-            # iterate throuch each type of notification setting
-            base_query = {
-                "provider": provider,
-                "user_id": user_id,
-                "team_id": team_id,
-                "scope_type": parent_scope_type.value,
-                "scope_identifier": parent_scope_identifier,
-            }
-            for type in VALID_VALUES_FOR_KEY_V2.keys():
-                query_args = {
-                    **base_query,
-                    "type": NOTIFICATION_SETTING_TYPES[type],
-                }
-                if provider in enabled_providers_by_type[type]:
-                    NotificationSettingProvider.objects.create_or_update(
-                        **query_args,
-                        values={"value": NotificationSettingsOptionEnum.ALWAYS.value},
-                    )
-                elif provider in disabled_providers_by_type[type]:
-                    NotificationSettingProvider.objects.create_or_update(
-                        **query_args,
-                        values={"value": NotificationSettingsOptionEnum.NEVER.value},
-                    )
-                    # if not explicitly enabled or disable, we should delete the row
-                else:
-                    types_to_delete.append(type)
-            # delete the rows that are not explicitly enabled or disabled
-            NotificationSettingProvider.objects.filter(
-                **base_query,
-                type__in=[NOTIFICATION_SETTING_TYPES[type] for type in types_to_delete],
-            ).delete()

tests/sentry/notifications/test_helpers.py

@@ -6,20 +6,12 @@ from sentry.models.notificationsettingoption import NotificationSettingOption
 from sentry.models.rule import Rule
 from sentry.notifications.helpers import (
     collect_groups_by_project,
-    get_scope_type,
-    get_settings_by_provider,
     get_subscription_from_attributes,
     get_team_members,
-    get_values_by_provider_by_type,
     team_is_valid_recipient,
     validate,
 )
-from sentry.notifications.notify import notification_providers
-from sentry.notifications.types import (
-    NotificationScopeType,
-    NotificationSettingOptionValues,
-    NotificationSettingTypes,
-)
+from sentry.notifications.types import NotificationSettingOptionValues, NotificationSettingTypes
 from sentry.notifications.utils import (
     get_email_link_extra_params,
     get_group_settings_link,
@@ -29,7 +21,6 @@ from sentry.services.hybrid_cloud.actor import RpcActor
 from sentry.silo.base import SiloMode
 from sentry.testutils.cases import TestCase
 from sentry.testutils.silo import assume_test_silo_mode
-from sentry.types.integrations import ExternalProviders
 
 
 def mock_event(*, transaction, data=None):
@@ -55,38 +46,6 @@ class NotificationHelpersTest(TestCase):
                 value="always",
             )
 
-    def test_get_deploy_values_by_provider_empty_settings(self):
-        values_by_provider = get_values_by_provider_by_type(
-            {},
-            notification_providers(),
-            NotificationSettingTypes.DEPLOY,
-        )
-        assert values_by_provider == {
-            ExternalProviders.EMAIL: NotificationSettingOptionValues.COMMITTED_ONLY,
-            ExternalProviders.SLACK: NotificationSettingOptionValues.COMMITTED_ONLY,
-            ExternalProviders.MSTEAMS: NotificationSettingOptionValues.NEVER,
-        }
-
-    def test_get_deploy_values_by_provider(self):
-        notification_settings_by_scope = {
-            NotificationScopeType.ORGANIZATION: {
-                ExternalProviders.SLACK: NotificationSettingOptionValues.COMMITTED_ONLY
-            },
-            NotificationScopeType.USER: {
-                ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS
-            },
-        }
-        values_by_provider = get_values_by_provider_by_type(
-            notification_settings_by_scope,
-            notification_providers(),
-            NotificationSettingTypes.DEPLOY,
-        )
-        assert values_by_provider == {
-            ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS,
-            ExternalProviders.SLACK: NotificationSettingOptionValues.COMMITTED_ONLY,
-            ExternalProviders.MSTEAMS: NotificationSettingOptionValues.NEVER,
-        }
-
     def test_validate(self):
         self.assertTrue(
             validate(NotificationSettingTypes.ISSUE_ALERTS, NotificationSettingOptionValues.ALWAYS)
@@ -129,22 +88,6 @@ class NotificationHelpersTest(TestCase):
             )
         )
 
-    def test_get_scope_type(self):
-        assert get_scope_type(NotificationSettingTypes.DEPLOY) == NotificationScopeType.ORGANIZATION
-        assert get_scope_type(NotificationSettingTypes.WORKFLOW) == NotificationScopeType.PROJECT
-        assert (
-            get_scope_type(NotificationSettingTypes.ISSUE_ALERTS) == NotificationScopeType.PROJECT
-        )
-        assert not get_scope_type(NotificationSettingTypes.DEPLOY) == NotificationScopeType.PROJECT
-        assert (
-            not get_scope_type(NotificationSettingTypes.WORKFLOW)
-            == NotificationScopeType.ORGANIZATION
-        )
-        assert (
-            not get_scope_type(NotificationSettingTypes.ISSUE_ALERTS)
-            == NotificationScopeType.ORGANIZATION
-        )
-
     def test_get_subscription_from_attributes(self):
         attrs = {"subscription": (True, True, None)}
         assert get_subscription_from_attributes(attrs) == (True, {"disabled": True})
@@ -155,18 +98,6 @@ class NotificationHelpersTest(TestCase):
     def test_collect_groups_by_project(self):
         assert collect_groups_by_project([self.group]) == {self.project.id: {self.group}}
 
-    def test_get_settings_by_provider(self):
-        settings = {
-            NotificationScopeType.USER: {
-                ExternalProviders.EMAIL: NotificationSettingOptionValues.NEVER
-            }
-        }
-        assert get_settings_by_provider(settings) == {
-            ExternalProviders.EMAIL: {
-                NotificationScopeType.USER: NotificationSettingOptionValues.NEVER
-            }
-        }
-
     def test_get_group_settings_link(self):
         rule: Rule = self.create_project_rule(self.project)
         rule_details = get_rules([rule], self.organization, self.project)

tests/sentry/notifications/utils/test_get_fallback_settings.py

@@ -1,45 +0,0 @@
-from sentry.notifications.helpers import get_fallback_settings
-from sentry.notifications.types import NotificationSettingTypes
-from sentry.services.hybrid_cloud.actor import RpcActor
-from sentry.silo import SiloMode
-from sentry.testutils.cases import TestCase
-from sentry.testutils.silo import assume_test_silo_mode, control_silo_test
-
-
-@control_silo_test
-class GetFallbackSettingsTest(TestCase):
-    def setUp(self) -> None:
-        with assume_test_silo_mode(SiloMode.REGION):
-            self.user = RpcActor.from_orm_user(self.create_user())
-        self.project = self.create_project()
-
-    def test_get_fallback_settings_minimal(self):
-        assert get_fallback_settings({NotificationSettingTypes.ISSUE_ALERTS}, {}, {}) == {}
-
-    def test_get_fallback_settings_user(self):
-        data = get_fallback_settings({NotificationSettingTypes.ISSUE_ALERTS}, {}, {}, self.user)
-        assert data == {
-            "alerts": {
-                "user": {
-                    self.user.id: {
-                        "email": "always",
-                        "slack": "always",
-                        "msteams": "never",
-                    }
-                }
-            }
-        }
-
-    def test_get_fallback_settings_projects(self):
-        data = get_fallback_settings({NotificationSettingTypes.ISSUE_ALERTS}, {self.project.id}, {})
-        assert data == {
-            "alerts": {
-                "project": {
-                    self.project.id: {
-                        "email": "default",
-                        "slack": "default",
-                        "msteams": "default",
-                    }
-                }
-            }
-        }

tests/sentry/notifications/utils/test_get_groups_for_query.py

@@ -1,103 +0,0 @@
-from sentry.models.group import Group
-from sentry.models.organization import Organization
-from sentry.models.project import Project
-from sentry.notifications.helpers import get_groups_for_query
-from sentry.notifications.types import NotificationScopeType, NotificationSettingOptionValues
-from sentry.testutils.cases import TestCase
-from sentry.testutils.silo import no_silo_test
-from sentry.types.integrations import ExternalProviders
-
-
-@no_silo_test
-class GetGroupsForQueryTestCase(TestCase):
-    def setUp(self) -> None:
-        super().setUp()
-
-    def test_get_groups_for_query_empty(self):
-        groups_by_project = {self.project: {self.group}}
-        notification_settings_by_scope = {
-            NotificationScopeType.PROJECT: {
-                self.project.id: {
-                    ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS,
-                },
-            },
-        }
-
-        assert (
-            get_groups_for_query(
-                groups_by_project={}, notification_settings_by_scope={}, user=self.user
-            )
-            == set()
-        )
-        assert get_groups_for_query(
-            groups_by_project, notification_settings_by_scope={}, user=self.user
-        ) == {self.group}
-        assert (
-            get_groups_for_query(
-                groups_by_project={},
-                notification_settings_by_scope=notification_settings_by_scope,
-                user=self.user,
-            )
-            == set()
-        )
-
-    def test_get_groups_for_query(self):
-        organization = Organization(id=1, slug="organization", name="My Company")
-        project_0 = Project(id=100, organization=organization)
-        project_1 = Project(id=101, organization=organization)
-        project_2 = Project(id=102, organization=organization)
-
-        groups_by_project = {
-            project_0.id: {Group(id=10, project=project_0), Group(id=11, project=project_0)},
-            project_1.id: {Group(id=12, project=project_0)},
-            project_2.id: {Group(id=13, project=project_0)},
-        }
-
-        notification_settings_by_scope = {
-            NotificationScopeType.PROJECT: {
-                project_0.id: {
-                    ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS,
-                },
-                project_1.id: {
-                    ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.NEVER,
-                },
-            }
-        }
-        query_groups = get_groups_for_query(
-            groups_by_project, notification_settings_by_scope, user=self.user
-        )
-        assert {group.id for group in query_groups} == {10, 11, 13}
-
-    def test_get_groups_for_query_simple(self):
-        assert get_groups_for_query(
-            {self.project: {self.group}},
-            {
-                NotificationScopeType.PROJECT: {
-                    self.project.id: {
-                        ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                        ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS,
-                    },
-                },
-            },
-            user=self.user,
-        ) == {self.group}
-
-    def test_get_groups_for_query_never(self):
-        assert (
-            get_groups_for_query(
-                {self.project.id: {self.group}},
-                {
-                    NotificationScopeType.PROJECT: {
-                        self.project.id: {
-                            ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                            ExternalProviders.EMAIL: NotificationSettingOptionValues.NEVER,
-                        },
-                    },
-                },
-                user=self.user,
-            )
-            == set()
-        )

tests/sentry/notifications/utils/test_get_most_specific.py

@@ -1,106 +0,0 @@
-from sentry.models.user import User
-from sentry.notifications.helpers import (
-    get_highest_notification_setting_value,
-    get_most_specific_notification_setting_value,
-)
-from sentry.notifications.types import (
-    NotificationScopeType,
-    NotificationSettingOptionValues,
-    NotificationSettingTypes,
-)
-from sentry.services.hybrid_cloud.actor import ActorType, RpcActor
-from sentry.testutils.cases import TestCase
-from sentry.testutils.silo import control_silo_test
-from sentry.types.integrations import ExternalProviders
-
-
-@control_silo_test
-class GetMostSpecificNotificationSettingValueTestCase(TestCase):
-    def setUp(self) -> None:
-        self.user = self.create_user()
-
-    def test_get_most_specific_notification_setting_value_empty_workflow(self):
-        value = get_most_specific_notification_setting_value(
-            notification_settings_by_scope={},
-            recipient=RpcActor(id=self.user.id, actor_type=ActorType.USER),
-            parent_id=1,
-            type=NotificationSettingTypes.WORKFLOW,
-        )
-        assert value == NotificationSettingOptionValues.SUBSCRIBE_ONLY
-
-    def test_get_most_specific_notification_setting_value_empty_alerts(self):
-        value = get_most_specific_notification_setting_value(
-            notification_settings_by_scope={},
-            recipient=RpcActor(id=self.user.id, actor_type=ActorType.USER),
-            parent_id=1,
-            type=NotificationSettingTypes.ISSUE_ALERTS,
-        )
-        assert value == NotificationSettingOptionValues.ALWAYS
-
-    def test_get_most_specific_notification_setting_value_user(self):
-        notification_settings_by_scope = {
-            NotificationScopeType.USER: {
-                self.user.id: {
-                    ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS,
-                },
-            },
-        }
-        value = get_most_specific_notification_setting_value(
-            notification_settings_by_scope,
-            recipient=RpcActor(id=self.user.id, actor_type=ActorType.USER),
-            parent_id=1,
-            type=NotificationSettingTypes.ISSUE_ALERTS,
-        )
-        assert value == NotificationSettingOptionValues.ALWAYS
-
-    def test_get_most_specific_notification_setting_value(self):
-        project_id = 1
-
-        notification_settings_by_scope = {
-            NotificationScopeType.USER: {
-                self.user.id: {
-                    ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS,
-                },
-            },
-            NotificationScopeType.PROJECT: {
-                project_id: {
-                    ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.NEVER,
-                },
-            },
-        }
-        value = get_most_specific_notification_setting_value(
-            notification_settings_by_scope,
-            recipient=RpcActor(id=self.user.id, actor_type=ActorType.USER),
-            parent_id=project_id,
-            type=NotificationSettingTypes.ISSUE_ALERTS,
-        )
-        assert value == NotificationSettingOptionValues.NEVER
-
-
-class GetHighestNotificationSettingValueTestCase(TestCase):
-    def setUp(self) -> None:
-        self.user = User(id=1)
-
-    def test_get_highest_notification_setting_value_empty(self):
-        assert get_highest_notification_setting_value({}) is None
-
-    def test_get_highest_notification_setting_value(self):
-        value = get_highest_notification_setting_value(
-            {
-                ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS,
-            }
-        )
-        assert value == NotificationSettingOptionValues.ALWAYS
-
-    def test_get_highest_notification_setting_value_never(self):
-        value = get_highest_notification_setting_value(
-            {
-                ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                ExternalProviders.EMAIL: NotificationSettingOptionValues.NEVER,
-            }
-        )
-        assert value == NotificationSettingOptionValues.NEVER

tests/sentry/notifications/utils/test_get_scope.py

@@ -1,44 +0,0 @@
-from sentry.models.organization import Organization
-from sentry.models.project import Project
-from sentry.models.team import Team
-from sentry.notifications.helpers import get_scope
-from sentry.notifications.types import NotificationScopeType
-from sentry.testutils.cases import TestCase
-from sentry.testutils.silo import control_silo_test
-
-
-@control_silo_test
-class GetScopeTestCase(TestCase):
-    def setUp(self) -> None:
-        self.user = self.create_user()
-
-    def test_get_scope_user(self):
-        scope_type, scope_identifier = get_scope(user=self.user)
-        assert scope_type == NotificationScopeType.USER
-        assert scope_identifier == self.user.id
-
-    def test_get_scope_team(self):
-        team = Team(id=1)
-        scope_type, scope_identifier = get_scope(team=team)
-        assert scope_type == NotificationScopeType.TEAM
-        assert scope_identifier == team.id
-
-    def test_get_scope_project(self):
-        project = Project(id=1)
-        scope_type, scope_identifier = get_scope(user=self.user, project=project)
-        assert scope_type == NotificationScopeType.PROJECT
-        assert scope_identifier == project.id
-
-        scope_type, scope_identifier = get_scope(user=self.user, project=project.id)
-        assert scope_type == NotificationScopeType.PROJECT
-        assert scope_identifier == project.id
-
-    def test_get_scope_organization(self):
-        organization = Organization(id=1)
-        scope_type, scope_identifier = get_scope(user=self.user, organization=organization)
-        assert scope_type == NotificationScopeType.ORGANIZATION
-        assert scope_identifier == organization.id
-
-        scope_type, scope_identifier = get_scope(user=self.user, organization=organization.id)
-        assert scope_type == NotificationScopeType.ORGANIZATION
-        assert scope_identifier == organization.id

tests/sentry/notifications/utils/test_get_setting_mapping_from_mapping.py

@@ -1,177 +0,0 @@
-from sentry.notifications.helpers import _get_setting_mapping_from_mapping
-from sentry.notifications.types import (
-    NotificationScopeType,
-    NotificationSettingOptionValues,
-    NotificationSettingTypes,
-)
-from sentry.services.hybrid_cloud.actor import RpcActor
-from sentry.silo import SiloMode
-from sentry.testutils.cases import TestCase
-from sentry.testutils.silo import assume_test_silo_mode, control_silo_test
-from sentry.types.integrations import ExternalProviders
-
-
-@control_silo_test
-class GetSettingMappingFromMappingTest(TestCase):
-    def setUp(self):
-        with assume_test_silo_mode(SiloMode.REGION):
-            self.user = RpcActor.from_orm_user(self.create_user())
-
-    def test_get_setting_mapping_from_mapping_issue_alerts(self):
-        notification_settings = {
-            self.user: {
-                NotificationScopeType.USER: {
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS
-                }
-            }
-        }
-        mapping = _get_setting_mapping_from_mapping(
-            notification_settings,
-            self.user,
-            NotificationSettingTypes.ISSUE_ALERTS,
-        )
-        assert mapping == {
-            ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS,
-            ExternalProviders.SLACK: NotificationSettingOptionValues.ALWAYS,
-            ExternalProviders.MSTEAMS: NotificationSettingOptionValues.NEVER,
-        }
-
-    def test_get_setting_mapping_from_mapping_deploy(self):
-        notification_settings = {
-            self.user: {
-                NotificationScopeType.USER: {
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.COMMITTED_ONLY
-                }
-            }
-        }
-        mapping = _get_setting_mapping_from_mapping(
-            notification_settings,
-            self.user,
-            NotificationSettingTypes.DEPLOY,
-        )
-        assert mapping == {
-            ExternalProviders.EMAIL: NotificationSettingOptionValues.COMMITTED_ONLY,
-            ExternalProviders.SLACK: NotificationSettingOptionValues.COMMITTED_ONLY,
-            ExternalProviders.MSTEAMS: NotificationSettingOptionValues.NEVER,
-        }
-
-    def test_get_setting_mapping_from_mapping_workflow(self):
-        notification_settings = {
-            self.user: {
-                NotificationScopeType.USER: {
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.SUBSCRIBE_ONLY
-                }
-            }
-        }
-        mapping = _get_setting_mapping_from_mapping(
-            notification_settings,
-            self.user,
-            NotificationSettingTypes.WORKFLOW,
-        )
-        assert mapping == {
-            ExternalProviders.EMAIL: NotificationSettingOptionValues.SUBSCRIBE_ONLY,
-            ExternalProviders.SLACK: NotificationSettingOptionValues.SUBSCRIBE_ONLY,
-            ExternalProviders.MSTEAMS: NotificationSettingOptionValues.NEVER,
-        }
-
-    def test_get_setting_mapping_from_mapping_empty(self):
-        mapping = _get_setting_mapping_from_mapping(
-            {}, self.user, NotificationSettingTypes.ISSUE_ALERTS
-        )
-        assert mapping == {
-            ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS,
-            ExternalProviders.SLACK: NotificationSettingOptionValues.ALWAYS,
-            ExternalProviders.MSTEAMS: NotificationSettingOptionValues.NEVER,
-        }
-
-    def test_get_setting_mapping_from_mapping_slack_never(self):
-        notification_settings = {
-            self.user: {
-                NotificationScopeType.USER: {
-                    ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER
-                }
-            }
-        }
-
-        mapping = _get_setting_mapping_from_mapping(
-            notification_settings,
-            self.user,
-            NotificationSettingTypes.ISSUE_ALERTS,
-        )
-        assert mapping == {
-            ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS,
-            ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-            ExternalProviders.MSTEAMS: NotificationSettingOptionValues.NEVER,
-        }
-
-    def test_get_setting_mapping_from_mapping_slack_always(self):
-        notification_settings = {
-            self.user: {
-                NotificationScopeType.USER: {
-                    ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER
-                }
-            }
-        }
-
-        mapping = _get_setting_mapping_from_mapping(
-            notification_settings,
-            self.user,
-            NotificationSettingTypes.ISSUE_ALERTS,
-        )
-        assert mapping[ExternalProviders.SLACK] == NotificationSettingOptionValues.NEVER
-
-    def test_get_setting_mapping_msteams_never(self):
-        notification_settings = {
-            self.user: {
-                NotificationScopeType.USER: {
-                    ExternalProviders.MSTEAMS: NotificationSettingOptionValues.NEVER
-                }
-            }
-        }
-
-        mapping = _get_setting_mapping_from_mapping(
-            notification_settings,
-            self.user,
-            NotificationSettingTypes.ISSUE_ALERTS,
-        )
-        assert mapping[ExternalProviders.MSTEAMS] == NotificationSettingOptionValues.NEVER
-
-    def test_get_setting_mapping_msteams_always(self):
-        notification_settings = {
-            self.user: {
-                NotificationScopeType.USER: {
-                    ExternalProviders.MSTEAMS: NotificationSettingOptionValues.ALWAYS
-                }
-            }
-        }
-
-        mapping = _get_setting_mapping_from_mapping(
-            notification_settings,
-            self.user,
-            NotificationSettingTypes.ISSUE_ALERTS,
-        )
-        assert mapping[ExternalProviders.MSTEAMS] == NotificationSettingOptionValues.ALWAYS
-
-    def test_get_setting_mapping_from_mapping_project(self):
-        notification_settings = {
-            self.user: {
-                NotificationScopeType.USER: {
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.NEVER,
-                    ExternalProviders.SLACK: NotificationSettingOptionValues.ALWAYS,
-                },
-                NotificationScopeType.PROJECT: {
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.NEVER,
-                },
-            }
-        }
-
-        mapping = _get_setting_mapping_from_mapping(
-            notification_settings,
-            self.user,
-            NotificationSettingTypes.ISSUE_ALERTS,
-        )
-        assert mapping == {
-            ExternalProviders.EMAIL: NotificationSettingOptionValues.NEVER,
-            ExternalProviders.SLACK: NotificationSettingOptionValues.ALWAYS,
-            ExternalProviders.MSTEAMS: NotificationSettingOptionValues.NEVER,
-        }

tests/sentry/notifications/utils/test_get_user_subscriptions_for_groups.py

@@ -1,93 +0,0 @@
-from sentry.models.group import Group
-from sentry.models.groupsubscription import GroupSubscription
-from sentry.models.project import Project
-from sentry.notifications.helpers import get_user_subscriptions_for_groups
-from sentry.notifications.types import NotificationScopeType, NotificationSettingOptionValues
-from sentry.testutils.cases import TestCase
-from sentry.testutils.silo import no_silo_test
-from sentry.types.integrations import ExternalProviders
-
-
-@no_silo_test
-class GetUserSubscriptionsForGroupsTestCase(TestCase):
-    def setUp(self) -> None:
-        self.group_subscription = GroupSubscription(is_active=True)
-        self.user = self.create_user()
-        self.project = Project(id=1)
-        self.group = Group(id=1)
-
-    def test_get_user_subscriptions_for_groups_empty(self):
-        groups_by_project = {self.project.id: {self.group}}
-        notification_settings_by_scope = {
-            NotificationScopeType.USER: {
-                self.user.id: {
-                    ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS,
-                },
-            },
-            NotificationScopeType.PROJECT: {
-                self.project.id: {
-                    ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.NEVER,
-                },
-            },
-        }
-
-        subscriptions_by_group_id = {self.group.id: self.group_subscription}
-        assert (
-            get_user_subscriptions_for_groups(
-                groups_by_project={},
-                notification_settings_by_scope={},
-                subscriptions_by_group_id={},
-                user=self.user,
-            )
-            == {}
-        )
-
-        assert (
-            get_user_subscriptions_for_groups(
-                groups_by_project={},
-                notification_settings_by_scope=notification_settings_by_scope,
-                subscriptions_by_group_id=subscriptions_by_group_id,
-                user=self.user,
-            )
-            == {}
-        )
-
-        assert get_user_subscriptions_for_groups(
-            groups_by_project=groups_by_project,
-            notification_settings_by_scope={},
-            subscriptions_by_group_id=subscriptions_by_group_id,
-            user=self.user,
-        ) == {self.group.id: (False, True, self.group_subscription)}
-
-        assert get_user_subscriptions_for_groups(
-            groups_by_project=groups_by_project,
-            notification_settings_by_scope=notification_settings_by_scope,
-            subscriptions_by_group_id={},
-            user=self.user,
-        ) == {self.group.id: (True, False, None)}
-
-    def test_get_user_subscriptions_for_groups(self):
-        groups_by_project = {self.project.id: {self.group}}
-        notification_settings_by_scope = {
-            NotificationScopeType.USER: {
-                self.user.id: {
-                    ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS,
-                },
-            },
-            NotificationScopeType.PROJECT: {
-                self.project.id: {
-                    ExternalProviders.SLACK: NotificationSettingOptionValues.NEVER,
-                    ExternalProviders.EMAIL: NotificationSettingOptionValues.NEVER,
-                },
-            },
-        }
-        subscriptions_by_group_id = {self.group.id: self.group_subscription}
-        assert get_user_subscriptions_for_groups(
-            groups_by_project,
-            notification_settings_by_scope,
-            subscriptions_by_group_id,
-            user=self.user,
-        ) == {self.group.id: (False, True, self.group_subscription)}