Repository: getsentry/sentry
Pull request: feat(issue-platform): Add logic to set status to ignored for issue platform issues
Sets up the issue platform API support the IGNORED (or archived) status. We add validation to ensure the provided (status, substatus) pair is valid before updating the group status, activity, and group history.

The expectation for the UNTIL_ESCALATING and UNTIL_CONDITION_MET substatuses is that the caller will handle any conditional evaluation and will call the API with the new status when the conditions are met.

Fixes https://github.com/getsentry/sentry/issues/58036

Problem statement:
Support updating occurrence status to `Archived` via API
Payload:
```
payload_type: "status_update",
status_update: {
  fingerprint: <>
  new_status: "archived"
  new_substatus: "forever" | "until_escalating" | "until_condition_met"
```

Expectations:
- [ ] The occurrence with the given fingerprint has the status Archived.
- [ ] The activity for the occurrence reflects the status change.
- [ ] The GroupHistory for the occurrence reflects the status change.
- [ ] An Archived occurrence behaves similarly to an archived issue - no alerts, does not appear on the issue stream, is searchable with `is:archived`.

Hints from issue discussion:

Commit message: Fix test conditions

Patch to review:
diff --git a/src/sentry/event_manager.py b/src/sentry/event_manager.py
index 4eb8b2c7fc71a1..b25cc3b56014ce 100644
--- a/src/sentry/event_manager.py
+++ b/src/sentry/event_manager.py
@@ -74,7 +74,7 @@
 from sentry.ingest.inbound_filters import FilterStatKeys
 from sentry.issues.grouptype import GroupCategory
 from sentry.issues.issue_occurrence import IssueOccurrence
-from sentry.issues.producer import produce_occurrence_to_kafka
+from sentry.issues.producer import PayloadType, produce_occurrence_to_kafka
 from sentry.killswitches import killswitch_matches_context
 from sentry.lang.native.utils import STORE_CRASH_REPORTS_ALL, convert_crashreport_count
 from sentry.locks import locks
@@ -2612,7 +2612,7 @@ def _send_occurrence_to_platform(jobs: Sequence[Job], projects: ProjectsMapping)
                 level=job["level"],
             )
 
-            produce_occurrence_to_kafka(occurrence)
+            produce_occurrence_to_kafka(payload_type=PayloadType.OCCURRENCE, occurrence=occurrence)
 
 
 @metrics.wraps("event_manager.save_transaction_events")
diff --git a/src/sentry/feedback/usecases/create_feedback.py b/src/sentry/feedback/usecases/create_feedback.py
index c6bf2c33e2a98e..0197e2fd092256 100644
--- a/src/sentry/feedback/usecases/create_feedback.py
+++ b/src/sentry/feedback/usecases/create_feedback.py
@@ -3,7 +3,7 @@
 
 from sentry.issues.grouptype import FeedbackGroup
 from sentry.issues.issue_occurrence import IssueEvidence, IssueOccurrence
-from sentry.issues.producer import produce_occurrence_to_kafka
+from sentry.issues.producer import PayloadType, produce_occurrence_to_kafka
 from sentry.utils.dates import ensure_aware
 
 
@@ -78,4 +78,6 @@ def create_feedback_issue(event, project_id):
     }
     _fix_for_issue_platform(event_data)
 
-    produce_occurrence_to_kafka(occurrence, event_data=event_data)
+    produce_occurrence_to_kafka(
+        payload_type=PayloadType.OCCURRENCE, occurrence=occurrence, event_data=event_data
+    )
diff --git a/src/sentry/issues/occurrence_status_change.py b/src/sentry/issues/occurrence_status_change.py
new file mode 100644
index 00000000000000..ef6510302f3d7a
--- /dev/null
+++ b/src/sentry/issues/occurrence_status_change.py
@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import Sequence
+
+from sentry.models.group import Group, GroupStatus
+from sentry.types.activity import ActivityType
+from sentry.types.group import IGNORED_SUBSTATUS_CHOICES
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class OccurrenceStatusChange:
+    fingerprint: Sequence[str]
+    project_id: int
+    new_status: int
+    new_substatus: int | None
+
+    def update_status_for_occurrence(self, group):
+        if group.status == self.new_status and group.substatus == self.new_substatus:
+            return
+
+        log_extra = {
+            "project_id": self.project_id,
+            "fingerprint": self.fingerprint,
+            "new_status": self.new_status,
+            "new_substatus": self.new_substatus,
+        }
+
+        # Validate the provided status and substatus - we only allow setting a substatus for unresolved or ignored groups.
+        if self.new_status in [GroupStatus.UNRESOLVED, GroupStatus.IGNORED]:
+            if self.new_substatus is None:
+                logger.error(
+                    "group.update_status.missing_substatus",
+                    extra={**log_extra},
+                )
+                return
+        else:
+            if self.new_substatus is not None:
+                logger.error(
+                    "group.update_status.unexpected_substatus",
+                    extra={**log_extra},
+                )
+                return
+
+        if self.new_status == GroupStatus.RESOLVED:
+            Group.objects.update_group_status(
+                groups=[group],
+                status=self.new_status,
+                substatus=self.new_substatus,
+                activity_type=ActivityType.SET_RESOLVED,
+            )
+        elif self.new_status == GroupStatus.IGNORED:
+            # The IGNORED status supports 3 substatuses. For UNTIL_ESCALATING and
+            # UNTIL_CONDITION_MET, we expect the caller to monitor the conditions/escalating
+            # logic and call the API with the new status when the conditions change.
+            if self.new_substatus not in IGNORED_SUBSTATUS_CHOICES:
+                logger.error(
+                    "group.update_status.invalid_substatus",
+                    extra={**log_extra},
+                )
+                return
+
+            Group.objects.update_group_status(
+                groups=[group],
+                status=self.new_status,
+                substatus=self.new_substatus,
+                activity_type=ActivityType.SET_IGNORED,
+            )
+        else:
+            raise NotImplementedError(f"Unsupported status: {self.new_status} {self.new_substatus}")
diff --git a/src/sentry/issues/producer.py b/src/sentry/issues/producer.py
index 3f23ad165aad6c..3b181b2b8c1fbd 100644
--- a/src/sentry/issues/producer.py
+++ b/src/sentry/issues/producer.py
@@ -1,16 +1,29 @@
 from __future__ import annotations
 
+import logging
 from typing import Any, Dict, MutableMapping, Optional, cast
 
 from arroyo import Topic
 from arroyo.backends.kafka import KafkaPayload, KafkaProducer, build_kafka_configuration
 from django.conf import settings
 
+from sentry import features
 from sentry.issues.issue_occurrence import IssueOccurrence
+from sentry.issues.occurrence_status_change import OccurrenceStatusChange
+from sentry.models.grouphash import GroupHash
+from sentry.models.project import Project
+from sentry.services.hybrid_cloud import ValueEqualityEnum
 from sentry.utils import json
 from sentry.utils.arroyo_producer import SingletonProducer
 from sentry.utils.kafka_config import get_kafka_producer_cluster_options, get_topic_definition
 
+logger = logging.getLogger(__name__)
+
+
+class PayloadType(ValueEqualityEnum):
+    OCCURRENCE = "occurrence"
+    STATUS_CHANGE = "status_change"
+
 
 def _get_occurrence_producer() -> KafkaProducer:
     cluster_name = get_topic_definition(settings.KAFKA_INGEST_OCCURRENCES)["cluster"]
@@ -26,25 +39,60 @@ def _get_occurrence_producer() -> KafkaProducer:
 
 
 def produce_occurrence_to_kafka(
-    occurrence: IssueOccurrence, event_data: Optional[Dict[str, Any]] = None
+    payload_type: PayloadType | None = PayloadType.OCCURRENCE,
+    occurrence: IssueOccurrence | None = None,
+    status_change: OccurrenceStatusChange | None = None,
+    event_data: Optional[Dict[str, Any]] = None,
 ) -> None:
-    if event_data and occurrence.event_id != event_data["event_id"]:
-        raise ValueError("Event id on occurrence and event_data must be the same")
-    if settings.SENTRY_EVENTSTREAM != "sentry.eventstream.kafka.KafkaEventStream":
-        # If we're not running Kafka then we're just in dev. Skip producing to Kafka and just
-        # write to the issue platform directly
-        from sentry.issues.occurrence_consumer import (
-            lookup_event_and_process_issue_occurrence,
-            process_event_and_issue_occurrence,
-        )
+    if payload_type == PayloadType.OCCURRENCE:
+        if not occurrence:
+            raise ValueError("occurrence must be provided")
+        if event_data and occurrence.event_id != event_data["event_id"]:
+            raise ValueError("Event id on occurrence and event_data must be the same")
+        if settings.SENTRY_EVENTSTREAM != "sentry.eventstream.kafka.KafkaEventStream":
+            # If we're not running Kafka then we're just in dev. Skip producing to Kafka and just
+            # write to the issue platform directly
+            from sentry.issues.occurrence_consumer import (
+                lookup_event_and_process_issue_occurrence,
+                process_event_and_issue_occurrence,
+            )
 
+            if event_data:
+                process_event_and_issue_occurrence(occurrence.to_dict(), event_data)
+            else:
+                lookup_event_and_process_issue_occurrence(occurrence.to_dict())
+            return
+        payload_data = cast(MutableMapping[str, Any], occurrence.to_dict())
         if event_data:
-            process_event_and_issue_occurrence(occurrence.to_dict(), event_data)
-        else:
-            lookup_event_and_process_issue_occurrence(occurrence.to_dict())
-        return
-    payload_data = cast(MutableMapping[str, Any], occurrence.to_dict())
-    if event_data:
-        payload_data["event"] = event_data
-    payload = KafkaPayload(None, json.dumps(payload_data).encode("utf-8"), [])
-    _occurrence_producer.produce(Topic(settings.KAFKA_INGEST_OCCURRENCES), payload)
+            payload_data["event"] = event_data
+        payload = KafkaPayload(None, json.dumps(payload_data).encode("utf-8"), [])
+        _occurrence_producer.produce(Topic(settings.KAFKA_INGEST_OCCURRENCES), payload)
+    elif payload_type == PayloadType.STATUS_CHANGE:
+        if not status_change:
+            raise ValueError("status_change must be provided")
+
+        organization = Project.objects.get(id=status_change.project_id).organization
+        if not features.has("organizations:issue-platform-api-crons-sd", organization):
+            return
+
+        grouphash = (
+            GroupHash.objects.filter(
+                project=status_change.project_id,
+                hash__in=status_change.fingerprint,
+            )
+            .select_related("group")
+            .first()
+        )
+        if not grouphash:
+            logger.error(
+                "grouphash.not_found",
+                extra={
+                    "project_id": status_change.project_id,
+                    "fingerprint": status_change.fingerprint,
+                },
+            )
+            return
+
+        status_change.update_status_for_occurrence(grouphash.group)
+    else:
+        raise NotImplementedError(f"Unknown payload type {payload_type}")
diff --git a/src/sentry/monitors/logic/mark_failed.py b/src/sentry/monitors/logic/mark_failed.py
index 62b0edcad508f0..ddf8dc305fbeef 100644
--- a/src/sentry/monitors/logic/mark_failed.py
+++ b/src/sentry/monitors/logic/mark_failed.py
@@ -15,6 +15,7 @@
     MonitorCheckInMissed,
     MonitorCheckInTimeout,
 )
+from sentry.issues.producer import PayloadType
 from sentry.models.organization import Organization
 from sentry.monitors.constants import SUBTITLE_DATETIME_FORMAT, TIMEOUT
 from sentry.monitors.models import (
@@ -297,8 +298,9 @@ def create_issue_platform_occurrence(
         trace_id = None
 
     produce_occurrence_to_kafka(
-        occurrence,
-        {
+        payload_type=PayloadType.OCCURRENCE,
+        occurrence=occurrence,
+        event_data={
             "contexts": {"monitor": get_monitor_environment_context(monitor_env)},
             "environment": monitor_env.environment.name,
             "event_id": occurrence.event_id,
diff --git a/src/sentry/replays/usecases/issue.py b/src/sentry/replays/usecases/issue.py
index 386c1706961e30..9a0dc641b84321 100644
--- a/src/sentry/replays/usecases/issue.py
+++ b/src/sentry/replays/usecases/issue.py
@@ -4,7 +4,7 @@
 
 from sentry.issues.grouptype import GroupType
 from sentry.issues.issue_occurrence import IssueEvidence, IssueOccurrence
-from sentry.issues.producer import produce_occurrence_to_kafka
+from sentry.issues.producer import PayloadType, produce_occurrence_to_kafka
 
 
 def new_issue_occurrence(
@@ -53,4 +53,6 @@ def new_issue_occurrence(
     }
     event_data.update(extra_event_data)
 
-    produce_occurrence_to_kafka(occurrence, event_data=event_data)
+    produce_occurrence_to_kafka(
+        payload_type=PayloadType.OCCURRENCE, occurrence=occurrence, event_data=event_data
+    )
diff --git a/src/sentry/statistical_detectors/issue_platform_adapter.py b/src/sentry/statistical_detectors/issue_platform_adapter.py
index 957b3d766981a8..9ad4a074b9152f 100644
--- a/src/sentry/statistical_detectors/issue_platform_adapter.py
+++ b/src/sentry/statistical_detectors/issue_platform_adapter.py
@@ -5,7 +5,7 @@
 
 from sentry.issues.grouptype import PerformanceDurationRegressionGroupType
 from sentry.issues.issue_occurrence import IssueEvidence, IssueOccurrence
-from sentry.issues.producer import produce_occurrence_to_kafka
+from sentry.issues.producer import PayloadType, produce_occurrence_to_kafka
 from sentry.seer.utils import BreakpointData
 from sentry.utils import metrics
 
@@ -71,4 +71,6 @@ def send_regressions_to_plaform(regressions: List[BreakpointData], automatic_det
         metrics.incr(
             "performance.trends.sent_occurrence", tags={"automatic_detection": automatic_detection}
         )
-        produce_occurrence_to_kafka(occurrence, event_data)
+        produce_occurrence_to_kafka(
+            payload_type=PayloadType.OCCURRENCE, occurrence=occurrence, event_data=event_data
+        )
diff --git a/tests/sentry/issues/test_producer.py b/tests/sentry/issues/test_producer.py
index d961ab91048317..5a19a0da857ce6 100644
--- a/tests/sentry/issues/test_producer.py
+++ b/tests/sentry/issues/test_producer.py
@@ -1,10 +1,22 @@
 import uuid
+from datetime import datetime
+from unittest.mock import MagicMock, patch
+
+import pytest
 
 from sentry.issues.issue_occurrence import IssueOccurrence
-from sentry.issues.producer import produce_occurrence_to_kafka
+from sentry.issues.occurrence_status_change import OccurrenceStatusChange
+from sentry.issues.producer import PayloadType, produce_occurrence_to_kafka
+from sentry.models.activity import Activity
+from sentry.models.group import GroupStatus
+from sentry.models.grouphash import GroupHash
+from sentry.models.grouphistory import STRING_TO_STATUS_LOOKUP, GroupHistory, GroupHistoryStatus
 from sentry.testutils.cases import TestCase
-from sentry.testutils.helpers.datetime import before_now
+from sentry.testutils.helpers.datetime import before_now, iso_format
+from sentry.testutils.helpers.features import apply_feature_flag_on_cls
 from sentry.testutils.skips import requires_snuba
+from sentry.types.activity import ActivityType
+from sentry.types.group import GROUP_SUBSTATUS_TO_GROUP_HISTORY_STATUS, GroupSubStatus
 from sentry.utils.samples import load_data
 from tests.sentry.issues.test_utils import OccurrenceTestMixin
 
@@ -16,13 +28,18 @@ def test_event_id_mismatch(self) -> None:
         with self.assertRaisesMessage(
             ValueError, "Event id on occurrence and event_data must be the same"
         ):
-            produce_occurrence_to_kafka(self.build_occurrence(), {"event_id": uuid.uuid4().hex})
+            produce_occurrence_to_kafka(
+                payload_type=PayloadType.OCCURRENCE,
+                occurrence=self.build_occurrence(),
+                event_data={"event_id": uuid.uuid4().hex},
+            )
 
     def test_with_event(self) -> None:
         occurrence = self.build_occurrence()
         produce_occurrence_to_kafka(
-            occurrence,
-            {
+            payload_type=PayloadType.OCCURRENCE,
+            occurrence=occurrence,
+            event_data={
                 "event_id": occurrence.event_id,
                 "project_id": self.project.id,
                 "title": "some problem",
@@ -40,8 +57,219 @@ def test_with_only_occurrence(self) -> None:
         event = self.store_event(data=load_data("transaction"), project_id=self.project.id)
         occurrence = self.build_occurrence(event_id=event.event_id, project_id=self.project.id)
         produce_occurrence_to_kafka(
-            occurrence,
+            payload_type=PayloadType.OCCURRENCE,
+            occurrence=occurrence,
         )
         stored_occurrence = IssueOccurrence.fetch(occurrence.id, occurrence.project_id)
         assert stored_occurrence
         assert occurrence.event_id == stored_occurrence.event_id
+
+
+@apply_feature_flag_on_cls("organizations:issue-platform-api-crons-sd")
+class TestProduceOccurrenceForStatusChange(TestCase, OccurrenceTestMixin):
+    def setUp(self):
+        self.event = self.store_event(
+            data={
+                "event_id": "a" * 32,
+                "message": "oh no",
+                "timestamp": iso_format(datetime.now()),
+                "fingerprint": ["group-1"],
+            },
+            project_id=self.project.id,
+        )
+        self.group = self.event.group
+        assert self.group
+        self.group_hash = GroupHash.objects.filter(group=self.group, project=self.project).first()
+        self.initial_status = self.group.status
+        self.initial_substatus = self.group.substatus
+
+    def test_with_invalid_payloads(self) -> None:
+        with pytest.raises(ValueError):
+            produce_occurrence_to_kafka(
+                payload_type=PayloadType.OCCURRENCE,
+            )
+
+            produce_occurrence_to_kafka(
+                payload_type=PayloadType.STATUS_CHANGE,
+            )
+
+        with pytest.raises(NotImplementedError):
+            produce_occurrence_to_kafka(payload_type="invalid")  # type: ignore
+
+    def test_with_no_status_change(self) -> None:
+        status_change = OccurrenceStatusChange(
+            fingerprint=[self.group_hash.hash],
+            project_id=self.group.project_id,
+            new_status=self.initial_status,
+            new_substatus=self.initial_substatus,
+        )
+        produce_occurrence_to_kafka(
+            payload_type=PayloadType.STATUS_CHANGE,
+            status_change=status_change,
+        )
+        self.group.refresh_from_db()
+        assert self.group.status == self.initial_status
+        assert self.group.substatus == self.initial_substatus
+
+        assert not Activity.objects.filter(group=self.group).exists()
+        assert not GroupHistory.objects.filter(group=self.group).exists()
+
+    def test_with_status_change_resolved(self) -> None:
+        status_change = OccurrenceStatusChange(
+            fingerprint=[self.group_hash.hash],
+            project_id=self.group.project_id,
+            new_status=GroupStatus.RESOLVED,
+            new_substatus=None,
+        )
+        produce_occurrence_to_kafka(
+            payload_type=PayloadType.STATUS_CHANGE,
+            status_change=status_change,
+        )
+        self.group.refresh_from_db()
+        assert self.group.status == GroupStatus.RESOLVED
+        assert self.group.substatus is None
+
+        assert Activity.objects.filter(
+            group=self.group, type=ActivityType.SET_RESOLVED.value
+        ).exists()
+        assert GroupHistory.objects.filter(
+            group=self.group, status=GroupHistoryStatus.RESOLVED
+        ).exists()
+
+    def test_with_status_change_archived(self) -> None:
+        for substatus in [
+            GroupSubStatus.UNTIL_ESCALATING,
+            GroupSubStatus.UNTIL_CONDITION_MET,
+            GroupSubStatus.FOREVER,
+        ]:
+            status_change = OccurrenceStatusChange(
+                fingerprint=[self.group_hash.hash],
+                project_id=self.group.project_id,
+                new_status=GroupStatus.IGNORED,
+                new_substatus=substatus,
+            )
+            produce_occurrence_to_kafka(
+                payload_type=PayloadType.STATUS_CHANGE,
+                status_change=status_change,
+            )
+            self.group.refresh_from_db()
+            assert self.group.status == GroupStatus.IGNORED
+            assert self.group.substatus == substatus
+
+            assert Activity.objects.filter(
+                group=self.group, type=ActivityType.SET_IGNORED.value
+            ).exists()
+
+            gh_status = GROUP_SUBSTATUS_TO_GROUP_HISTORY_STATUS[substatus]
+            assert GroupHistory.objects.filter(
+                group=self.group,
+                status=STRING_TO_STATUS_LOOKUP[gh_status],
+            ).exists()
+
+    @patch("sentry.issues.occurrence_status_change.logger.error")
+    def test_with_invalid_status_change(self, mock_logger_error: MagicMock) -> None:
+        bad_status_change = OccurrenceStatusChange(
+            fingerprint=[self.group_hash.hash],
+            project_id=self.group.project_id,
+            new_status=GroupStatus.RESOLVED,
+            new_substatus=GroupSubStatus.FOREVER,
+        )
+        produce_occurrence_to_kafka(
+            payload_type=PayloadType.STATUS_CHANGE,
+            status_change=bad_status_change,
+        )
+        self.group.refresh_from_db()
+        mock_logger_error.assert_called_with(
+            "group.update_status.unexpected_substatus",
+            extra={
+                "project_id": self.group.project_id,
+                "fingerprint": [self.group_hash.hash],
+                "new_status": GroupStatus.RESOLVED,
+                "new_substatus": GroupSubStatus.FOREVER,
+            },
+        )
+        assert self.group.status == self.initial_status
+        assert self.group.substatus == self.initial_substatus
+
+        bad_status_change = OccurrenceStatusChange(
+            fingerprint=[self.group_hash.hash],
+            project_id=self.group.project_id,
+            new_status=GroupStatus.IGNORED,
+            new_substatus=None,
+        )
+        produce_occurrence_to_kafka(
+            payload_type=PayloadType.STATUS_CHANGE,
+            status_change=bad_status_change,
+        )
+        self.group.refresh_from_db()
+        mock_logger_error.assert_called_with(
+            "group.update_status.missing_substatus",
+            extra={
+                "project_id": self.group.project_id,
+                "fingerprint": [self.group_hash.hash],
+                "new_status": GroupStatus.IGNORED,
+                "new_substatus": None,
+            },
+        )
+        assert self.group.status == self.initial_status
+        assert self.group.substatus == self.initial_substatus
+
+        bad_status_change = OccurrenceStatusChange(
+            fingerprint=[self.group_hash.hash],
+            project_id=self.group.project_id,
+            new_status=GroupStatus.IGNORED,
+            new_substatus=GroupSubStatus.REGRESSED,
+        )
+        produce_occurrence_to_kafka(
+            payload_type=PayloadType.STATUS_CHANGE,
+            status_change=bad_status_change,
+        )
+        self.group.refresh_from_db()
+        mock_logger_error.assert_called_with(
+            "group.update_status.invalid_substatus",
+            extra={
+                "project_id": self.group.project_id,
+                "fingerprint": [self.group_hash.hash],
+                "new_status": GroupStatus.IGNORED,
+                "new_substatus": GroupSubStatus.REGRESSED,
+            },
+        )
+        assert self.group.status == self.initial_status
+        assert self.group.substatus == self.initial_substatus
+
+    @patch("sentry.issues.producer.logger.error")
+    def test_invalid_hashes(self, mock_logger_error) -> None:
+        event = self.store_event(
+            data={
+                "event_id": "a" * 32,
+                "message": "oh no",
+                "timestamp": iso_format(datetime.now()),
+                "fingerprint": ["group-2"],
+            },
+            project_id=self.project.id,
+        )
+        group = event.group
+        assert group
+        initial_status = group.status
+        initial_substatus = group.substatus
+
+        bad_status_change_resolve = OccurrenceStatusChange(
+            fingerprint=["wronghash"],
+            project_id=group.project_id,
+            new_status=GroupStatus.RESOLVED,
+            new_substatus=GroupSubStatus.FOREVER,
+        )
+        produce_occurrence_to_kafka(
+            payload_type=PayloadType.STATUS_CHANGE,
+            status_change=bad_status_change_resolve,
+        )
+        group.refresh_from_db()
+        mock_logger_error.assert_called_with(
+            "grouphash.not_found",
+            extra={
+                "project_id": group.project_id,
+                "fingerprint": ["wronghash"],
+            },
+        )
+        assert group.status == initial_status
+        assert group.substatus == initial_substatus
diff --git a/tests/sentry/monitors/logic/test_mark_failed.py b/tests/sentry/monitors/logic/test_mark_failed.py
index b33ff9f7b7a456..fbb7ddd3ef95c3 100644
--- a/tests/sentry/monitors/logic/test_mark_failed.py
+++ b/tests/sentry/monitors/logic/test_mark_failed.py
@@ -267,7 +267,9 @@ def test_mark_failed_default_params_issue_platform(self, mock_produce_occurrence
 
         assert len(mock_produce_occurrence_to_kafka.mock_calls) == 1
 
-        occurrence, event = mock_produce_occurrence_to_kafka.mock_calls[0].args
+        kwargs = mock_produce_occurrence_to_kafka.call_args.kwargs
+        occurrence = kwargs["occurrence"]
+        event = kwargs["event_data"]
         occurrence = occurrence.to_dict()
 
         assert dict(
@@ -370,7 +372,9 @@ def test_mark_failed_with_reason_issue_platform(self, mock_produce_occurrence_to
 
         assert len(mock_produce_occurrence_to_kafka.mock_calls) == 1
 
-        occurrence, event = mock_produce_occurrence_to_kafka.mock_calls[0].args
+        kwargs = mock_produce_occurrence_to_kafka.call_args.kwargs
+        occurrence = kwargs["occurrence"]
+        event = kwargs["event_data"]
         occurrence = occurrence.to_dict()
 
         assert dict(
@@ -476,7 +480,9 @@ def test_mark_failed_with_missed_reason_issue_platform(self, mock_produce_occurr
 
         assert len(mock_produce_occurrence_to_kafka.mock_calls) == 1
 
-        occurrence, event = mock_produce_occurrence_to_kafka.mock_calls[0].args
+        kwargs = mock_produce_occurrence_to_kafka.call_args.kwargs
+        occurrence = kwargs["occurrence"]
+        event = kwargs["event_data"]
         occurrence = occurrence.to_dict()
 
         assert dict(
@@ -664,7 +670,8 @@ def test_mark_failed_issue_threshold(self, mock_produce_occurrence_to_kafka):
         # assert correct number of occurrences was sent
         assert len(mock_produce_occurrence_to_kafka.mock_calls) == failure_issue_threshold
         # assert that the correct uuid fingerprint was sent
-        occurrence, event = mock_produce_occurrence_to_kafka.mock_calls[0].args
+        kwargs = mock_produce_occurrence_to_kafka.call_args.kwargs
+        occurrence = kwargs["occurrence"]
         occurrence = occurrence.to_dict()
         assert occurrence["fingerprint"][0] == monitor_incident.grouphash
 
@@ -688,9 +695,8 @@ def test_mark_failed_issue_threshold(self, mock_produce_occurrence_to_kafka):
         # assert correct number of occurrences was sent
         assert len(mock_produce_occurrence_to_kafka.mock_calls) == failure_issue_threshold + 1
         # assert that the correct uuid fingerprint was sent
-        occurrence, event = mock_produce_occurrence_to_kafka.mock_calls[
-            failure_issue_threshold
-        ].args
+        kwargs = mock_produce_occurrence_to_kafka.call_args.kwargs
+        occurrence = kwargs["occurrence"]
         occurrence = occurrence.to_dict()
         assert occurrence["fingerprint"][0] == monitor_incident.grouphash
 
diff --git a/tests/sentry/statistical_detectors/test_issue_platform_adapter.py b/tests/sentry/statistical_detectors/test_issue_platform_adapter.py
index 2c33772fb314f7..83c611f6663e81 100644
--- a/tests/sentry/statistical_detectors/test_issue_platform_adapter.py
+++ b/tests/sentry/statistical_detectors/test_issue_platform_adapter.py
@@ -29,7 +29,9 @@ def test_send_regressions_to_plaform(mock_produce_occurrence_to_kafka):
 
     assert len(mock_produce_occurrence_to_kafka.mock_calls) == 1
 
-    occurrence, event = mock_produce_occurrence_to_kafka.mock_calls[0].args
+    kwargs = mock_produce_occurrence_to_kafka.call_args.kwargs
+    occurrence = kwargs["occurrence"]
+    event = kwargs["event_data"]
     occurrence = occurrence.to_dict()
 
     assert dict(


Code review comments:
diff_hunk: @@ -75,127 +77,167 @@ def test_with_status_change(self):
             },
             project_id=self.project.id,
         )
-        group = event.group
-        assert group
-        group_hash = GroupHash.objects.filter(group=group, project=self.project).first()
-        initial_status = group.status
-        initial_substatus = group.substatus
+        self.group = self.event.group
+        assert self.group
+        self.group_hash = GroupHash.objects.filter(group=self.group, project=self.project).first()
+        self.initial_status = self.group.status
+        self.initial_substatus = self.group.substatus
 
-        status_change_resolve = OccurrenceStatusChange(
-            fingerprint=[group_hash.hash],
-            project_id=group.project_id,
-            new_status=initial_status,
-            new_substatus=initial_substatus,
+    def test_with_invalid_payloads(self) -> None:
+        with pytest.raises(ValueError):
+            produce_occurrence_to_kafka(
+                payload_type=PayloadType.OCCURRENCE,
+            )
+
+            produce_occurrence_to_kafka(
+                payload_type=PayloadType.STATUS_CHANGE,
path: tests/sentry/issues/test_producer.py
text: @user1:
Why does this raise a value error?

@author:
The status change payload type needs to be accompanied with the StatusChangeOccurrence object - added comments to the tests to explain more.