[EventHubs] exception/kwargs testing (#26867)

* fix flushing list batches in bp

* add send list bp test

Author: swathipil

sdk/eventhub/azure-eventhub/azure/eventhub/_buffered_producer/_buffered_producer.py

@@ -165,15 +165,21 @@ def flush(self, timeout_time=None, raise_error=True):
                             self.partition_id,
                             len(batch),
                         )
-                        self._on_success(batch._internal_events, self.partition_id)
+                        try:
+                            self._on_success(batch._internal_events, self.partition_id)
+                        except AttributeError:
+                            self._on_success(batch, self.partition_id)
                     except Exception as exc:  # pylint: disable=broad-except
                         _LOGGER.info(
                             "Partition %r sending %r events failed due to exception: %r ",
                             self.partition_id,
                             len(batch),
                             exc,
                         )
-                        self._on_error(batch._internal_events, self.partition_id, exc)
+                        try:
+                            self._on_error(batch._internal_events, self.partition_id, exc)
+                        except AttributeError:
+                            self._on_error(batch, self.partition_id, exc)
                     finally:
                         self._cur_buffered_len -= len(batch)
                 else:

sdk/eventhub/azure-eventhub/azure/eventhub/aio/_buffered_producer/_buffered_producer_async.py

@@ -167,15 +167,21 @@ async def _flush(self, timeout_time=None, raise_error=True):
                         self.partition_id,
                         len(batch),
                     )
-                    await self._on_success(batch._internal_events, self.partition_id)
+                    try:
+                        await self._on_success(batch._internal_events, self.partition_id)
+                    except AttributeError:
+                        await self._on_success(batch, self.partition_id)
                 except Exception as exc:  # pylint: disable=broad-except
                     _LOGGER.info(
                         "Partition %r sending %r events failed due to exception: %r",
                         self.partition_id,
                         len(batch),
                         exc,
                     )
-                    await self._on_error(batch._internal_events, self.partition_id, exc)
+                    try:
+                        await self._on_error(batch._internal_events, self.partition_id, exc)
+                    except AttributeError:
+                        await self._on_error(batch, self.partition_id, exc)
                 finally:
                     self._cur_buffered_len -= len(batch)
             # If flush could not get the semaphore, we log and raise error if wanted

sdk/eventhub/azure-eventhub/tests/livetest/asynctests/test_buffered_producer_async.py

@@ -407,7 +407,7 @@ async def on_error(events, pid, err):
 
     async with producer:
         partitions = await producer.get_partition_ids()
-        await producer.send_event(EventData('data'))
+        await producer.send_batch([EventData('data')])
         await asyncio.sleep(5)
         assert not sent_events
         await asyncio.sleep(20)

sdk/eventhub/azure-eventhub/tests/livetest/synctests/test_buffered_producer.py

@@ -419,7 +419,7 @@ def on_error(events, pid, err):
 
     with producer:
         partitions = producer.get_partition_ids()
-        producer.send_event(EventData('data'))
+        producer.send_batch([EventData('data')])
         time.sleep(5)
         assert not sent_events
         time.sleep(10)