fix(celery): respect task filters for exceptions (#807)

Author: marandaneto

.sampo/changesets/cantankerous-stormcaller-vainamoinen.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: patch
+---
+
+Respect Celery task filters for exception capture

posthog/integrations/celery.py

@@ -365,13 +365,18 @@ def _handle_task_end(
             if exception:
                 task_properties["error_type"] = type(exception).__name__
                 task_properties["error_message"] = str(exception)
-                if self.capture_exceptions:
-                    self._capture_exception(exception)
 
             task_name = task_properties.get("celery_task_name")
-            if self.capture_task_lifecycle_events and self._should_track(
-                task_name, task_properties
+            should_track = False
+            if self.capture_task_lifecycle_events or (
+                exception and self.capture_exceptions
             ):
+                should_track = self._should_track(task_name, task_properties)
+
+            if exception and self.capture_exceptions and should_track:
+                self._capture_exception(exception)
+
+            if self.capture_task_lifecycle_events and should_track:
                 self._capture_event(f"celery task {state}", properties=task_properties)
         except Exception:
             logger.exception("Failed to process Celery %s state", state)

posthog/test/integrations/test_celery_integration.py

@@ -509,6 +509,55 @@ def test_task_filter_applies_to_worker_lifecycle_events(self):
 
         mock_client.capture.assert_not_called()
 
+    def test_task_filter_applies_to_failure_exception_and_lifecycle_event(self):
+        mock_client = Mock()
+        task_filter = Mock(return_value=False)
+        integration = PosthogCeleryIntegration(
+            client=mock_client,
+            task_filter=task_filter,
+        )
+        request = SimpleNamespace(headers={}, delivery_info={})
+        task = SimpleNamespace(name="app.tasks.filtered_failure", request=request)
+        exception = ValueError("task failed")
+        context_before = contexts._get_current_context()
+
+        integration._on_task_prerun(sender=task, task_id="task-filtered-failure")
+        task_filter.reset_mock()
+        integration._on_task_failure(
+            sender=task,
+            task_id="task-filtered-failure",
+            exception=exception,
+        )
+
+        task_filter.assert_called_once()
+        mock_client.capture.assert_not_called()
+        mock_client.capture_exception.assert_not_called()
+        self.assertIs(contexts._get_current_context(), context_before)
+
+    def test_task_filter_applies_to_retry_lifecycle_event(self):
+        mock_client = Mock()
+        task_filter = Mock(return_value=False)
+        integration = PosthogCeleryIntegration(
+            client=mock_client,
+            task_filter=task_filter,
+        )
+        request = SimpleNamespace(headers={}, delivery_info={})
+        task = SimpleNamespace(name="app.tasks.filtered_retry", request=request)
+        context_before = contexts._get_current_context()
+
+        integration._on_task_prerun(sender=task, task_id="task-filtered-retry")
+        task_filter.reset_mock()
+        integration._on_task_retry(
+            sender=task,
+            task_id="task-filtered-retry",
+            reason=ConnectionError("broker down"),
+        )
+
+        task_filter.assert_called_once()
+        mock_client.capture.assert_not_called()
+        mock_client.capture_exception.assert_not_called()
+        self.assertIs(contexts._get_current_context(), context_before)
+
     def test_task_failure_captures_exception_when_lifecycle_events_disabled(self):
         mock_client = Mock()
         integration = PosthogCeleryIntegration(