From 7cc2513aa38778654a98b46a1cdb15317cedb11c Mon Sep 17 00:00:00 2001 From: Shubham Padkonde Date: Sun, 27 Sep 2026 16:09:58 +0530 Subject: [PATCH] fix: report event batch serialization failures --- src/amplitude/worker.py | 10 +++++++-- src/test/test_worker.py | 45 ++++++++++++++++++++++++++++++++++++++++- 2 files changed, 52 insertions(+), 3 deletions(-) diff --git a/src/amplitude/worker.py b/src/amplitude/worker.py index ec78825..0fff7bf 100644 --- a/src/amplitude/worker.py +++ b/src/amplitude/worker.py @@ -3,7 +3,7 @@ from threading import Thread, RLock from amplitude.exception import InvalidAPIKeyError -from amplitude.http_client import HttpClient +from amplitude.http_client import HttpClient, HttpStatus from amplitude.processor import ResponseProcessor @@ -77,7 +77,13 @@ def wait_for_all(): def send(self, events): url = self.configuration.server_url - payload = self.get_payload(events) + try: + payload = self.get_payload(events) + except (TypeError, ValueError) as error: + message = "Could not serialize event batch: " + str(error) + self.configuration.logger.error(message) + self.response_processor.callback(events, HttpStatus.INVALID_REQUEST.value, message) + raise res = HttpClient.post(url, payload) try: self.response_processor.process_response(res, events) diff --git a/src/test/test_worker.py b/src/test/test_worker.py index 587e9e3..71f97ae 100644 --- a/src/test/test_worker.py +++ b/src/test/test_worker.py @@ -1,9 +1,10 @@ import time +from datetime import datetime import unittest from collections import defaultdict import random from threading import Thread -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch from amplitude import Config, BaseEvent from amplitude.storage import InMemoryStorage @@ -28,6 +29,48 @@ def tearDown(self) -> None: with self.workers.storage.lock: self.workers.storage.lock.notify() + def test_worker_serialization_errors_notify_callbacks_and_log(self): + for value, error_type in [(datetime(2024, 1, 1), TypeError), (b"bytes", TypeError)]: + with self.subTest(error_type=error_type): + event_callback = MagicMock() + callback = MagicMock() + self.workers.configuration.callback = callback + event = BaseEvent("invalid", "test_user", event_properties={}, callback=event_callback) + # Enrichment can modify properties after constructor validation. + event.event_properties["value"] = value + valid_event = BaseEvent("valid", "test_user") + events = [event, valid_event] + with patch.object(HttpClient, "post") as post: + with self.assertLogs("amplitude", "ERROR") as logs: + with self.assertRaises(error_type) as raised: + self.workers.send(events) + post.assert_not_called() + message = "Could not serialize event batch: " + str(raised.exception) + self.assertIn(message, logs.output[0]) + self.assertEqual(callback.call_count, 2) + callback.assert_any_call(event, 400, message) + callback.assert_any_call(valid_event, 400, message) + event_callback.assert_called_once_with(event, 400, message) + self.assertEqual(event.retry, 0) + + def test_worker_flush_serialization_failure_keeps_future_error(self): + callback = MagicMock() + self.workers.configuration.callback = callback + event = BaseEvent("invalid", "test_user", event_properties={}) + event.event_properties["date"] = datetime(2024, 1, 1) + # Avoid starting the periodic consumer so flush owns this batch. + self.workers.is_started = True + self.workers.storage.push(event) + with patch.object(HttpClient, "post") as post: + with self.assertLogs("amplitude", "ERROR"): + with self.assertRaises(TypeError): + self.workers.flush().result(timeout=5) + post.assert_not_called() + self.assertEqual(callback.call_count, 1) + self.assertEqual(callback.call_args[0][:2], (event, 400)) + self.assertEqual(self.workers.storage.total_events, 0) + self.workers.threads_pool.shutdown() + def test_worker_initialize_setup_success(self): self.assertTrue(self.workers.is_active) self.assertFalse(self.workers.is_started)