+ def test_msg_stream_with_clock_snapshots(self):
+ params = {'with_cc': True, 'with_stream_msgs_clock_snapshots': True}
+
+ self._src_comp = self._graph.add_component(self._src, 'my_source', params)
+ self._msg_iter = TestOutputPortMessageIterator(
+ self._graph, self._src_comp.output_ports['out']
+ )
+ msgs = list(self._msg_iter)
+
+ msg_stream_beg = msgs[0]
+ self.assertIsInstance(msg_stream_beg, bt2._StreamBeginningMessageConst)
+ self.assertIs(
+ type(msg_stream_beg.default_clock_snapshot),
+ bt2_clock_snapshot._ClockSnapshotConst,
+ )
+ self.assertEqual(msg_stream_beg.default_clock_snapshot.value, 0)
+
+ msg_stream_end = msgs[7]
+ self.assertIsInstance(msg_stream_end, bt2._StreamEndMessageConst)
+ self.assertIs(
+ type(msg_stream_end.default_clock_snapshot),
+ bt2_clock_snapshot._ClockSnapshotConst,
+ )
+ self.assertEqual(msg_stream_end.default_clock_snapshot.value, 7)
+
+ def test_stream_beg_msg(self):
+ msg = utils.get_stream_beginning_message()
+ self.assertIs(type(msg.stream), bt2_stream._Stream)
+
+ def test_stream_end_msg(self):
+ msg = utils.get_stream_end_message()
+ self.assertIs(type(msg.stream), bt2_stream._Stream)
+
+ def test_packet_beg_msg(self):
+ msg = utils.get_packet_beginning_message()
+ self.assertIs(type(msg.packet), bt2_packet._Packet)
+
+ def test_packet_end_msg(self):
+ msg = utils.get_packet_end_message()
+ self.assertIs(type(msg.packet), bt2_packet._Packet)
+
+ def test_event_msg(self):
+ msg = utils.get_event_message()
+ self.assertIs(type(msg.event), bt2_event._Event)
+
+
+if __name__ == '__main__':
+ unittest.main()