test(rules): Improve type coverage

Author: kcons

src/sentry/rules/actions/integrations/create_ticket/utils.py

@@ -2,6 +2,7 @@
 
 import logging
 from collections.abc import Callable, Sequence
+from typing import Any, cast
 
 from rest_framework.response import Response
 
@@ -109,18 +110,18 @@ def create_issue(event: GroupEvent, futures: Sequence[RuleFuture]) -> None:
     organization = event.group.project.organization
 
     for future in futures:
-        rule_id = future.rule.id
-        data = future.kwargs.get("data")
-        provider = future.kwargs.get("provider")
-        integration_id = future.kwargs.get("integration_id")
-        generate_footer = future.kwargs.get("generate_footer")
+        rule_id: int = future.rule.id
+        data = cast(dict[str, Any], future.kwargs.get("data"))
+        provider = cast(str, future.kwargs.get("provider"))
+        integration_id = cast(str, future.kwargs.get("integration_id"))
+        generate_footer = cast(Callable[[str], str], future.kwargs.get("generate_footer"))
 
         # If we invoked this handler from the notification action, we need to replace the rule_id with the legacy_rule_id, so we link notifications correctly
         action_id = None
         if features.has("organizations:workflow-engine-trigger-actions", organization):
             # In the Notification Action, we store the rule_id in the action_id field
             action_id = rule_id
-            rule_id = data.get("legacy_rule_id")
+            rule_id = cast(int, data.get("legacy_rule_id"))
 
         integration = integration_service.get_integration(
             integration_id=integration_id,

src/sentry/rules/base.py

@@ -2,9 +2,8 @@
 
 import abc
 import logging
-from collections import namedtuple
 from collections.abc import Callable, MutableMapping, Sequence
-from typing import TYPE_CHECKING, Any, ClassVar
+from typing import TYPE_CHECKING, Any, ClassVar, NamedTuple
 
 from django import forms
 
@@ -48,10 +47,14 @@
 - [ACTION:I want to group events when] [RULE:an event matches [FORM]]
 """
 
+
 # Encapsulates a reference to the callback, including arguments. The `key`
 # attribute may be specifically used to key the callbacks when they are
 # collated during rule processing.
-CallbackFuture = namedtuple("CallbackFuture", ["callback", "kwargs", "key"])
+class CallbackFuture(NamedTuple):
+    callback: Callable[[GroupEvent, Sequence[RuleFuture]], None]
+    kwargs: dict[str, Any]
+    key: str | None
 
 
 class RuleBase(abc.ABC):

src/sentry/sentry_apps/tasks/sentry_apps.py

@@ -667,7 +667,7 @@ def notify_sentry_app(event: GroupEvent, futures: Sequence[RuleFuture]):
 
         # If the future comes from a rule with a UI component form in the schema, append the issue alert payload
         # TODO(ecosystem): We need to change this payload format after alerts create issues
-        id = f.rule.id
+        id: str | int = f.rule.id
 
         # if we are using the new workflow engine, we need to use the legacy rule id
         # Ignore test notifications

src/sentry/types/rules.py

@@ -1,7 +1,12 @@
-from collections import namedtuple
 from dataclasses import dataclass
+from typing import Any, NamedTuple
 
-RuleFuture = namedtuple("RuleFuture", ["rule", "kwargs"])
+from sentry.models.rule import Rule
+
+
+class RuleFuture(NamedTuple):
+    rule: Rule
+    kwargs: dict[str, Any]
 
 
 @dataclass