lib: add precondition checks that count of discarded events/packets is greater than 0

[babeltrace.git] / tests / bindings / python / bt2 / test_message.py

diff --git a/tests/bindings/python/bt2/test_message.py b/tests/bindings/python/bt2/test_message.py
index a99972aadc7d54f226f9de639e34f602a9c58bb2..d9e2fdbc186d5243edbfd7d6545efeae7ef8d962 100644 (file)
--- a/tests/bindings/python/bt2/test_message.py
+++ b/tests/bindings/python/bt2/test_message.py
@@ -418,6 +418,22 @@ class CreateDiscardedEventMessageTestCase(unittest.TestCase):
         self.assertIs(type(msg), bt2._DiscardedEventsMessage)
         self.assertEqual(msg.count, 242)
 
+    # With event count == 0.
+    def test_create_with_count_zero_raises(self):
+        def create_stream_class(tc, cc):
+            return tc.create_stream_class(supports_discarded_events=True)
+
+        def msg_iter_next(msg_iter, stream):
+            with self.assertRaisesRegex(
+                ValueError, 'discarded event count is 0',
+            ):
+                msg_iter._create_discarded_events_message(stream, count=0)
+
+            return 123
+
+        res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
+        self.assertEqual(res, 123)
+
     # With clock snapshots.
     def test_create_with_clock_snapshots(self):
         def create_stream_class(tc, cc):
@@ -547,6 +563,24 @@ class CreateDiscardedPacketMessageTestCase(unittest.TestCase):
         self.assertIs(type(msg), bt2._DiscardedPacketsMessage)
         self.assertEqual(msg.count, 242)
 
+    # With packet count == 0.
+    def test_create_with_count_zero_raises(self):
+        def create_stream_class(tc, cc):
+            return tc.create_stream_class(
+                supports_packets=True, supports_discarded_packets=True
+            )
+
+        def msg_iter_next(msg_iter, stream):
+            with self.assertRaisesRegex(
+                ValueError, 'discarded packet count is 0',
+            ):
+                msg_iter._create_discarded_packets_message(stream, count=0)
+
+            return 123
+
+        res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
+        self.assertEqual(res, 123)
+
     # With clock snapshots.
     def test_create_with_clock_snapshots(self):
         def create_stream_class(tc, cc):