@pre
<code>bt_stream_class_discarded_events_have_default_clock_snapshots(bt_stream_borrow_class_const(stream))</code>
returns #BT_TRUE.
+ \bt_p{beginning_clock_snapshot_value} ⩽ \bt_p{end_clock_snapshot_value}
@bt_post_success_frozen{stream}
*/
@pre
<code>bt_stream_class_discarded_packets_have_default_clock_snapshots(bt_stream_borrow_class_const(stream))</code>
returns #BT_TRUE.
+ \bt_p{beginning_clock_snapshot_value} ⩽ \bt_p{end_clock_snapshot_value}
@bt_post_success_frozen{stream}
*/
utils._check_uint64(beg_clock_snapshot)
utils._check_uint64(end_clock_snapshot)
+
+ if beg_clock_snapshot > end_clock_snapshot:
+ raise ValueError(
+ 'beginning default clock snapshot value ({}) is greater than end default clock snapshot value ({})'.format(
+ beg_clock_snapshot, end_clock_snapshot
+ )
+ )
+
ptr = native_bt.message_discarded_events_create_with_default_clock_snapshots(
self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
)
utils._check_uint64(beg_clock_snapshot)
utils._check_uint64(end_clock_snapshot)
+
+ if beg_clock_snapshot > end_clock_snapshot:
+ raise ValueError(
+ 'beginning default clock snapshot value ({}) is greater than end default clock snapshot value ({})'.format(
+ beg_clock_snapshot, end_clock_snapshot
+ )
+ )
+
ptr = native_bt.message_discarded_packets_create_with_default_clock_snapshots(
self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
)
uint64_t end_raw_value)
{
BT_ASSERT_PRE_DEV_NO_ERROR();
+ BT_ASSERT_PRE_BEGIN_LE_END(message_iterator, beginning_raw_value, end_raw_value);
return create_discarded_items_message(message_iterator,
BT_MESSAGE_TYPE_DISCARDED_EVENTS, (void *) stream,
uint64_t end_raw_value)
{
BT_ASSERT_PRE_DEV_NO_ERROR();
+ BT_ASSERT_PRE_BEGIN_LE_END(message_iterator, beginning_raw_value, end_raw_value);
return create_discarded_items_message(message_iterator,
BT_MESSAGE_TYPE_DISCARDED_PACKETS, (void *) stream,
_BT_ASSERT_PRE_MSG_IS_TYPE_FMT, \
bt_message_type_string(_type), (_msg))
+#define BT_ASSERT_PRE_BEGIN_LE_END(_msg_iter, _begin, _end) \
+ BT_ASSERT_PRE( \
+ _begin <= _end, \
+ "Beginning default clock snapshot value is greater " \
+ "than end default clock snapshot value: " \
+ "cs-begin-val=%" PRIu64 ", cs-end-val=%" PRIu64 ", " \
+ "%![msg-iter-]i", \
+ _begin, _end, _msg_iter);
+
BT_HIDDEN
void bt_message_init(struct bt_message *message,
enum bt_message_type type,
res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
self.assertEqual(res, 123)
+ # End clock snapshot greater than beginning clock snapshot.
+ def test_create_clock_snapshots_end_gt_begin_raises(self):
+ def create_stream_class(tc, cc):
+ return tc.create_stream_class(
+ default_clock_class=cc,
+ supports_discarded_events=True,
+ discarded_events_have_default_clock_snapshots=True,
+ )
+
+ def msg_iter_next(msg_iter, stream):
+ with self.assertRaisesRegex(
+ ValueError,
+ r'beginning default clock snapshot value \(20\) is greater than end default clock snapshot value \(10\)',
+ ):
+ msg_iter._create_discarded_events_message(
+ stream, beg_clock_snapshot=20, end_clock_snapshot=10
+ )
+
+ return 123
+
+ res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
+ self.assertEqual(res, 123)
+
class CreateDiscardedPacketMessageTestCase(unittest.TestCase):
# Most basic case.
res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
self.assertEqual(res, 123)
+ # End clock snapshot greater than beginning clock snapshot.
+ def test_create_clock_snapshots_end_gt_begin_raises(self):
+ def create_stream_class(tc, cc):
+ return tc.create_stream_class(
+ default_clock_class=cc,
+ supports_packets=True,
+ supports_discarded_packets=True,
+ discarded_packets_have_default_clock_snapshots=True,
+ )
+
+ def msg_iter_next(msg_iter, stream):
+ with self.assertRaisesRegex(
+ ValueError,
+ r'beginning default clock snapshot value \(20\) is greater than end default clock snapshot value \(10\)',
+ ):
+ msg_iter._create_discarded_packets_message(
+ stream, beg_clock_snapshot=20, end_clock_snapshot=10
+ )
+
+ return 123
+
+ res = utils.run_in_message_iterator_next(create_stream_class, msg_iter_next)
+ self.assertEqual(res, 123)
+
if __name__ == '__main__':
unittest.main()