Fix an audit trail race condition when deleting segment overrides

Author: emyller

api/features/models.py

@@ -1,5 +1,6 @@
 from __future__ import unicode_literals
 
+from contextlib import suppress
 import datetime
 import logging
 import typing
@@ -962,25 +963,30 @@ def get_create_log_message(self, history_instance) -> typing.Optional[str]:  # t
 
         return audit_helpers.get_environment_feature_state_created_audit_message(self)
 
-    def get_update_log_message(self, history_instance) -> typing.Optional[str]:  # type: ignore[no-untyped-def]
-        if self.change_request and self.is_scheduled:
-            live_from: datetime.datetime = timezone.localtime(self.live_from)
-            return FEATURE_STATE_SCHEDULED_TO_UPDATE_MESSAGE % (
-                self.feature.name,
-                self.change_request.title,
-                formats.date_format(live_from, settings.DATETIME_FORMAT),
-            )
-        if self.identity:
-            return IDENTITY_FEATURE_STATE_UPDATED_MESSAGE % (
-                self.feature.name,
-                self.identity.identifier,
-            )
-        elif self.feature_segment:
-            return SEGMENT_FEATURE_STATE_UPDATED_MESSAGE % (
-                self.feature.name,
-                self.feature_segment.segment.name,
-            )
-        return FEATURE_STATE_UPDATED_MESSAGE % self.feature.name
+    def get_update_log_message(self, history_instance: "FeatureState") -> str | None:
+        if self.change_request_id and self.is_scheduled:
+            live_from = timezone.localtime(self.live_from)
+            with suppress(ObjectDoesNotExist):
+                return FEATURE_STATE_SCHEDULED_TO_UPDATE_MESSAGE % (
+                    self.feature.name,
+                    self.change_request.title,
+                    formats.date_format(live_from, settings.DATETIME_FORMAT),
+                )
+        elif self.identity_id:
+            with suppress(ObjectDoesNotExist):
+                return IDENTITY_FEATURE_STATE_UPDATED_MESSAGE % (
+                    self.feature.name,
+                    self.identity.identifier,
+                )
+        elif self.feature_segment_id:
+            with suppress(ObjectDoesNotExist):
+                return SEGMENT_FEATURE_STATE_UPDATED_MESSAGE % (
+                    self.feature.name,
+                    self.feature_segment.segment.name,
+                )
+        else:
+            return FEATURE_STATE_UPDATED_MESSAGE % self.feature.name
+        return None
 
     def get_delete_log_message(self, history_instance) -> typing.Optional[str]:  # type: ignore[no-untyped-def]
         try:

api/tests/unit/audit/test_unit_audit_tasks.py

@@ -225,6 +225,44 @@ def test_create_audit_log_from_historical_record_creates_audit_log_with_correct_
     )
 
 
+def test_create_audit_log_from_historical_record__cascade_deleted_feature_segment__does_nothing(
+    admin_user: FFAdminUser,
+    feature: Feature,
+    segment: Segment,
+    environment: Environment,
+    mocker: MockerFixture,
+) -> None:
+    """https://github.com/Flagsmith/flagsmith/issues/6792"""
+    # Given
+    feature_segment = FeatureSegment.objects.create(
+        environment=environment,
+        feature=feature,
+        segment=segment,
+    )
+    feature_state = FeatureState.objects.create(
+        environment=environment,
+        feature=feature,
+        feature_segment=feature_segment,
+    )
+    feature_state.enabled = not feature_state.enabled
+    feature_state.save()  # creates a "~" historical record
+    history_instance = feature_state.history.filter(history_type="~").first()
+    assert history_instance is not None
+    feature_segment.delete()  # cascade-deletes the FeatureState
+    get_update_log_message = mocker.spy(FeatureState, "get_update_log_message")
+
+    # When
+    create_audit_log_from_historical_record(
+        history_instance.history_id,
+        admin_user.id,
+        feature_state.history_record_class_path,
+    )
+
+    # Then
+    assert get_update_log_message.call_count == 1
+    assert get_update_log_message.spy_return is None
+
+
 def test_create_segment_priorities_changed_audit_log(
     admin_user: FFAdminUser,
     feature_segment: FeatureSegment,