build out types for status and checks, for passing information back to other notifiers

Author: nora-shap

services/notification/__init__.py

@@ -37,10 +37,7 @@
     ChecksWithFallback,
 )
 from services.notification.notifiers.codecov_slack_app import CodecovSlackAppNotifier
-from services.notification.notifiers.mixins.status import (
-    StatusState,
-    custom_target_helper_text,
-)
+from services.notification.notifiers.mixins.status import StatusState
 from services.yaml import read_yaml_field
 from services.yaml.reader import get_components_from_yaml
 
@@ -261,7 +258,7 @@ def notify(self, comparison: ComparisonProxy) -> list[IndividualResult]:
         )
 
         results = []
-        add_custom_target_helper_text_to_pr_comment = False
+        status_or_checks_helper_text = {}
         notifiers_to_notify = [
             notifier
             for notifier in self.get_notifiers_instances()
@@ -280,34 +277,33 @@ def notify(self, comparison: ComparisonProxy) -> list[IndividualResult]:
             if notifier.notification_type not in notification_type_status_or_checks
         ]
 
-        if status_or_checks_notifiers and all_other_notifiers:
+        status_or_checks_results = [
+            self.notify_individual_notifier(notifier, comparison)
+            for notifier in status_or_checks_notifiers
+        ]
+
+        if status_or_checks_results and all_other_notifiers:
             # if the status/check fails, sometimes we want to add helper text to the message of the other notifications,
             # to better surface that the status/check failed.
             # so if there are status_and_checks_notifiers and all_other_notifiers, do the status_and_checks_notifiers first,
             # look at the results of the checks, if any failed AND they are the type we have helper text for,
             # add that text onto the other notifiers messages.
-            status_or_checks_results = [
-                self.notify_individual_notifier(notifier, comparison)
-                for notifier in status_or_checks_notifiers
-            ]
-
             for result in status_or_checks_results:
                 notification_result = result["result"]
                 if (
                     notification_result is not None
                     and notification_result.data_sent.get("state")
                     == StatusState.failure.value
-                ):
-                    if custom_target_helper_text in notification_result.data_sent.get(
-                        "message"
-                    ):
-                        add_custom_target_helper_text_to_pr_comment = True
-            results = results + status_or_checks_results
+                ) and notification_result.data_sent.get("included_helper_text"):
+                    status_or_checks_helper_text.update(
+                        notification_result.data_sent["included_helper_text"]
+                    )
+                    # TODO: pass status_or_checks_helper_text to all_other_notifiers,
+                    #  where they can integrate the helper text into their messages
+        results = results + status_or_checks_results
 
         results = results + [
-            self.notify_individual_notifier(
-                notifier, comparison, add_custom_target_helper_text_to_pr_comment
-            )
+            self.notify_individual_notifier(notifier, comparison)
             for notifier in all_other_notifiers
         ]
         return results
@@ -316,7 +312,6 @@ def notify_individual_notifier(
         self,
         notifier: AbstractBaseNotifier,
         comparison: ComparisonProxy,
-        add_custom_target_helper_text_to_pr_comment: bool = False,
     ) -> IndividualResult:
         commit = comparison.head.commit
         base_commit = comparison.project_coverage_base.commit
@@ -337,9 +332,7 @@ def notify_individual_notifier(
             notifier=notifier.name, title=notifier.title, result=None
         )
         try:
-            res = notifier.notify(
-                comparison, add_custom_target_helper_text_to_pr_comment
-            )
+            res = notifier.notify(comparison)
             individual_result["result"] = res
 
             notifier.store_results(comparison, res)

services/notification/notifiers/checks/patch.py

@@ -1,18 +1,34 @@
+from typing import Any, TypedDict
+
 from database.enums import Notification
 from services.comparison import ComparisonProxy, FilteredComparison
 from services.notification.notifiers.checks.base import ChecksNotifier
-from services.notification.notifiers.mixins.status import StatusPatchMixin
+from services.notification.notifiers.mixins.status import StatusPatchMixin, StatusState
 from services.yaml import read_yaml_field
 
 
+class CheckOutput(TypedDict):
+    title: str
+    summary: str
+    annotations: list[Any]
+
+
+class CheckResult(TypedDict):
+    state: StatusState
+    output: CheckOutput
+    included_helper_text: dict[str, str]
+
+
 class PatchChecksNotifier(StatusPatchMixin, ChecksNotifier):
     context = "patch"
 
     @property
     def notification_type(self) -> Notification:
         return Notification.checks_patch
 
-    def build_payload(self, comparison: ComparisonProxy | FilteredComparison) -> dict:
+    def build_payload(
+        self, comparison: ComparisonProxy | FilteredComparison
+    ) -> CheckResult:
         """
         This method build the paylod of the patch github checks.
 
@@ -21,17 +37,21 @@ def build_payload(self, comparison: ComparisonProxy | FilteredComparison) -> dic
         """
         if self.is_empty_upload():
             state, message = self.get_status_check_for_empty_upload()
-            return {
-                "state": state,
-                "output": {
-                    "title": "Empty Upload",
-                    "summary": message,
-                },
-            }
-        state, message = self.get_patch_status(comparison)
+            result = CheckResult(
+                state=state,
+                output=CheckOutput(
+                    title="Empty Upload",
+                    summary=message,
+                    annotations=[],
+                ),
+                included_helper_text={},
+            )
+            return result
+        status_result = self.get_patch_status(comparison)
         codecov_link = self.get_codecov_pr_link(comparison)
 
-        title = message
+        title = status_result["message"]
+        message = status_result["message"]
 
         should_use_upgrade = self.should_use_upgrade_decoration()
         if should_use_upgrade:
@@ -54,23 +74,27 @@ def build_payload(self, comparison: ComparisonProxy | FilteredComparison) -> dic
             or should_use_upgrade
             or should_annotate is False
         ):
-            return {
-                "state": state,
-                "output": {
-                    "title": f"{title}",
-                    "summary": "\n\n".join([codecov_link, message]),
-                },
-            }
+            result = CheckResult(
+                state=status_result["state"],
+                output=CheckOutput(
+                    title=title,
+                    summary="\n\n".join([codecov_link, message]),
+                    annotations=[],
+                ),
+                included_helper_text=status_result["included_helper_text"],
+            )
+            return result
         diff = comparison.get_diff(use_original_base=True)
         #  TODO: Look into why the apply diff in get_patch_status is not saving state at this point
         comparison.head.report.apply_diff(diff)
         annotations = self.create_annotations(comparison, diff)
-
-        return {
-            "state": state,
-            "output": {
-                "title": f"{title}",
-                "summary": "\n\n".join([codecov_link, message]),
-                "annotations": annotations,
-            },
-        }
+        result = CheckResult(
+            state=status_result["state"],
+            output=CheckOutput(
+                title=title,
+                summary="\n\n".join([codecov_link, message]),
+                annotations=annotations,
+            ),
+            included_helper_text=status_result["included_helper_text"],
+        )
+        return result

services/notification/notifiers/mixins/status.py

@@ -1,6 +1,7 @@
 import logging
 from decimal import Decimal, InvalidOperation
 from enum import Enum
+from typing import Literal, TypedDict
 
 from services.comparison import ComparisonProxy, FilteredComparison
 from services.yaml.reader import round_number
@@ -13,10 +14,35 @@ class StatusState(Enum):
     failure = "failure"
 
 
-custom_target_helper_text = "Your project check has failed because the patch coverage is below the target coverage. You can increase the patch coverage or adjust the [target](https://docs.codecov.com/docs/commit-status#target) coverage."
+class StatusResult(TypedDict):
+    """
+    The mixins in this file do the calculations and decide the SuccessState for all Status and Checks Notifiers.
+    Checks have different fields than Statuses, so Checks are converted to the CheckResult type later.
+    """
+
+    state: Literal["success", "failure"]  # StatusState values
+    message: str
+    included_helper_text: dict[str, str]
+
+
+CUSTOM_TARGET_TEXT_PATCH_KEY = "custom_target_helper_text_patch"
+CUSTOM_TARGET_TEXT_PROJECT_KEY = "custom_target_helper_text_project"
+CUSTOM_TARGET_TEXT_VALUE = (
+    "Your {context} check has failed because the patch coverage is below the target coverage. "
+    "You can increase the patch coverage or adjust the "
+    "[target](https://docs.codecov.com/docs/commit-status#target) coverage."
+)
+
+
+HELPER_TEXT_MAP = {
+    CUSTOM_TARGET_TEXT_PATCH_KEY: CUSTOM_TARGET_TEXT_VALUE,
+    CUSTOM_TARGET_TEXT_PROJECT_KEY: CUSTOM_TARGET_TEXT_VALUE,
+}
 
 
 class StatusPatchMixin(object):
+    context = "patch"
+
     def _get_threshold(self) -> Decimal:
         """
         Threshold can be configured by user, default is 0.0
@@ -55,10 +81,11 @@ def _get_target(
 
     def get_patch_status(
         self, comparison: ComparisonProxy | FilteredComparison
-    ) -> tuple[str, str]:
+    ) -> StatusResult:
         threshold = self._get_threshold()
         target_coverage, is_custom_target = self._get_target(comparison)
         totals = comparison.get_patch_totals()
+        included_helper_text = {}
 
         # coverage affected
         if totals and totals.lines > 0:
@@ -89,9 +116,16 @@ def get_patch_status(
                     message = (
                         f"{coverage_rounded}% of diff hit (target {target_rounded}%)"
                     )
-                if state == StatusState.failure.value and is_custom_target:
-                    message = message + " - " + custom_target_helper_text
-            return state, message
+                # TODO:
+                # if state == StatusState.failure.value and is_custom_target:
+                #     helper_text = HELPER_TEXT_MAP[CUSTOM_TARGET_TEXT_PATCH_KEY].format(
+                #         context=self.context
+                #     )
+                #     included_helper_text[CUSTOM_TARGET_TEXT_PATCH_KEY] = helper_text
+                #     message = message + " - " + helper_text
+            return StatusResult(
+                state=state, message=message, included_helper_text=included_helper_text
+            )
 
         # coverage not affected
         if comparison.project_coverage_base.commit:
@@ -101,10 +135,16 @@ def get_patch_status(
             )
         else:
             description = "Coverage not affected"
-        return StatusState.success.value, description
+        return StatusResult(
+            state=StatusState.success.value,
+            message=description,
+            included_helper_text=included_helper_text,
+        )
 
 
 class StatusChangesMixin(object):
+    context = "changes"
+
     def is_a_change_worth_noting(self, change) -> bool:
         if not change.new and not change.deleted:
             # has totals and not -10m => 10h
@@ -151,6 +191,7 @@ def get_changes_status(
 
 class StatusProjectMixin(object):
     DEFAULT_REMOVED_CODE_BEHAVIOR = "adjust_base"
+    context = "project"
 
     def _apply_removals_only_behavior(
         self, comparison: ComparisonProxy | FilteredComparison
@@ -427,8 +468,10 @@ def _get_project_status(
             # use rounded numbers for messages
             target_rounded = round_number(self.current_yaml, target_coverage)
             message = f"{head_coverage_rounded}% (target {target_rounded}%)"
-            if state == StatusState.failure.value:
-                message = message + " - " + custom_target_helper_text
+            # TODO:
+            # helper_text = HELPER_TEXT_MAP[CUSTOM_TARGET_TEXT].format(context=self.context)
+            # included_helper_text[CUSTOM_TARGET_TEXT] = helper_text
+            # message = message + " - " + helper_text
             return state, message
 
         # use rounded numbers for messages

services/notification/notifiers/status/patch.py

@@ -1,6 +1,6 @@
 from database.enums import Notification
 from services.comparison import ComparisonProxy, FilteredComparison
-from services.notification.notifiers.mixins.status import StatusPatchMixin
+from services.notification.notifiers.mixins.status import StatusPatchMixin, StatusResult
 from services.notification.notifiers.status.base import StatusNotifier
 
 
@@ -22,13 +22,16 @@ class PatchStatusNotifier(StatusPatchMixin, StatusNotifier):
     def notification_type(self) -> Notification:
         return Notification.status_patch
 
-    def build_payload(self, comparison: ComparisonProxy | FilteredComparison) -> dict:
+    def build_payload(
+        self, comparison: ComparisonProxy | FilteredComparison
+    ) -> StatusResult:
         if self.is_empty_upload():
             state, message = self.get_status_check_for_empty_upload()
-            return {"state": state, "message": message}
+            result = StatusResult(state=state, message=message, included_helper_text={})
+            return result
 
-        state, message = self.get_patch_status(comparison)
+        result = self.get_patch_status(comparison)
         if self.should_use_upgrade_decoration():
-            message = self.get_upgrade_message()
+            result["message"] = self.get_upgrade_message()
 
-        return {"state": state, "message": message}
+        return result